import weakref
import atexit
from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType, BaseTypes
-from . vpp_serializer import VPPMessage, vpp_get_type
-from . vpp_format import VPPFormat
+from . vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias
+from . macaddress import MACAddress, mac_pton, mac_ntop
+
+logger = logging.getLogger(__name__)
if sys.version[0] == '2':
import Queue as queue
vpp_instance.disconnect()
-def vpp_iterator(d):
- if sys.version[0] == '2':
+if sys.version[0] == '2':
+ def vpp_iterator(d):
return d.iteritems()
- else:
+else:
+ def vpp_iterator(d):
return d.items()
+def call_logger(msgdef, kwargs):
+ s = 'Calling {}('.format(msgdef.name)
+ for k, v in kwargs.items():
+ s += '{}:{} '.format(k, v)
+ s += ')'
+ return s
+
+
+def return_logger(r):
+ s = 'Return from {}'.format(r)
+ return s
+
+
class VppApiDynamicMethodHolder(object):
pass
return self._func(**kwargs)
+class VPPApiError(Exception):
+ pass
+
+
+class VPPNotImplementedError(NotImplementedError):
+ pass
+
+
+class VPPIOError(IOError):
+ pass
+
+
+class VPPRuntimeError(RuntimeError):
+ pass
+
+
+class VPPValueError(ValueError):
+ pass
+
+
class VPP(object):
"""VPP interface.
provides a means to register a callback function to receive
these messages in a background thread.
"""
+ VPPApiError = VPPApiError
+ VPPRuntimeError = VPPRuntimeError
+ VPPValueError = VPPValueError
+ VPPNotImplementedError = VPPNotImplementedError
+ VPPIOError = VPPIOError
def process_json_file(self, apidef_file):
api = json.load(apidef_file)
for t in api['types']:
t[0] = 'vl_api_' + t[0] + '_t'
types[t[0]] = {'type': 'type', 'data': t}
+ for t, v in api['aliases'].items():
+ types['vl_api_' + t + '_t'] = {'type': 'alias', 'data': v}
+ self.services.update(api['services'])
i = 0
while True:
unresolved = {}
for k, v in types.items():
t = v['data']
- if not vpp_get_type(t[0]):
+ if not vpp_get_type(k):
if v['type'] == 'enum':
try:
VPPEnumType(t[0], t[1:])
VPPType(t[0], t[1:])
except ValueError:
unresolved[k] = v
+ elif v['type'] == 'alias':
+ try:
+ VPPTypeAlias(k, t)
+ except ValueError:
+ unresolved[k] = v
if len(unresolved) == 0:
break
if i > 3:
- raise ValueError('Unresolved type definitions {}'
- .format(unresolved))
+ raise VPPValueError('Unresolved type definitions {}'
+ .format(unresolved))
types = unresolved
i += 1
for m in api['messages']:
try:
self.messages[m[0]] = VPPMessage(m[0], m[1:])
- except NotImplementedError:
+ except VPPNotImplementedError:
self.logger.error('Not implemented error for {}'.format(m[0]))
def __init__(self, apifiles=None, testmode=False, async_thread=True,
self.logger = logger
self.messages = {}
+ self.services = {}
self.id_names = []
self.id_msgdef = []
self.header = VPPType('header', [['u16', 'msgid'],
if testmode:
apifiles = []
else:
- raise
+ raise VPPRuntimeError
for file in apifiles:
with open(file) as apidef_file:
# Basic sanity check
if len(self.messages) == 0 and not testmode:
- raise ValueError(1, 'Missing JSON message definitions')
+ raise VPPValueError(1, 'Missing JSON message definitions')
self.transport = VppTransport(self, read_timeout=read_timeout,
server_address=server_address)
if api_dir is None:
api_dir = cls.find_api_dir()
if api_dir is None:
- raise RuntimeError("api_dir cannot be located")
+ raise VPPApiError("api_dir cannot be located")
if isinstance(patterns, list) or isinstance(patterns, tuple):
patterns = [p.strip() + '.api.json' for p in patterns]
@property
def api(self):
if not hasattr(self, "_api"):
- raise Exception("Not connected, api definitions not available")
+ raise VPPApiError("Not connected, api definitions not available")
return self._api
def make_function(self, msg, i, multipart, do_async):
if i > 0:
self.id_msgdef[i] = msg
self.id_names[i] = name
- # TODO: Fix multipart (use services)
- multipart = True if name.find('_dump') > 0 else False
- f = self.make_function(msg, i, multipart, do_async)
- setattr(self._api, name, FuncWrapper(f))
+
+ # Create function for client side messages.
+ if name in self.services:
+ if 'stream' in self.services[name] and \
+ self.services[name]['stream']:
+ multipart = True
+ else:
+ multipart = False
+ f = self.make_function(msg, i, multipart, do_async)
+ setattr(self._api, name, FuncWrapper(f))
else:
self.logger.debug(
'No such message type or failed CRC checksum: %s', n)
rv = self.transport.connect(name.encode(), pfx, msg_handler, rx_qlen)
if rv != 0:
- raise IOError(2, 'Connect failed')
+ raise VPPIOError(2, 'Connect failed')
self.vpp_dictionary_maxid = self.transport.msg_table_max_index()
self._register_functions(do_async=do_async)
# No context -> async notification that we feed to the callback
self.message_queue.put_nowait(r)
else:
- raise IOError(2, 'RPC reply message received in event handler')
+ raise VPPIOError(2, 'RPC reply message received in event handler')
+
+ def has_context(self, msg):
+ if len(msg) < 10:
+ return False
+
+ header = VPPType('header_with_context', [['u16', 'msgid'],
+ ['u32', 'client_index'],
+ ['u32', 'context']])
+
+ (i, ci, context), size = header.unpack(msg, 0)
+ if self.id_names[i] == 'rx_thread_exit':
+ return
- def decode_incoming_msg(self, msg):
+ #
+ # Decode message and returns a tuple.
+ #
+ msgobj = self.id_msgdef[i]
+ if 'context' in msgobj.field_by_name and context >= 0:
+ return True
+ return False
+
+ def decode_incoming_msg(self, msg, no_type_conversion=False):
if not msg:
self.logger.warning('vpp_api.read failed')
return
+
(i, ci), size = self.header.unpack(msg, 0)
if self.id_names[i] == 'rx_thread_exit':
return
#
msgobj = self.id_msgdef[i]
if not msgobj:
- raise IOError(2, 'Reply message undefined')
+ raise VPPIOError(2, 'Reply message undefined')
- r, size = msgobj.unpack(msg)
+ r, size = msgobj.unpack(msg, ntc=no_type_conversion)
return r
def msg_handler_async(self, msg):
def validate_args(self, msg, kwargs):
d = set(kwargs.keys()) - set(msg.field_by_name.keys())
if d:
- raise ValueError('Invalid argument {} to {}'
- .format(list(d), msg.name))
+ raise VPPValueError('Invalid argument {} to {}'
+ .format(list(d), msg.name))
- def _call_vpp(self, i, msg, multipart, **kwargs):
+ def _call_vpp(self, i, msgdef, multipart, **kwargs):
"""Given a message, send the message and await a reply.
msgdef - the message packing definition
context = kwargs['context']
kwargs['_vl_msg_id'] = i
+ no_type_conversion = kwargs.pop('_no_type_conversion', False)
+
try:
if self.transport.socket_index:
kwargs['client_index'] = self.transport.socket_index
except AttributeError:
pass
- self.validate_args(msg, kwargs)
- b = msg.pack(kwargs)
+ self.validate_args(msgdef, kwargs)
+
+ logging.debug(call_logger(msgdef, kwargs))
+
+ b = msgdef.pack(kwargs)
self.transport.suspend()
self.transport.write(b)
while (True):
msg = self.transport.read()
if not msg:
- raise IOError(2, 'VPP API client: read failed')
- r = self.decode_incoming_msg(msg)
+ raise VPPIOError(2, 'VPP API client: read failed')
+ r = self.decode_incoming_msg(msg, no_type_conversion)
msgname = type(r).__name__
if context not in r or r.context == 0 or context != r.context:
# Message being queued
self.transport.resume()
+ logger.debug(return_logger(rl))
return rl
def _call_vpp_async(self, i, msg, **kwargs):
Code Review / vpp.git / blobdiff