import sys
import multiprocessing as mp
import os
+import queue
import logging
import functools
import json
import atexit
import time
from . vpp_format import verify_enum_hint
-from . vpp_serializer import VPPType, VPPEnumType, VPPUnionType
+from . vpp_serializer import VPPType, VPPEnumType, VPPEnumFlagType, VPPUnionType
from . vpp_serializer import VPPMessage, vpp_get_type, VPPTypeAlias
-if sys.version[0] == '2':
- import Queue as queue
-else:
- import queue as queue
+try:
+ import VppTransport
+except ModuleNotFoundError:
+ class V:
+ """placeholder for VppTransport as the implementation is dependent on
+ VPPAPIClient's initialization values
+ """
+
+ VppTransport = V
+
+from . vpp_transport_socket import VppTransport
-__all__ = ('FuncWrapper', 'VPP', 'VppApiDynamicMethodHolder',
- 'VppEnum', 'VppEnumType',
+logger = logging.getLogger('vpp_papi')
+logger.addHandler(logging.NullHandler())
+
+__all__ = ('FuncWrapper', 'VppApiDynamicMethodHolder',
+ 'VppEnum', 'VppEnumType', 'VppEnumFlag',
'VPPIOError', 'VPPRuntimeError', 'VPPValueError',
'VPPApiClient', )
@metaclass(VppEnumType)
-class VppEnum(object):
+class VppEnum:
+ pass
+
+
+@metaclass(VppEnumType)
+class VppEnumFlag:
pass
"""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')
+ logger.debug('Cleaning up VPP on exit')
vpp_instance.disconnect()
-if sys.version[0] == '2':
- def vpp_iterator(d):
- return d.iteritems()
-else:
- def vpp_iterator(d):
- return d.items()
-
-
def add_convenience_methods():
# provide convenience methods to IP[46]Address.vapi_af
def _vapi_af(self):
ipaddress._IPAddressBase.vapi_af_name = property(_vapi_af_name)
-class VppApiDynamicMethodHolder(object):
+class VppApiDynamicMethodHolder:
pass
-class FuncWrapper(object):
+class FuncWrapper:
def __init__(self, func):
self._func = func
self.__name__ = func.__name__
pass
-class VPPApiJSONFiles(object):
+class VPPApiJSONFiles:
@classmethod
def find_api_dir(cls, dirs):
"""Attempt to find the best directory in which API definition
return None
@classmethod
- def find_api_files(cls, api_dir=None, patterns='*'):
+ def find_api_files(cls, api_dir=None, patterns='*'): # -> list
"""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
@classmethod
def process_json_file(self, apidef_file):
api = json.load(apidef_file)
+ return self._process_json(api)
+
+ @classmethod
+ def process_json_str(self, json_str):
+ api = json.loads(json_str)
+ return self._process_json(api)
+
+ @staticmethod
+ def _process_json(api): # -> Tuple[Dict, Dict]
types = {}
services = {}
messages = {}
- 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}
- services.update(api['services'])
+ try:
+ for t in api['enums']:
+ t[0] = 'vl_api_' + t[0] + '_t'
+ types[t[0]] = {'type': 'enum', 'data': t}
+ except KeyError:
+ pass
+ try:
+ for t in api['enumflags']:
+ t[0] = 'vl_api_' + t[0] + '_t'
+ types[t[0]] = {'type': 'enum', 'data': t}
+ except KeyError:
+ pass
+ try:
+ for t in api['unions']:
+ t[0] = 'vl_api_' + t[0] + '_t'
+ types[t[0]] = {'type': 'union', 'data': t}
+ except KeyError:
+ pass
+
+ try:
+ for t in api['types']:
+ t[0] = 'vl_api_' + t[0] + '_t'
+ types[t[0]] = {'type': 'type', 'data': t}
+ except KeyError:
+ pass
+
+ try:
+ for t, v in api['aliases'].items():
+ types['vl_api_' + t + '_t'] = {'type': 'alias', 'data': v}
+ except KeyError:
+ pass
+
+ try:
+ services.update(api['services'])
+ except KeyError:
+ pass
i = 0
while True:
VPPEnumType(t[0], t[1:])
except ValueError:
unresolved[k] = v
+ if not vpp_get_type(k):
+ if v['type'] == 'enumflag':
+ try:
+ VPPEnumFlagType(t[0], t[1:])
+ except ValueError:
+ unresolved[k] = v
elif v['type'] == 'union':
try:
VPPUnionType(t[0], t[1:])
.format(unresolved))
types = unresolved
i += 1
-
- for m in api['messages']:
- try:
- messages[m[0]] = VPPMessage(m[0], m[1:])
- except VPPNotImplementedError:
- ### OLE FIXME
- self.logger.error('Not implemented error for {}'.format(m[0]))
+ try:
+ for m in api['messages']:
+ try:
+ messages[m[0]] = VPPMessage(m[0], m[1:])
+ except VPPNotImplementedError:
+ ### OLE FIXME
+ logger.error('Not implemented error for {}'.format(m[0]))
+ except KeyError:
+ pass
return messages, services
-class VPPApiClient(object):
+class VPPApiClient:
"""VPP interface.
This class provides the APIs to VPP. The APIs are loaded
VPPIOError = VPPIOError
- def __init__(self, apifiles=None, testmode=False, async_thread=True,
+ def __init__(self, *, apifiles=None, testmode=False, async_thread=True,
logger=None, loglevel=None,
- read_timeout=5, use_socket=False,
+ read_timeout=5, use_socket=True,
server_address='/run/vpp/api.sock'):
"""Create a VPP API object.
self.async_thread = async_thread
self.event_thread = None
self.testmode = testmode
- self.use_socket = use_socket
self.server_address = server_address
self._apifiles = apifiles
self.stats = {}
- 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
try:
apifiles = VPPApiJSONFiles.find_api_files(self.apidir)
- except RuntimeError:
+ except (RuntimeError, VPPApiError):
# In test mode we don't care that we can't find the API files
if testmode:
apifiles = []
def get_function(self, name):
return getattr(self._api, name)
- class ContextId(object):
+ class ContextId:
"""Multiprocessing-safe provider of unique context IDs."""
def __init__(self):
self.context = mp.Value(ctypes.c_uint, 0)
self.id_names = [None] * (self.vpp_dictionary_maxid + 1)
self.id_msgdef = [None] * (self.vpp_dictionary_maxid + 1)
self._api = VppApiDynamicMethodHolder()
- for name, msg in vpp_iterator(self.messages):
+ for name, msg in self.messages.items():
n = name + '_' + msg.crc[2:]
i = self.transport.get_msg_index(n)
if i > 0:
# 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)
+ f = self.make_function(msg, i, self.services[name], do_async)
setattr(self._api, name, FuncWrapper(f))
else:
self.logger.debug(
def decode_incoming_msg(self, msg, no_type_conversion=False):
if not msg:
- self.logger.warning('vpp_api.read failed')
+ logger.warning('vpp_api.read failed')
return
(i, ci), size = self.header.unpack(msg, 0)
n[1]['avg'], n[1]['max'])
return s
- def _call_vpp(self, i, msgdef, multipart, **kwargs):
+ def get_field_options(self, msg, fld_name):
+ # when there is an option, the msgdef has 3 elements.
+ # ['u32', 'ring_size', {'default': 1024}]
+ for _def in self.messages[msg].msgdef:
+ if isinstance(_def, list) and \
+ len(_def) == 3 and \
+ _def[1] == fld_name:
+ return _def[2]
+
+ def _call_vpp(self, i, msgdef, service, **kwargs):
"""Given a message, send the message and await a reply.
msgdef - the message packing definition
self.transport.write(b)
- if multipart:
- # Send a ping after the request - we use its response
- # to detect that we have seen all results.
- self._control_ping(context)
+ msgreply = service['reply']
+ stream = True if 'stream' in service else False
+ if stream:
+ if 'stream_msg' in service:
+ # New service['reply'] = _reply and service['stream_message'] = _details
+ stream_message = service['stream_msg']
+ modern =True
+ else:
+ # Old service['reply'] = _details
+ stream_message = msgreply
+ msgreply = 'control_ping_reply'
+ modern = False
+ # Send a ping after the request - we use its response
+ # to detect that we have seen all results.
+ self._control_ping(context)
# Block until we get a reply.
rl = []
# Message being queued
self.message_queue.put_nowait(r)
continue
-
- if not multipart:
+ if msgname != msgreply and (stream and (msgname != stream_message)):
+ print('REPLY MISMATCH', msgreply, msgname, stream_message, stream)
+ if not stream:
rl = r
break
- if msgname == 'control_ping_reply':
+ if msgname == msgreply:
+ if modern: # Return both reply and list
+ rl = r, rl
break
rl.append(r)
def __repr__(self):
return "<VPPApiClient apifiles=%s, testmode=%s, async_thread=%s, " \
- "logger=%s, read_timeout=%s, use_socket=%s, " \
+ "logger=%s, read_timeout=%s, " \
"server_address='%s'>" % (
self._apifiles, self.testmode, self.async_thread,
- self.logger, self.read_timeout, self.use_socket,
- self.server_address)
-
-
-# Provide the old name for backward compatibility.
-VPP = VPPApiClient
+ self.logger, self.read_timeout, self.server_address)
-# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
+ def details_iter(self, f, **kwargs):
+ cursor = 0
+ while True:
+ kwargs['cursor'] = cursor
+ rv, details = f(**kwargs)
+ for d in details:
+ yield d
+ if rv.retval == 0 or rv.retval != -165:
+ break
+ cursor = rv.cursor
Code Review / vpp.git / blobdiff