1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
|
#!/router/bin/python
import zmq
import json
import re
from collections import namedtuple
import zlib
import struct
from .trex_stl_types import *
from .utils.common import random_id_gen
class bcolors:
BLUE = '\033[94m'
GREEN = '\033[32m'
YELLOW = '\033[93m'
RED = '\033[31m'
MAGENTA = '\033[35m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
# sub class to describe a batch
class BatchMessage(object):
def __init__ (self, rpc_client):
self.rpc_client = rpc_client
self.batch_list = []
def add (self, method_name, params = None, api_class = 'core'):
id, msg = self.rpc_client.create_jsonrpc_v2(method_name, params, api_class, encode = False)
self.batch_list.append(msg)
def invoke(self, block = False):
if not self.rpc_client.connected:
return RC_ERR("Not connected to server")
msg = json.dumps(self.batch_list)
return self.rpc_client.send_msg(msg)
# JSON RPC v2.0 client
class JsonRpcClient(object):
MSG_COMPRESS_THRESHOLD = 4096
MSG_COMPRESS_HEADER_MAGIC = 0xABE85CEA
def __init__ (self, default_server, default_port, client):
self.client_api = client.api_h
self.logger = client.logger
self.connected = False
# default values
self.port = default_port
self.server = default_server
self.id_gen = random_id_gen()
def get_connection_details (self):
rc = {}
rc['server'] = self.server
rc['port'] = self.port
return rc
# pretty print for JSON
def pretty_json (self, json_str, use_colors = True):
pretty_str = json.dumps(json.loads(json_str), indent = 4, separators=(',', ': '), sort_keys = True)
if not use_colors:
return pretty_str
try:
# int numbers
pretty_str = re.sub(r'([ ]*:[ ]+)(\-?[1-9][0-9]*[^.])',r'\1{0}\2{1}'.format(bcolors.BLUE, bcolors.ENDC), pretty_str)
# float
pretty_str = re.sub(r'([ ]*:[ ]+)(\-?[1-9][0-9]*\.[0-9]+)',r'\1{0}\2{1}'.format(bcolors.MAGENTA, bcolors.ENDC), pretty_str)
# strings
pretty_str = re.sub(r'([ ]*:[ ]+)("[^"]*")',r'\1{0}\2{1}'.format(bcolors.RED, bcolors.ENDC), pretty_str)
pretty_str = re.sub(r"('[^']*')", r'{0}\1{1}'.format(bcolors.MAGENTA, bcolors.RED), pretty_str)
except :
pass
return pretty_str
def verbose_msg (self, msg):
self.logger.log("\n\n[verbose] " + msg, level = self.logger.VERBOSE_HIGH)
# batch messages
def create_batch (self):
return BatchMessage(self)
def create_jsonrpc_v2 (self, method_name, params = None, api_class = 'core', encode = True):
msg = {}
msg["jsonrpc"] = "2.0"
msg["method"] = method_name
msg["id"] = next(self.id_gen)
msg["params"] = params if params is not None else {}
# if this RPC has an API class - add it's handler
if api_class:
msg["params"]["api_h"] = self.client_api[api_class]
if encode:
return id, json.dumps(msg)
else:
return id, msg
def invoke_rpc_method (self, method_name, params = None, api_class = 'core'):
if not self.connected:
return RC_ERR("Not connected to server")
id, msg = self.create_jsonrpc_v2(method_name, params, api_class)
return self.send_msg(msg)
def compress_msg (self, msg):
# compress
compressed = zlib.compress(msg)
new_msg = struct.pack(">II", self.MSG_COMPRESS_HEADER_MAGIC, len(msg)) + compressed
return new_msg
def decompress_msg (self, msg):
if len(msg) < 8:
return None
t = struct.unpack(">II", msg[:8])
if (t[0] != self.MSG_COMPRESS_HEADER_MAGIC):
return None
x = zlib.decompress(msg[8:])
if len(x) != t[1]:
return None
return x
def send_msg (self, msg):
# print before
if self.logger.check_verbose(self.logger.VERBOSE_HIGH):
self.verbose_msg("Sending Request To Server:\n\n" + self.pretty_json(msg) + "\n")
# encode string to buffer
buffer = msg.encode()
if len(buffer) > self.MSG_COMPRESS_THRESHOLD:
response = self.send_raw_msg(self.compress_msg(buffer))
if response:
response = self.decompress_msg(response)
else:
response = self.send_raw_msg(buffer)
if not response:
return response
# return to string
response = response.decode()
# print after
if self.logger.check_verbose(self.logger.VERBOSE_HIGH):
self.verbose_msg("Server Response:\n\n" + self.pretty_json(response) + "\n")
# process response (batch and regular)
try:
response_json = json.loads(response)
except (TypeError, ValueError):
return RC_ERR("*** [RPC] - Failed to decode response from server")
if isinstance(response_json, list):
return self.process_batch_response(response_json)
else:
return self.process_single_response(response_json)
# low level send of string message
def send_raw_msg (self, msg):
tries = 0
while True:
try:
self.socket.send(msg)
break
except zmq.Again:
tries += 1
if tries > 5:
self.disconnect()
return RC_ERR("*** [RPC] - Failed to send message to server")
tries = 0
while True:
try:
response = self.socket.recv()
break
except zmq.Again:
tries += 1
if tries > 5:
self.disconnect()
return RC_ERR("*** [RPC] - Failed to get server response at {0}".format(self.transport))
return response
# processs a single response from server
def process_single_response (self, response_json):
if (response_json.get("jsonrpc") != "2.0"):
return RC_ERR("Malformed Response ({0})".format(str(response_json)))
# error reported by server
if ("error" in response_json):
if "specific_err" in response_json["error"]:
return RC_ERR(response_json["error"]["specific_err"])
else:
return RC_ERR(response_json["error"]["message"])
# if no error there should be a result
if ("result" not in response_json):
return RC_ERR("Malformed Response ({0})".format(str(response_json)))
return RC_OK(response_json["result"])
# process a batch response
def process_batch_response (self, response_json):
rc_batch = RC()
for single_response in response_json:
rc = self.process_single_response(single_response)
rc_batch.add(rc)
return rc_batch
def disconnect (self):
if self.connected:
self.socket.close(linger = 0)
self.context.destroy(linger = 0)
self.connected = False
return RC_OK()
else:
return RC_ERR("Not connected to server")
def connect(self, server = None, port = None):
if self.connected:
self.disconnect()
self.context = zmq.Context()
self.server = (server if server else self.server)
self.port = (port if port else self.port)
# Socket to talk to server
self.transport = "tcp://{0}:{1}".format(self.server, self.port)
self.socket = self.context.socket(zmq.REQ)
try:
self.socket.connect(self.transport)
except zmq.error.ZMQError as e:
return RC_ERR("ZMQ Error: Bad server or port name: " + str(e))
self.socket.setsockopt(zmq.SNDTIMEO, 10000)
self.socket.setsockopt(zmq.RCVTIMEO, 10000)
self.connected = True
rc = self.invoke_rpc_method('ping', api_class = None)
if not rc:
self.connected = False
return rc
return RC_OK()
def reconnect(self):
# connect using current values
return self.connect()
if not self.connected:
return RC_ERR("Not connected to server")
# reconnect
return self.connect(self.server, self.port)
def is_connected(self):
return self.connected
def __del__(self):
self.logger.log("Shutting down RPC client\n")
if hasattr(self, "context"):
self.context.destroy(linger=0)
|