X-Git-Url: https://gerrit.fd.io/r/gitweb?a=blobdiff_plain;f=src%2Fvpp-api%2Fpython%2Fvpp_papi%2Fvpp_papi.py;h=9c4ede90d48d4ae056f8d6f79e397b23f88af42a;hb=8006c6a;hp=0c40f1710e5a59db188fe25d4921b5c2a72b43b6;hpb=dfc9b7cac857a3a49555f9fc448bd2c6aa3400a6;p=vpp.git diff --git a/src/vpp-api/python/vpp_papi/vpp_papi.py b/src/vpp-api/python/vpp_papi/vpp_papi.py index 0c40f1710e5..9c4ede90d48 100644 --- a/src/vpp-api/python/vpp_papi/vpp_papi.py +++ b/src/vpp-api/python/vpp_papi/vpp_papi.py @@ -15,24 +15,72 @@ # from __future__ import print_function -import sys, os, logging, collections, struct, json, threading, glob -import atexit, Queue +from __future__ import absolute_import +import sys +import os +import logging +import collections +import struct +import json +import threading +import fnmatch +import weakref +import atexit +from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType, BaseTypes +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 +else: + import queue as queue + + +class VppEnumType(type): + def __getattr__(cls, name): + t = vpp_get_type(name) + return t.enum + + +# Python3 +# class VppEnum(metaclass=VppEnumType): +# pass +class VppEnum(object): + __metaclass__ = VppEnumType + + +def vpp_atexit(vpp_weakref): + """Clean up VPP connection on shutdown.""" + vpp_instance = vpp_weakref() + if vpp_instance and vpp_instance.transport.connected: + vpp_instance.logger.debug('Cleaning up VPP on exit') + vpp_instance.disconnect() -logging.basicConfig(level=logging.DEBUG) -import vpp_api -def eprint(*args, **kwargs): - """Print critical diagnostics to stderr.""" - print(*args, file=sys.stderr, **kwargs) +if sys.version[0] == '2': + def vpp_iterator(d): + return d.iteritems() +else: + def vpp_iterator(d): + return d.items() -def vpp_atexit(self): - """Clean up VPP connection on shutdown.""" - if self.connected: - eprint ('Cleaning up VPP on exit') - self.disconnect() + +def call_logger(msgdef, kwargs): + s = 'Calling {}('.format(msgdef.name) + for k, v in kwargs.items(): + s += '{}:{} '.format(k, v) + s += ')' + return s -class Empty(object): +def return_logger(r): + s = 'Return from {}'.format(r) + return s + + +class VppApiDynamicMethodHolder(object): pass @@ -45,7 +93,27 @@ class FuncWrapper(object): return self._func(**kwargs) -class VPP(): +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. This class provides the APIs to VPP. The APIs are loaded @@ -57,7 +125,72 @@ class VPP(): provides a means to register a callback function to receive these messages in a background thread. """ - def __init__(self, apifiles = None, testmode = False, async_thread = True): + VPPApiError = VPPApiError + VPPRuntimeError = VPPRuntimeError + VPPValueError = VPPValueError + VPPNotImplementedError = VPPNotImplementedError + VPPIOError = VPPIOError + + def process_json_file(self, apidef_file): + api = json.load(apidef_file) + types = {} + for t in api['enums']: + t[0] = 'vl_api_' + t[0] + '_t' + types[t[0]] = {'type': 'enum', 'data': t} + for t in api['unions']: + t[0] = 'vl_api_' + t[0] + '_t' + types[t[0]] = {'type': 'union', 'data': t} + 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(k): + if v['type'] == 'enum': + try: + VPPEnumType(t[0], t[1:]) + except ValueError: + unresolved[k] = v + elif v['type'] == 'union': + try: + VPPUnionType(t[0], t[1:]) + except ValueError: + unresolved[k] = v + elif v['type'] == 'type': + try: + 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 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 VPPNotImplementedError: + self.logger.error('Not implemented error for {}'.format(m[0])) + + def __init__(self, apifiles=None, testmode=False, async_thread=True, + logger=None, loglevel=None, + read_timeout=5, use_socket=False, + server_address='/run/vpp-api.sock'): """Create a VPP API object. apifiles is a list of files containing API @@ -65,351 +198,280 @@ class VPP(): dynamically created reflecting these APIs. If not provided this will load the API files from VPP's default install location. + + logger, if supplied, is the logging logger object to log to. + loglevel, if supplied, is the log level this logger is set + to report at (from the loglevels in the logging module). """ + if logger is None: + logger = logging.getLogger(__name__) + if loglevel is not None: + logger.setLevel(loglevel) + self.logger = logger + self.messages = {} + self.services = {} self.id_names = [] self.id_msgdef = [] - self.buffersize = 10000 - self.connected = False - self.header = struct.Struct('>HI') + self.header = VPPType('header', [['u16', 'msgid'], + ['u32', 'client_index']]) self.apifiles = [] self.event_callback = None - self.message_queue = Queue.Queue() - self.read_timeout = 0 - self.vpp_api = vpp_api - if async_thread: - self.event_thread = threading.Thread(target=self.thread_msg_handler) - self.event_thread.daemon = True - self.event_thread.start() + self.message_queue = queue.Queue() + self.read_timeout = read_timeout + self.async_thread = async_thread + + if use_socket: + from . vpp_transport_socket import VppTransport + else: + from . vpp_transport_shmem import VppTransport if not apifiles: # Pick up API definitions from default directory - apifiles = glob.glob('/usr/share/vpp/api/*.api.json') + try: + apifiles = self.find_api_files() + except RuntimeError: + # In test mode we don't care that we can't find the API files + if testmode: + apifiles = [] + else: + raise VPPRuntimeError for file in apifiles: with open(file) as apidef_file: - api = json.load(apidef_file) - for t in api['types']: - self.add_type(t[0], t[1:]) + self.process_json_file(apidef_file) - for m in api['messages']: - self.add_message(m[0], m[1:]) - self.apifiles = apifiles + self.apifiles = apifiles # 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) # Make sure we allow VPP to clean up the message rings. - atexit.register(vpp_atexit, self) + atexit.register(vpp_atexit, weakref.ref(self)) class ContextId(object): """Thread-safe provider of unique context IDs.""" def __init__(self): self.context = 0 - self.lock = threading.Lock() + self.lock = threading.Lock() + def __call__(self): """Get a new unique (or, at least, not recently used) context.""" - with self.lock: - self.context += 1 - return self.context + with self.lock: + self.context += 1 + return self.context get_context = ContextId() - def status(self): - """Debug function: report current VPP API status to stdout.""" - print('Connected') if self.connected else print('Not Connected') - print('Read API definitions from', ', '.join(self.apifiles)) - - def __struct (self, t, n = None, e = -1, vl = None): - """Create a packing structure for a message.""" - base_types = { 'u8' : 'B', - 'u16' : 'H', - 'u32' : 'I', - 'i32' : 'i', - 'u64' : 'Q', - 'f64' : 'd', - } - pack = None - if t in base_types: - pack = base_types[t] - if not vl: - if e > 0 and t == 'u8': - # Fixed byte array - return struct.Struct('>' + str(e) + 's') - if e > 0: - # Fixed array of base type - return [e, struct.Struct('>' + base_types[t])] - elif e == 0: - # Old style variable array - return [-1, struct.Struct('>' + base_types[t])] - else: - # Variable length array - return [vl, struct.Struct('>s')] if t == 'u8' else \ - [vl, struct.Struct('>' + base_types[t])] - - return struct.Struct('>' + base_types[t]) - - if t in self.messages: - ### Return a list in case of array ### - if e > 0 and not vl: - return [e, lambda self, encode, buf, offset, args: ( - self.__struct_type(encode, self.messages[t], buf, offset, - args))] - if vl: - return [vl, lambda self, encode, buf, offset, args: ( - self.__struct_type(encode, self.messages[t], buf, offset, - args))] - elif e == 0: - # Old style VLA - raise NotImplementedError(1, 'No support for compound types ' + t) - return lambda self, encode, buf, offset, args: ( - self.__struct_type(encode, self.messages[t], buf, offset, args) - ) - - raise ValueError(1, 'Invalid message type: ' + t) - - def __struct_type(self, encode, msgdef, buf, offset, kwargs): - """Get a message packer or unpacker.""" - if encode: - return self.__struct_type_encode(msgdef, buf, offset, kwargs) - else: - return self.__struct_type_decode(msgdef, buf, offset) - - def __struct_type_encode(self, msgdef, buf, offset, kwargs): - off = offset - size = 0 - - for k in kwargs: - if k not in msgdef['args']: - raise ValueError(1, 'Invalid field-name in message call ' + k) - - for k,v in msgdef['args'].iteritems(): - off += size - if k in kwargs: - if type(v) is list: - if callable(v[1]): - e = kwargs[v[0]] if v[0] in kwargs else v[0] - size = 0 - for i in range(e): - size += v[1](self, True, buf, off + size, - kwargs[k][i]) - else: - if v[0] in kwargs: - l = kwargs[v[0]] - else: - l = len(kwargs[k]) - if v[1].size == 1: - buf[off:off + l] = bytearray(kwargs[k]) - size = l - else: - size = 0 - for i in kwargs[k]: - v[1].pack_into(buf, off + size, i) - size += v[1].size - else: - if callable(v): - size = v(self, True, buf, off, kwargs[k]) - else: - v.pack_into(buf, off, kwargs[k]) - size = v.size - else: - size = v.size if not type(v) is list else 0 - - return off + size - offset + def get_type(self, name): + return vpp_get_type(name) + @classmethod + def find_api_dir(cls): + """Attempt to find the best directory in which API definition + files may reside. If the value VPP_API_DIR exists in the environment + then it is first on the search list. If we're inside a recognized + location in a VPP source tree (src/scripts and src/vpp-api/python) + then entries from there to the likely locations in build-root are + added. Finally the location used by system packages is added. - def __getitem__(self, name): - if name in self.messages: - return self.messages[name] - return None + :returns: A single directory name, or None if no such directory + could be found. + """ + dirs = [] - def encode(self, msgdef, kwargs): - # Make suitably large buffer - buf = bytearray(self.buffersize) - offset = 0 - size = self.__struct_type(True, msgdef, buf, offset, kwargs) - return buf[:offset + size] - - def decode(self, msgdef, buf): - return self.__struct_type(False, msgdef, buf, 0, None)[1] - - def __struct_type_decode(self, msgdef, buf, offset): - res = [] - off = offset - size = 0 - for k,v in msgdef['args'].iteritems(): - off += size - if type(v) is list: - lst = [] - if callable(v[1]): # compound type - size = 0 - if v[0] in msgdef['args']: # vla - e = res[v[2]] - else: # fixed array - e = v[0] - res.append(lst) - for i in range(e): - (s,l) = v[1](self, False, buf, off + size, None) - lst.append(l) - size += s - continue - if v[1].size == 1: - if type(v[0]) is int: - size = len(buf) - off - else: - size = res[v[2]] - res.append(buf[off:off + size]) - else: - e = v[0] if type(v[0]) is int else res[v[2]] - if e == -1: - e = (len(buf) - off) / v[1].size - lst = [] - res.append(lst) - size = 0 - for i in range(e): - lst.append(v[1].unpack_from(buf, off + size)[0]) - size += v[1].size - else: - if callable(v): - (s,l) = v(self, False, buf, off, None) - res.append(l) - size += s - else: - res.append(v.unpack_from(buf, off)[0]) - size = v.size + if 'VPP_API_DIR' in os.environ: + dirs.append(os.environ['VPP_API_DIR']) - return off + size - offset, msgdef['return_tuple']._make(res) + # perhaps we're in the 'src/scripts' or 'src/vpp-api/python' dir; + # in which case, plot a course to likely places in the src tree + import __main__ as main + if hasattr(main, '__file__'): + # get the path of the calling script + localdir = os.path.dirname(os.path.realpath(main.__file__)) + else: + # use cwd if there is no calling script + localdir = os.getcwd() + localdir_s = localdir.split(os.path.sep) + + def dmatch(dir): + """Match dir against right-hand components of the script dir""" + d = dir.split('/') # param 'dir' assumes a / separator + length = len(d) + return len(localdir_s) > length and localdir_s[-length:] == d + + def sdir(srcdir, variant): + """Build a path from srcdir to the staged API files of + 'variant' (typically '' or '_debug')""" + # Since 'core' and 'plugin' files are staged + # in separate directories, we target the parent dir. + return os.path.sep.join(( + srcdir, + 'build-root', + 'install-vpp%s-native' % variant, + 'vpp', + 'share', + 'vpp', + 'api', + )) + + srcdir = None + if dmatch('src/scripts'): + srcdir = os.path.sep.join(localdir_s[:-2]) + elif dmatch('src/vpp-api/python'): + srcdir = os.path.sep.join(localdir_s[:-3]) + elif dmatch('test'): + # we're apparently running tests + srcdir = os.path.sep.join(localdir_s[:-1]) + + if srcdir: + # we're in the source tree, try both the debug and release + # variants. + dirs.append(sdir(srcdir, '_debug')) + dirs.append(sdir(srcdir, '')) + + # Test for staged copies of the scripts + # For these, since we explicitly know if we're running a debug versus + # release variant, target only the relevant directory + if dmatch('build-root/install-vpp_debug-native/vpp/bin'): + srcdir = os.path.sep.join(localdir_s[:-4]) + dirs.append(sdir(srcdir, '_debug')) + if dmatch('build-root/install-vpp-native/vpp/bin'): + srcdir = os.path.sep.join(localdir_s[:-4]) + dirs.append(sdir(srcdir, '')) + + # finally, try the location system packages typically install into + dirs.append(os.path.sep.join(('', 'usr', 'share', 'vpp', 'api'))) + + # check the directories for existance; first one wins + for dir in dirs: + if os.path.isdir(dir): + return dir - def ret_tup(self, name): - if name in self.messages and 'return_tuple' in self.messages[name]: - return self.messages[name]['return_tuple'] return None - def add_message(self, name, msgdef): - if name in self.messages: - raise ValueError('Duplicate message name: ' + name) - - args = collections.OrderedDict() - argtypes = collections.OrderedDict() - fields = [] - msg = {} - for i, f in enumerate(msgdef): - if type(f) is dict and 'crc' in f: - msg['crc'] = f['crc'] - continue - field_type = f[0] - field_name = f[1] - if len(f) == 3 and f[2] == 0 and i != len(msgdef) - 2: - raise ValueError('Variable Length Array must be last: ' + name) - args[field_name] = self.__struct(*f) - argtypes[field_name] = field_type - if len(f) == 4: # Find offset to # elements field - args[field_name].append(args.keys().index(f[3]) - i) - fields.append(field_name) - msg['return_tuple'] = collections.namedtuple(name, fields, - rename = True) - self.messages[name] = msg - self.messages[name]['args'] = args - self.messages[name]['argtypes'] = argtypes - return self.messages[name] - - def add_type(self, name, typedef): - return self.add_message('vl_api_' + name + '_t', typedef) - - def make_function(self, name, i, msgdef, multipart, async): - if (async): - f = lambda **kwargs: (self._call_vpp_async(i, msgdef, **kwargs)) + @classmethod + def find_api_files(cls, api_dir=None, patterns='*'): + """Find API definition files from the given directory tree with the + given pattern. If no directory is given then find_api_dir() is used + to locate one. If no pattern is given then all definition files found + in the directory tree are used. + + :param api_dir: A directory tree in which to locate API definition + files; subdirectories are descended into. + If this is None then find_api_dir() is called to discover it. + :param patterns: A list of patterns to use in each visited directory + when looking for files. + This can be a list/tuple object or a comma-separated string of + patterns. Each value in the list will have leading/trialing + whitespace stripped. + The pattern specifies the first part of the filename, '.api.json' + is appended. + The results are de-duplicated, thus overlapping patterns are fine. + If this is None it defaults to '*' meaning "all API files". + :returns: A list of file paths for the API files found. + """ + if api_dir is None: + api_dir = cls.find_api_dir() + if api_dir is None: + raise VPPApiError("api_dir cannot be located") + + if isinstance(patterns, list) or isinstance(patterns, tuple): + patterns = [p.strip() + '.api.json' for p in patterns] else: - f = lambda **kwargs: (self._call_vpp(i, msgdef, multipart, **kwargs)) - args = self.messages[name]['args'] - argtypes = self.messages[name]['argtypes'] - f.__name__ = str(name) - f.__doc__ = ", ".join(["%s %s" % (argtypes[k], k) for k in args.keys()]) - return f + patterns = [p.strip() + '.api.json' for p in patterns.split(",")] + + api_files = [] + for root, dirnames, files in os.walk(api_dir): + # iterate all given patterns and de-dup the result + files = set(sum([fnmatch.filter(files, p) for p in patterns], [])) + for filename in files: + api_files.append(os.path.join(root, filename)) + + return api_files @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 _register_functions(self, async=False): + def make_function(self, msg, i, multipart, do_async): + if (do_async): + def f(**kwargs): + return self._call_vpp_async(i, msg, **kwargs) + else: + def f(**kwargs): + return self._call_vpp(i, msg, multipart, **kwargs) + + f.__name__ = str(msg.name) + f.__doc__ = ", ".join(["%s %s" % + (msg.fieldtypes[j], k) + for j, k in enumerate(msg.fields)]) + return f + + def _register_functions(self, do_async=False): self.id_names = [None] * (self.vpp_dictionary_maxid + 1) self.id_msgdef = [None] * (self.vpp_dictionary_maxid + 1) - self._api = Empty() - for name, msgdef in self.messages.iteritems(): - if name in self.vpp_dictionary: - if self.messages[name]['crc'] != self.vpp_dictionary[name]['crc']: - raise ValueError(3, 'Failed CRC checksum ' + name + - ' ' + self.messages[name]['crc'] + - ' ' + self.vpp_dictionary[name]['crc']) - i = self.vpp_dictionary[name]['id'] - self.id_msgdef[i] = msgdef + self._api = VppApiDynamicMethodHolder() + for name, msg in vpp_iterator(self.messages): + n = name + '_' + msg.crc[2:] + i = self.transport.get_msg_index(n.encode()) + if i > 0: + self.id_msgdef[i] = msg self.id_names[i] = name - multipart = True if name.find('_dump') > 0 else False - f = self.make_function(name, i, msgdef, multipart, async) - setattr(self._api, name, FuncWrapper(f)) - - # old API stuff starts here - will be removed in 17.07 - if hasattr(self, name): - raise NameError( - 3, "Conflicting name in JSON definition: `%s'" % name) - setattr(self, name, f) - # old API stuff ends here - - def _write (self, buf): - """Send a binary-packed message to VPP.""" - if not self.connected: - raise IOError(1, 'Not connected') - return vpp_api.write(str(buf)) - - def _read (self): - if not self.connected: - raise IOError(1, 'Not connected') - - return vpp_api.read(self.read_timeout) - - def _load_dictionary(self): - self.vpp_dictionary = {} - self.vpp_dictionary_maxid = 0 - d = vpp_api.msg_table() - - if not d: - raise IOError(3, 'Cannot get VPP API dictionary') - for i,n in d: - name, crc = n.rsplit('_', 1) - crc = '0x' + crc - self.vpp_dictionary[name] = { 'id' : i, 'crc' : crc } - self.vpp_dictionary_maxid = max(self.vpp_dictionary_maxid, i) - - def connect_internal(self, name, msg_handler, chroot_prefix, rx_qlen, async): - rv = vpp_api.connect(name, msg_handler, chroot_prefix, rx_qlen) - if rv != 0: - raise IOError(2, 'Connect failed') - self.connected = True - self._load_dictionary() - self._register_functions(async=async) + # 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) + + def connect_internal(self, name, msg_handler, chroot_prefix, rx_qlen, + do_async): + pfx = chroot_prefix.encode() if chroot_prefix else None + + rv = self.transport.connect(name.encode(), pfx, msg_handler, rx_qlen) + if rv != 0: + raise VPPIOError(2, 'Connect failed') + self.vpp_dictionary_maxid = self.transport.msg_table_max_index() + self._register_functions(do_async=do_async) # Initialise control ping - self.control_ping_index = self.vpp_dictionary['control_ping']['id'] + crc = self.messages['control_ping'].crc + self.control_ping_index = self.transport.get_msg_index( + ('control_ping' + '_' + crc[2:]).encode()) self.control_ping_msgdef = self.messages['control_ping'] + if self.async_thread: + self.event_thread = threading.Thread( + target=self.thread_msg_handler) + self.event_thread.daemon = True + self.event_thread.start() + return rv - def connect(self, name, chroot_prefix = None, async = False, rx_qlen = 32): + def connect(self, name, chroot_prefix=None, do_async=False, rx_qlen=32): """Attach to VPP. name - the name of the client. chroot_prefix - if VPP is chroot'ed, the prefix of the jail - async - if true, messages are sent without waiting for a reply + do_async - if true, messages are sent without waiting for a reply rx_qlen - the length of the VPP message receive queue between client and server. """ - msg_handler = self.msg_handler_sync if not async \ - else self.msg_handler_async + msg_handler = self.transport.get_callback(do_async) return self.connect_internal(name, msg_handler, chroot_prefix, rx_qlen, - async) + do_async) - def connect_sync (self, name, chroot_prefix = None, rx_qlen = 32): + def connect_sync(self, name, chroot_prefix=None, rx_qlen=32): """Attach to VPP in synchronous mode. Application must poll for events. name - the name of the client. @@ -419,12 +481,12 @@ class VPP(): """ return self.connect_internal(name, None, chroot_prefix, rx_qlen, - async=False) + do_async=False) def disconnect(self): """Detach from VPP.""" - rv = vpp_api.disconnect() - self.connected = False + rv = self.transport.disconnect() + self.message_queue.put("terminate event thread") return rv def msg_handler_sync(self, msg): @@ -442,32 +504,49 @@ class VPP(): if hasattr(r, 'context') and r.context > 0: context = r.context - msgname = type(r).__name__ - if context == 0: # 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 + + # + # 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): + def decode_incoming_msg(self, msg, no_type_conversion=False): if not msg: - eprint('vpp_api.read failed') + self.logger.warning('vpp_api.read failed') return - i, ci = self.header.unpack_from(msg, 0) + (i, ci), size = self.header.unpack(msg, 0) if self.id_names[i] == 'rx_thread_exit': return # # Decode message and returns a tuple. # - msgdef = self.id_msgdef[i] - if not msgdef: - raise IOError(2, 'Reply message undefined') - - r = self.decode(msgdef, msg) + msgobj = self.id_msgdef[i] + if not msgobj: + raise VPPIOError(2, 'Reply message undefined') + r, size = msgobj.unpack(msg, ntc=no_type_conversion) return r def msg_handler_async(self, msg): @@ -481,15 +560,21 @@ class VPP(): msgname = type(r).__name__ - if self.event_callback: - self.event_callback(msgname, r) + if self.event_callback: + self.event_callback(msgname, r) def _control_ping(self, context): """Send a ping command.""" self._call_vpp_async(self.control_ping_index, - self.control_ping_msgdef, + self.control_ping_msgdef, context=context) + def validate_args(self, msg, kwargs): + d = set(kwargs.keys()) - set(msg.field_by_name.keys()) + if d: + raise VPPValueError('Invalid argument {} to {}' + .format(list(d), msg.name)) + def _call_vpp(self, i, msgdef, multipart, **kwargs): """Given a message, send the message and await a reply. @@ -506,16 +591,28 @@ class VPP(): no response within the timeout window. """ - if not 'context' in kwargs: + if 'context' not in kwargs: context = self.get_context() kwargs['context'] = context else: context = kwargs['context'] kwargs['_vl_msg_id'] = i - b = self.encode(msgdef, kwargs) - vpp_api.suspend() - self._write(b) + 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(msgdef, kwargs) + + logging.debug(call_logger(msgdef, kwargs)) + + b = msgdef.pack(kwargs) + self.transport.suspend() + + self.transport.write(b) if multipart: # Send a ping after the request - we use its response @@ -525,14 +622,13 @@ class VPP(): # Block until we get a reply. rl = [] while (True): - msg = self._read() + msg = self.transport.read() if not msg: - print('PNEUM ERROR: OH MY GOD') - raise IOError(2, 'PNEUM 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 not context in r or r.context == 0 or context != r.context: + if context not in r or r.context == 0 or context != r.context: + # Message being queued self.message_queue.put_nowait(r) continue @@ -544,11 +640,12 @@ class VPP(): rl.append(r) - vpp_api.resume() + self.transport.resume() + logger.debug(return_logger(rl)) return rl - def _call_vpp_async(self, i, msgdef, **kwargs): + def _call_vpp_async(self, i, msg, **kwargs): """Given a message, send the message and await a reply. msgdef - the message packing definition @@ -557,15 +654,20 @@ class VPP(): supplied. The remainder of the kwargs are the arguments to the API call. """ - if not 'context' in kwargs: + if 'context' not in kwargs: context = self.get_context() kwargs['context'] = context else: context = kwargs['context'] + try: + if self.transport.socket_index: + kwargs['client_index'] = self.transport.socket_index + except AttributeError: + kwargs['client_index'] = 0 kwargs['_vl_msg_id'] = i - b = self.encode(msgdef, kwargs) + b = msg.pack(kwargs) - self._write(b) + self.transport.write(b) def register_event_callback(self, callback): """Register a callback for async messages. @@ -586,7 +688,7 @@ class VPP(): self.event_callback = callback def thread_msg_handler(self): - """Python thread calling the user registerd message handler. + """Python thread calling the user registered message handler. This is to emulate the old style event callback scheme. Modern clients should provide their own thread to poll the event @@ -594,6 +696,11 @@ class VPP(): """ while True: r = self.message_queue.get() + if r == "terminate event thread": + break msgname = type(r).__name__ - if self.event_callback: - self.event_callback(msgname, r) + if self.event_callback: + self.event_callback(msgname, r) + + +# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4