protocol.py
Go to the documentation of this file.
1 # Software License Agreement (BSD License)
2 #
3 # Copyright (c) 2012, Willow Garage, Inc.
4 # All rights reserved.
5 #
6 # Redistribution and use in source and binary forms, with or without
7 # modification, are permitted provided that the following conditions
8 # are met:
9 #
10 # * Redistributions of source code must retain the above copyright
11 # notice, this list of conditions and the following disclaimer.
12 # * Redistributions in binary form must reproduce the above
13 # copyright notice, this list of conditions and the following
14 # disclaimer in the documentation and/or other materials provided
15 # with the distribution.
16 # * Neither the name of Willow Garage, Inc. nor the names of its
17 # contributors may be used to endorse or promote products derived
18 # from this software without specific prior written permission.
19 #
20 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
23 # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
24 # COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
25 # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
26 # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
27 # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
28 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
29 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
30 # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
31 # POSSIBILITY OF SUCH DAMAGE.
32 
33 import rospy
34 import time
35 
36 from rosbridge_library.internal.exceptions import InvalidArgumentException
37 from rosbridge_library.internal.exceptions import MissingArgumentException
38 
39 from rosbridge_library.capabilities.fragmentation import Fragmentation
40 from rosbridge_library.util import json, bson
41 
42 
43 def is_number(s):
44  try:
45  float(s)
46  return True
47  except ValueError:
48  return False
49 
50 
51 def has_binary(obj):
52  """ Returns True if obj is a binary or contains a binary attribute
53  """
54 
55  if isinstance(obj, list):
56  return any(has_binary(item) for item in obj)
57 
58  if isinstance(obj, dict):
59  return any(has_binary(obj[item]) for item in obj)
60 
61  return isinstance(obj, bson.binary.Binary)
62 
63 
64 class Protocol:
65  """ The interface for a single client to interact with ROS.
66 
67  See rosbridge_protocol for the default protocol used by rosbridge
68 
69  The lifecycle for a Protocol instance is as follows:
70  - Pass incoming messages from the client to incoming
71  - Propagate outgoing messages to the client by overriding outgoing
72  - Call finish to clean up resources when the client is finished
73 
74  """
75 
76  # fragment_size can be set per client (each client has its own instance of protocol)
77  # ..same for other parameters
78  fragment_size = None
79  png = None
80  # buffer used to gather partial JSON-objects (could be caused by small tcp-buffers or similar..)
81  buffer = ""
82  old_buffer = ""
83  busy = False
84  # if this is too low, ("simple")clients network stacks will get flooded (when sending fragments of a huge message..)
85  # .. depends on message_size/bandwidth/performance/client_limits/...
86  # !! this might be related to (or even be avoided by using) throttle_rate !!
87  delay_between_messages = 0
88  # global list of non-ros advertised services
89  external_service_list = {}
90  # Use only BSON for the whole communication if the server has been started with bson_only_mode:=True
91  bson_only_mode = False
92 
93  parameters = None
94 
95  def __init__(self, client_id):
96  """ Keyword arguments:
97  client_id -- a unique ID for this client to take. Uniqueness is
98  important otherwise there will be conflicts between multiple clients
99  with shared resources
100 
101  """
102  self.client_id = client_id
103  self.capabilities = []
104  self.operations = {}
105 
106  if self.parameters:
107  self.fragment_size = self.parameters["max_message_size"]
108  self.delay_between_messages = self.parameters["delay_between_messages"]
109  self.bson_only_mode = self.parameters.get('bson_only_mode', False)
110 
111  # added default message_string="" to allow recalling incoming until buffer is empty without giving a parameter
112  # --> allows to get rid of (..or minimize) delay between client-side sends
113  def incoming(self, message_string=""):
114  """ Process an incoming message from the client
115 
116  Keyword arguments:
117  message_string -- the wire-level message sent by the client
118 
119  """
120  self.buffer = self.buffer + message_string
121  msg = None
122 
123  # take care of having multiple JSON-objects in receiving buffer
124  # ..first, try to load the whole buffer as a JSON-object
125  try:
126  msg = self.deserialize(self.buffer)
127  self.buffer = ""
128 
129  # if loading whole object fails try to load part of it (from first opening bracket "{" to next closing bracket "}"
130  # .. this causes Exceptions on "inner" closing brackets --> so I suppressed logging of deserialization errors
131  except Exception as e:
132  if self.bson_only_mode:
133  # Since BSON should be used in conjunction with a network handler
134  # that receives exactly one full BSON message.
135  # This will then be passed to self.deserialize and shouldn't cause any
136  # exceptions because of fragmented messages (broken or invalid messages might still be sent tough)
137  self.log("error", "Exception in deserialization of BSON")
138 
139  else:
140  # TODO: handling of partial/multiple/broken json data in incoming buffer
141  # this way is problematic when json contains nested json-objects ( e.g. { ... { "config": [0,1,2,3] } ... } )
142  # .. if outer json is not fully received, stepping through opening brackets will find { "config" : ... } as a valid json object
143  # .. and pass this "inner" object to rosbridge and throw away the leading part of the "outer" object..
144  # solution for now:
145  # .. check for "op"-field. i can still imagine cases where a nested message ( e.g. complete service_response fits into the data field of a fragment..)
146  # .. would cause trouble, but if a response fits as a whole into a fragment, simply do not pack it into a fragment.
147  #
148  # --> from that follows current limitiation:
149  # fragment data must NOT (!) contain a complete json-object that has an "op-field"
150  #
151  # an alternative solution would be to only check from first opening bracket and have a time out on data in input buffer.. (to handle broken data)
152  opening_brackets = [i for i, letter in enumerate(self.buffer) if letter == '{']
153  closing_brackets = [i for i, letter in enumerate(self.buffer) if letter == '}']
154 
155  for start in opening_brackets:
156  for end in closing_brackets:
157  try:
158  msg = self.deserialize(self.buffer[start:end+1])
159  if msg.get("op",None) != None:
160  # TODO: check if throwing away leading data like this is okay.. loops look okay..
161  self.buffer = self.buffer[end+1:len(self.buffer)]
162  # jump out of inner loop if json-decode succeeded
163  break
164  except Exception as e:
165  # debug json-decode errors with this line
166  #print e
167  pass
168  # if load was successfull --> break outer loop, too.. -> no need to check if json begins at a "later" opening bracket..
169  if msg != None:
170  break
171 
172  # if decoding of buffer failed .. simply return
173  if msg is None:
174  return
175 
176  # process fields JSON-message object that "control" rosbridge
177  mid = None
178  if "id" in msg:
179  mid = msg["id"]
180  if "op" not in msg:
181  if "receiver" in msg:
182  self.log("error", "Received a rosbridge v1.0 message. Please refer to rosbridge.org for the correct format of rosbridge v2.0 messages. Original message was: %s" % message_string)
183  else:
184  self.log("error", "Received a message without an op. All messages require 'op' field with value one of: %s. Original message was: %s" % (list(self.operations.keys()), message_string), mid)
185  return
186  op = msg["op"]
187  if op not in self.operations:
188  self.log("error", "Unknown operation: %s. Allowed operations: %s" % (op, list(self.operations.keys())), mid)
189  return
190  # this way a client can change/overwrite it's active values anytime by just including parameter field in any message sent to rosbridge
191  # maybe need to be improved to bind parameter values to specific operation..
192  if "fragment_size" in msg.keys():
193  self.fragment_size = msg["fragment_size"]
194  #print "fragment size set to:", self.fragment_size
195  if "message_intervall" in msg.keys() and is_number(msg["message_intervall"]):
196  self.delay_between_messages = msg["message_intervall"]
197  if "png" in msg.keys():
198  self.png = msg["msg"]
199 
200  # now try to pass message to according operation
201  try:
202  self.operations[op](msg)
203  except Exception as exc:
204  self.log("error", "%s: %s" % (op, str(exc)), mid)
205 
206  # if anything left in buffer .. re-call self.incoming
207  # TODO: check what happens if we have "garbage" on tcp-stack --> infinite loop might be triggered! .. might get out of it when next valid JSON arrives since only data after last 'valid' closing bracket is kept
208  if len(self.buffer) > 0:
209  # try to avoid infinite loop..
210  if self.old_buffer != self.buffer:
211  self.old_buffer = self.buffer
212  self.incoming()
213 
214 
215 
216  def outgoing(self, message):
217  """ Pass an outgoing message to the client. This method should be
218  overridden.
219 
220  Keyword arguments:
221  message -- the wire-level message to send to the client
222 
223  """
224  pass
225 
226  def send(self, message, cid=None):
227  """ Called internally in preparation for sending messages to the client
228 
229  This method pre-processes the message then passes it to the overridden
230  outgoing method.
231 
232  Keyword arguments:
233  message -- a dict of message values to be marshalled and sent
234  cid -- (optional) an associated id
235 
236  """
237  serialized = self.serialize(message, cid)
238  if serialized is not None:
239  if self.png == "png":
240  # TODO: png compression on outgoing messages
241  # encode message
242  pass
243 
244  fragment_list = None
245  if self.fragment_size != None and len(serialized) > self.fragment_size:
246  mid = message.get("id", None)
247 
248  # TODO: think about splitting into fragments that have specified size including header-fields!
249  # --> estimate header size --> split content into fragments that have the requested overall size, rather than requested content size
250  fragment_list = Fragmentation(self).fragment(message, self.fragment_size, mid )
251 
252  # fragment list not empty -> send fragments
253  if fragment_list != None:
254  for fragment in fragment_list:
255  if self.bson_only_mode:
256  self.outgoing(bson.BSON.encode(fragment))
257  else:
258  self.outgoing(json.dumps(fragment))
259  # okay to use delay here (sender's send()-function) because rosbridge is sending next request only to service provider when last one had finished)
260  # --> if this was not the case this delay needed to be implemented in service-provider's (meaning message receiver's) send_message()-function in rosbridge_tcp.py)
261  time.sleep(self.delay_between_messages)
262  # else send message as it is
263  else:
264  self.outgoing(serialized)
265  time.sleep(self.delay_between_messages)
266 
267  def finish(self):
268  """ Indicate that the client is finished and clean up resources.
269 
270  All clients should call this method after disconnecting.
271 
272  """
273  for capability in self.capabilities:
274  capability.finish()
275 
276  def serialize(self, msg, cid=None):
277  """ Turns a dictionary of values into the appropriate wire-level
278  representation.
279 
280  Default behaviour uses JSON. Override to use a different container.
281 
282  Keyword arguments:
283  msg -- the dictionary of values to serialize
284  cid -- (optional) an ID associated with this. Will be logged on err.
285 
286  Returns a JSON string representing the dictionary
287  """
288  try:
289  if type(msg) == bytearray:
290  return msg
291  if has_binary(msg) or self.bson_only_mode:
292  return bson.BSON.encode(msg)
293  else:
294  return json.dumps(msg)
295  except:
296  if cid is not None:
297  # Only bother sending the log message if there's an id
298  self.log("error", "Unable to serialize %s message to client"
299  % msg["op"], cid)
300  return None
301 
302  def deserialize(self, msg, cid=None):
303 
304  """ Turns the wire-level representation into a dictionary of values
305 
306  Default behaviour assumes JSON. Override to use a different container.
307 
308  Keyword arguments:
309  msg -- the wire-level message to deserialize
310  cid -- (optional) an ID associated with this. Is logged on error
311 
312  Returns a dictionary of values
313 
314  """
315  try:
316  if self.bson_only_mode:
317  bson_message = bson.BSON(msg)
318  return bson_message.decode()
319  else:
320  return json.loads(msg)
321  except Exception as e:
322  # if we did try to deserialize whole buffer .. first try to let self.incoming check for multiple/partial json-decodes before logging error
323  # .. this means, if buffer is not == msg --> we tried to decode part of buffer
324 
325  # TODO: implement a way to have a final Exception when nothing works out to decode (multiple/broken/partial JSON..)
326 
327  # supressed logging of exception on json-decode to keep rosbridge-logs "clean", otherwise console logs would get spammed for every failed json-decode try
328 # if msg != self.buffer:
329 # error_msg = "Unable to deserialize message from client: %s" % msg
330 # error_msg += "\nException was: " +str(e)
331 #
332 # self.log("error", error_msg, cid)
333 
334  # re-raise Exception to allow handling outside of deserialize function instead of returning None
335  raise
336  #return None
337 
338  def register_operation(self, opcode, handler):
339  """ Register a handler for an opcode
340 
341  Keyword arguments:
342  opcode -- the opcode to register this handler for
343  handler -- a callback function to call for messages with this opcode
344 
345  """
346  self.operations[opcode] = handler
347 
348  def unregister_operation(self, opcode):
349  """ Unregister a handler for an opcode
350 
351  Keyword arguments:
352  opcode -- the opcode to unregister the handler for
353 
354  """
355  if opcode in self.operations:
356  del self.operations[opcode]
357 
358  def add_capability(self, capability_class):
359  """ Add a capability to the protocol.
360 
361  This method is for convenience; assumes the default capability
362  constructor
363 
364  Keyword arguments:
365  capability_class -- the class of the capability to add
366 
367  """
368  self.capabilities.append(capability_class(self))
369 
370  def log(self, level, message, lid=None):
371  """ Log a message to the client. By default just sends to stdout
372 
373  Keyword arguments:
374  level -- the logger level of this message
375  message -- the string message to send to the user
376  lid -- an associated for this log message
377 
378  """
379  stdout_formatted_msg = None
380  if lid is not None:
381  stdout_formatted_msg = "[Client %s] [id: %s] %s" % (self.client_id, lid, message)
382  else:
383  stdout_formatted_msg = "[Client %s] %s" % (self.client_id, message)
384 
385  if level == "error" or level == "err":
386  rospy.logerr(stdout_formatted_msg)
387  elif level == "warning" or level == "warn":
388  rospy.logwarn(stdout_formatted_msg)
389  elif level == "info" or level == "information":
390  rospy.loginfo(stdout_formatted_msg)
391  else:
392  rospy.logdebug(stdout_formatted_msg)
def register_operation(self, opcode, handler)
Definition: protocol.py:338
def unregister_operation(self, opcode)
Definition: protocol.py:348
def __init__(self, client_id)
Definition: protocol.py:95
def add_capability(self, capability_class)
Definition: protocol.py:358
def deserialize(self, msg, cid=None)
Definition: protocol.py:302
def log(self, level, message, lid=None)
Definition: protocol.py:370
def send(self, message, cid=None)
Definition: protocol.py:226
def incoming(self, message_string="")
Definition: protocol.py:113
def serialize(self, msg, cid=None)
Definition: protocol.py:276


rosbridge_library
Author(s): Jonathan Mace
autogenerated on Wed Jun 3 2020 03:55:14