rpc integration

12 files changed, 812 insertions, 0 deletions

diff --git a/src/console/zmq/backend/cffi/__init__.py b/src/console/zmq/backend/cffi/__init__.py
new file mode 100755
index 00000000..ca3164d3
--- /dev/null
+++ b/src/console/zmq/backend/cffi/__init__.py
@@ -0,0 +1,22 @@
+"""CFFI backend (for PyPY)"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+from zmq.backend.cffi import (constants, error, message, context, socket,
+                           _poll, devices, utils)
+
+__all__ = []
+for submod in (constants, error, message, context, socket,
+               _poll, devices, utils):
+    __all__.extend(submod.__all__)
+
+from .constants import *
+from .error import *
+from .message import *
+from .context import *
+from .socket import *
+from .devices import *
+from ._poll import *
+from ._cffi import zmq_version_info, ffi
+from .utils import *
diff --git a/src/console/zmq/backend/cffi/_cdefs.h b/src/console/zmq/backend/cffi/_cdefs.h
new file mode 100644
index 00000000..d3300575
--- /dev/null
+++ b/src/console/zmq/backend/cffi/_cdefs.h
@@ -0,0 +1,68 @@
+void zmq_version(int *major, int *minor, int *patch);
+
+void* zmq_socket(void *context, int type);
+int zmq_close(void *socket);
+
+int zmq_bind(void *socket, const char *endpoint);
+int zmq_connect(void *socket, const char *endpoint);
+
+int zmq_errno(void);
+const char * zmq_strerror(int errnum);
+
+void* zmq_stopwatch_start(void);
+unsigned long zmq_stopwatch_stop(void *watch);
+void zmq_sleep(int seconds_);
+int zmq_device(int device, void *frontend, void *backend);
+
+int zmq_unbind(void *socket, const char *endpoint);
+int zmq_disconnect(void *socket, const char *endpoint);
+void* zmq_ctx_new();
+int zmq_ctx_destroy(void *context);
+int zmq_ctx_get(void *context, int opt);
+int zmq_ctx_set(void *context, int opt, int optval);
+int zmq_proxy(void *frontend, void *backend, void *capture);
+int zmq_socket_monitor(void *socket, const char *addr, int events);
+
+int zmq_curve_keypair (char *z85_public_key, char *z85_secret_key);
+int zmq_has (const char *capability);
+
+typedef struct { ...; } zmq_msg_t;
+typedef ... zmq_free_fn;
+
+int zmq_msg_init(zmq_msg_t *msg);
+int zmq_msg_init_size(zmq_msg_t *msg, size_t size);
+int zmq_msg_init_data(zmq_msg_t *msg,
+                      void *data,
+                      size_t size,
+                      zmq_free_fn *ffn,
+                      void *hint);
+
+size_t zmq_msg_size(zmq_msg_t *msg);
+void *zmq_msg_data(zmq_msg_t *msg);
+int zmq_msg_close(zmq_msg_t *msg);
+
+int zmq_msg_send(zmq_msg_t *msg, void *socket, int flags);
+int zmq_msg_recv(zmq_msg_t *msg, void *socket, int flags);
+
+int zmq_getsockopt(void *socket,
+                   int option_name,
+                   void *option_value,
+                   size_t *option_len);
+
+int zmq_setsockopt(void *socket,
+                   int option_name,
+                   const void *option_value,
+                   size_t option_len);
+typedef struct
+{
+    void *socket;
+    int fd;
+    short events;
+    short revents;
+} zmq_pollitem_t;
+
+int zmq_poll(zmq_pollitem_t *items, int nitems, long timeout);
+
+// miscellany
+void * memcpy(void *restrict s1, const void *restrict s2, size_t n);
+int get_ipc_path_max_len(void);
diff --git a/src/console/zmq/backend/cffi/_cffi.py b/src/console/zmq/backend/cffi/_cffi.py
new file mode 100755
index 00000000..c73ebf83
--- /dev/null
+++ b/src/console/zmq/backend/cffi/_cffi.py
@@ -0,0 +1,127 @@
+# coding: utf-8
+"""The main CFFI wrapping of libzmq"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+
+import json
+import os
+from os.path import dirname, join
+from cffi import FFI
+
+from zmq.utils.constant_names import all_names, no_prefix
+
+
+base_zmq_version = (3,2,2)
+
+def load_compiler_config():
+    """load pyzmq compiler arguments"""
+    import zmq
+    zmq_dir = dirname(zmq.__file__)
+    zmq_parent = dirname(zmq_dir)
+    
+    fname = join(zmq_dir, 'utils', 'compiler.json')
+    if os.path.exists(fname):
+        with open(fname) as f:
+            cfg = json.load(f)
+    else:
+        cfg = {}
+    
+    cfg.setdefault("include_dirs", [])
+    cfg.setdefault("library_dirs", [])
+    cfg.setdefault("runtime_library_dirs", [])
+    cfg.setdefault("libraries", ["zmq"])
+    
+    # cast to str, because cffi can't handle unicode paths (?!)
+    cfg['libraries'] = [str(lib) for lib in cfg['libraries']]
+    for key in ("include_dirs", "library_dirs", "runtime_library_dirs"):
+        # interpret paths relative to parent of zmq (like source tree)
+        abs_paths = []
+        for p in cfg[key]:
+            if p.startswith('zmq'):
+                p = join(zmq_parent, p)
+            abs_paths.append(str(p))
+        cfg[key] = abs_paths
+    return cfg
+
+
+def zmq_version_info():
+    """Get libzmq version as tuple of ints"""
+    major = ffi.new('int*')
+    minor = ffi.new('int*')
+    patch = ffi.new('int*')
+
+    C.zmq_version(major, minor, patch)
+
+    return (int(major[0]), int(minor[0]), int(patch[0]))
+
+
+cfg = load_compiler_config()
+ffi = FFI()
+
+def _make_defines(names):
+    _names = []
+    for name in names:
+        define_line = "#define %s ..." % (name)
+        _names.append(define_line)
+
+    return "\n".join(_names)
+
+c_constant_names = []
+for name in all_names:
+    if no_prefix(name):
+        c_constant_names.append(name)
+    else:
+        c_constant_names.append("ZMQ_" + name)
+
+# load ffi definitions
+here = os.path.dirname(__file__)
+with open(os.path.join(here, '_cdefs.h')) as f:
+    _cdefs = f.read()
+
+with open(os.path.join(here, '_verify.c')) as f:
+    _verify = f.read()
+
+ffi.cdef(_cdefs)
+ffi.cdef(_make_defines(c_constant_names))
+
+try:
+    C = ffi.verify(_verify,
+        modulename='_cffi_ext',
+        libraries=cfg['libraries'],
+        include_dirs=cfg['include_dirs'],
+        library_dirs=cfg['library_dirs'],
+        runtime_library_dirs=cfg['runtime_library_dirs'],
+    )
+    _version_info = zmq_version_info()
+except Exception as e:
+    raise ImportError("PyZMQ CFFI backend couldn't find zeromq: %s\n"
+    "Please check that you have zeromq headers and libraries." % e)
+
+if _version_info < (3,2,2):
+    raise ImportError("PyZMQ CFFI backend requires zeromq >= 3.2.2,"
+        " but found %i.%i.%i" % _version_info
+    )
+
+nsp = new_sizet_pointer = lambda length: ffi.new('size_t*', length)
+
+new_uint64_pointer = lambda: (ffi.new('uint64_t*'),
+                              nsp(ffi.sizeof('uint64_t')))
+new_int64_pointer = lambda: (ffi.new('int64_t*'),
+                             nsp(ffi.sizeof('int64_t')))
+new_int_pointer = lambda: (ffi.new('int*'),
+                           nsp(ffi.sizeof('int')))
+new_binary_data = lambda length: (ffi.new('char[%d]' % (length)),
+                                  nsp(ffi.sizeof('char') * length))
+
+value_uint64_pointer = lambda val : (ffi.new('uint64_t*', val),
+                                     ffi.sizeof('uint64_t'))
+value_int64_pointer = lambda val: (ffi.new('int64_t*', val),
+                                   ffi.sizeof('int64_t'))
+value_int_pointer = lambda val: (ffi.new('int*', val),
+                                 ffi.sizeof('int'))
+value_binary_data = lambda val, length: (ffi.new('char[%d]' % (length + 1), val),
+                                         ffi.sizeof('char') * length)
+
+IPC_PATH_MAX_LEN = C.get_ipc_path_max_len()
diff --git a/src/console/zmq/backend/cffi/_poll.py b/src/console/zmq/backend/cffi/_poll.py
new file mode 100755
index 00000000..9bca34ca
--- /dev/null
+++ b/src/console/zmq/backend/cffi/_poll.py
@@ -0,0 +1,56 @@
+# coding: utf-8
+"""zmq poll function"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+from ._cffi import C, ffi, zmq_version_info
+
+from .constants import *
+
+from zmq.error import _check_rc
+
+
+def _make_zmq_pollitem(socket, flags):
+    zmq_socket = socket._zmq_socket
+    zmq_pollitem = ffi.new('zmq_pollitem_t*')
+    zmq_pollitem.socket = zmq_socket
+    zmq_pollitem.fd = 0
+    zmq_pollitem.events = flags
+    zmq_pollitem.revents = 0
+    return zmq_pollitem[0]
+
+def _make_zmq_pollitem_fromfd(socket_fd, flags):
+    zmq_pollitem = ffi.new('zmq_pollitem_t*')
+    zmq_pollitem.socket = ffi.NULL
+    zmq_pollitem.fd = socket_fd
+    zmq_pollitem.events = flags
+    zmq_pollitem.revents = 0
+    return zmq_pollitem[0]
+
+def zmq_poll(sockets, timeout):
+    cffi_pollitem_list = []
+    low_level_to_socket_obj = {}
+    for item in sockets:
+        if isinstance(item[0], int):
+            low_level_to_socket_obj[item[0]] = item
+            cffi_pollitem_list.append(_make_zmq_pollitem_fromfd(item[0], item[1]))
+        else:
+            low_level_to_socket_obj[item[0]._zmq_socket] = item
+            cffi_pollitem_list.append(_make_zmq_pollitem(item[0], item[1]))
+    items = ffi.new('zmq_pollitem_t[]', cffi_pollitem_list)
+    list_length = ffi.cast('int', len(cffi_pollitem_list))
+    c_timeout = ffi.cast('long', timeout)
+    rc = C.zmq_poll(items, list_length, c_timeout)
+    _check_rc(rc)
+    result = []
+    for index in range(len(items)):
+        if not items[index].socket == ffi.NULL:
+            if items[index].revents > 0:
+                result.append((low_level_to_socket_obj[items[index].socket][0],
+                            items[index].revents))
+        else:
+            result.append((items[index].fd, items[index].revents))
+    return result
+
+__all__ = ['zmq_poll']
diff --git a/src/console/zmq/backend/cffi/_verify.c b/src/console/zmq/backend/cffi/_verify.c
new file mode 100644
index 00000000..547840eb
--- /dev/null
+++ b/src/console/zmq/backend/cffi/_verify.c
@@ -0,0 +1,12 @@
+#include <stdio.h>
+#include <sys/un.h>
+#include <string.h>
+
+#include <zmq.h>
+#include <zmq_utils.h>
+#include "zmq_compat.h"
+
+int get_ipc_path_max_len(void) {
+    struct sockaddr_un *dummy;
+    return sizeof(dummy->sun_path) - 1;
+}
diff --git a/src/console/zmq/backend/cffi/constants.py b/src/console/zmq/backend/cffi/constants.py
new file mode 100755
index 00000000..ee293e74
--- /dev/null
+++ b/src/console/zmq/backend/cffi/constants.py
@@ -0,0 +1,15 @@
+# coding: utf-8
+"""zmq constants"""
+
+from ._cffi import C, c_constant_names
+from zmq.utils.constant_names import all_names
+
+g = globals()
+for cname in c_constant_names:
+    if cname.startswith("ZMQ_"):
+        name = cname[4:]
+    else:
+        name = cname
+    g[name] = getattr(C, cname)
+
+__all__ = all_names
diff --git a/src/console/zmq/backend/cffi/context.py b/src/console/zmq/backend/cffi/context.py
new file mode 100755
index 00000000..16a7b257
--- /dev/null
+++ b/src/console/zmq/backend/cffi/context.py
@@ -0,0 +1,100 @@
+# coding: utf-8
+"""zmq Context class"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+import weakref
+
+from ._cffi import C, ffi
+
+from .socket import *
+from .constants import *
+
+from zmq.error import ZMQError, _check_rc
+
+class Context(object):
+    _zmq_ctx = None
+    _iothreads = None
+    _closed = None
+    _sockets = None
+    _shadow = False
+
+    def __init__(self, io_threads=1, shadow=None):
+        
+        if shadow:
+            self._zmq_ctx = ffi.cast("void *", shadow)
+            self._shadow = True
+        else:
+            self._shadow = False
+            if not io_threads >= 0:
+                raise ZMQError(EINVAL)
+        
+            self._zmq_ctx = C.zmq_ctx_new()
+        if self._zmq_ctx == ffi.NULL:
+            raise ZMQError(C.zmq_errno())
+        if not shadow:
+            C.zmq_ctx_set(self._zmq_ctx, IO_THREADS, io_threads)
+        self._closed = False
+        self._sockets = set()
+    
+    @property
+    def underlying(self):
+        """The address of the underlying libzmq context"""
+        return int(ffi.cast('size_t', self._zmq_ctx))
+    
+    @property
+    def closed(self):
+        return self._closed
+
+    def _add_socket(self, socket):
+        ref = weakref.ref(socket)
+        self._sockets.add(ref)
+        return ref
+
+    def _rm_socket(self, ref):
+        if ref in self._sockets:
+            self._sockets.remove(ref)
+
+    def set(self, option, value):
+        """set a context option
+        
+        see zmq_ctx_set
+        """
+        rc = C.zmq_ctx_set(self._zmq_ctx, option, value)
+        _check_rc(rc)
+
+    def get(self, option):
+        """get context option
+        
+        see zmq_ctx_get
+        """
+        rc = C.zmq_ctx_get(self._zmq_ctx, option)
+        _check_rc(rc)
+        return rc
+
+    def term(self):
+        if self.closed:
+            return
+
+        C.zmq_ctx_destroy(self._zmq_ctx)
+
+        self._zmq_ctx = None
+        self._closed = True
+
+    def destroy(self, linger=None):
+        if self.closed:
+            return
+
+        sockets = self._sockets
+        self._sockets = set()
+        for s in sockets:
+            s = s()
+            if s and not s.closed:
+                if linger:
+                    s.setsockopt(LINGER, linger)
+                s.close()
+        
+        self.term()
+
+__all__ = ['Context']
diff --git a/src/console/zmq/backend/cffi/devices.py b/src/console/zmq/backend/cffi/devices.py
new file mode 100755
index 00000000..c7a514a8
--- /dev/null
+++ b/src/console/zmq/backend/cffi/devices.py
@@ -0,0 +1,24 @@
+# coding: utf-8
+"""zmq device functions"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+from ._cffi import C, ffi, zmq_version_info
+from .socket import Socket
+from zmq.error import ZMQError, _check_rc
+
+def device(device_type, frontend, backend):
+    rc = C.zmq_proxy(frontend._zmq_socket, backend._zmq_socket, ffi.NULL)
+    _check_rc(rc)
+
+def proxy(frontend, backend, capture=None):
+    if isinstance(capture, Socket):
+        capture = capture._zmq_socket
+    else:
+        capture = ffi.NULL
+
+    rc = C.zmq_proxy(frontend._zmq_socket, backend._zmq_socket, capture)
+    _check_rc(rc)
+
+__all__ = ['device', 'proxy']
diff --git a/src/console/zmq/backend/cffi/error.py b/src/console/zmq/backend/cffi/error.py
new file mode 100755
index 00000000..3bb64de0
--- /dev/null
+++ b/src/console/zmq/backend/cffi/error.py
@@ -0,0 +1,13 @@
+"""zmq error functions"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+from ._cffi import C, ffi
+
+def strerror(errno):
+    return ffi.string(C.zmq_strerror(errno))
+
+zmq_errno = C.zmq_errno
+
+__all__ = ['strerror', 'zmq_errno']
diff --git a/src/console/zmq/backend/cffi/message.py b/src/console/zmq/backend/cffi/message.py
new file mode 100755
index 00000000..c35decb6
--- /dev/null
+++ b/src/console/zmq/backend/cffi/message.py
@@ -0,0 +1,69 @@
+"""Dummy Frame object"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+from ._cffi import ffi, C
+
+import zmq
+from zmq.utils.strtypes import unicode
+
+try:
+    view = memoryview
+except NameError:
+    view = buffer
+
+_content = lambda x: x.tobytes() if type(x) == memoryview else x
+
+class Frame(object):
+    _data = None
+    tracker = None
+    closed = False
+    more = False
+    buffer = None
+
+
+    def __init__(self, data, track=False):
+        try:
+            view(data)
+        except TypeError:
+            raise
+
+        self._data = data
+
+        if isinstance(data, unicode):
+            raise TypeError("Unicode objects not allowed. Only: str/bytes, " +
+                            "buffer interfaces.")
+
+        self.more = False
+        self.tracker = None
+        self.closed = False
+        if track:
+            self.tracker = zmq.MessageTracker()
+
+        self.buffer = view(self.bytes)
+
+    @property
+    def bytes(self):
+        data = _content(self._data)
+        return data
+
+    def __len__(self):
+        return len(self.bytes)
+
+    def __eq__(self, other):
+        return self.bytes == _content(other)
+
+    def __str__(self):
+        if str is unicode:
+            return self.bytes.decode()
+        else:
+            return self.bytes
+
+    @property
+    def done(self):
+        return True
+
+Message = Frame
+
+__all__ = ['Frame', 'Message']
diff --git a/src/console/zmq/backend/cffi/socket.py b/src/console/zmq/backend/cffi/socket.py
new file mode 100755
index 00000000..3c427739
--- /dev/null
+++ b/src/console/zmq/backend/cffi/socket.py
@@ -0,0 +1,244 @@
+# coding: utf-8
+"""zmq Socket class"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+import random
+import codecs
+
+import errno as errno_mod
+
+from ._cffi import (C, ffi, new_uint64_pointer, new_int64_pointer,
+                    new_int_pointer, new_binary_data, value_uint64_pointer,
+                    value_int64_pointer, value_int_pointer, value_binary_data,
+                    IPC_PATH_MAX_LEN)
+
+from .message import Frame
+from .constants import *
+
+import zmq
+from zmq.error import ZMQError, _check_rc, _check_version
+from zmq.utils.strtypes import unicode
+
+
+def new_pointer_from_opt(option, length=0):
+    from zmq.sugar.constants import (
+        int64_sockopts, bytes_sockopts,
+    )
+    if option in int64_sockopts:
+        return new_int64_pointer()
+    elif option in bytes_sockopts:
+        return new_binary_data(length)
+    else:
+        # default
+        return new_int_pointer()
+
+def value_from_opt_pointer(option, opt_pointer, length=0):
+    from zmq.sugar.constants import (
+        int64_sockopts, bytes_sockopts,
+    )
+    if option in int64_sockopts:
+        return int(opt_pointer[0])
+    elif option in bytes_sockopts:
+        return ffi.buffer(opt_pointer, length)[:]
+    else:
+        return int(opt_pointer[0])
+
+def initialize_opt_pointer(option, value, length=0):
+    from zmq.sugar.constants import (
+        int64_sockopts, bytes_sockopts,
+    )
+    if option in int64_sockopts:
+        return value_int64_pointer(value)
+    elif option in bytes_sockopts:
+        return value_binary_data(value, length)
+    else:
+        return value_int_pointer(value)
+
+
+class Socket(object):
+    context = None
+    socket_type = None
+    _zmq_socket = None
+    _closed = None
+    _ref = None
+    _shadow = False
+
+    def __init__(self, context=None, socket_type=None, shadow=None):
+        self.context = context
+        if shadow is not None:
+            self._zmq_socket = ffi.cast("void *", shadow)
+            self._shadow = True
+        else:
+            self._shadow = False
+            self._zmq_socket = C.zmq_socket(context._zmq_ctx, socket_type)
+        if self._zmq_socket == ffi.NULL:
+            raise ZMQError()
+        self._closed = False
+        if context:
+            self._ref = context._add_socket(self)
+    
+    @property
+    def underlying(self):
+        """The address of the underlying libzmq socket"""
+        return int(ffi.cast('size_t', self._zmq_socket))
+    
+    @property
+    def closed(self):
+        return self._closed
+
+    def close(self, linger=None):
+        rc = 0
+        if not self._closed and hasattr(self, '_zmq_socket'):
+            if self._zmq_socket is not None:
+                rc = C.zmq_close(self._zmq_socket)
+            self._closed = True
+            if self.context:
+                self.context._rm_socket(self._ref)
+        return rc
+
+    def bind(self, address):
+        if isinstance(address, unicode):
+            address = address.encode('utf8')
+        rc = C.zmq_bind(self._zmq_socket, address)
+        if rc < 0:
+            if IPC_PATH_MAX_LEN and C.zmq_errno() == errno_mod.ENAMETOOLONG:
+                # py3compat: address is bytes, but msg wants str
+                if str is unicode:
+                    address = address.decode('utf-8', 'replace')
+                path = address.split('://', 1)[-1]
+                msg = ('ipc path "{0}" is longer than {1} '
+                                'characters (sizeof(sockaddr_un.sun_path)).'
+                                .format(path, IPC_PATH_MAX_LEN))
+                raise ZMQError(C.zmq_errno(), msg=msg)
+            else:
+                _check_rc(rc)
+
+    def unbind(self, address):
+        _check_version((3,2), "unbind")
+        if isinstance(address, unicode):
+            address = address.encode('utf8')
+        rc = C.zmq_unbind(self._zmq_socket, address)
+        _check_rc(rc)
+
+    def connect(self, address):
+        if isinstance(address, unicode):
+            address = address.encode('utf8')
+        rc = C.zmq_connect(self._zmq_socket, address)
+        _check_rc(rc)
+
+    def disconnect(self, address):
+        _check_version((3,2), "disconnect")
+        if isinstance(address, unicode):
+            address = address.encode('utf8')
+        rc = C.zmq_disconnect(self._zmq_socket, address)
+        _check_rc(rc)
+
+    def set(self, option, value):
+        length = None
+        if isinstance(value, unicode):
+            raise TypeError("unicode not allowed, use bytes")
+        
+        if isinstance(value, bytes):
+            if option not in zmq.constants.bytes_sockopts:
+                raise TypeError("not a bytes sockopt: %s" % option)
+            length = len(value)
+        
+        c_data = initialize_opt_pointer(option, value, length)
+
+        c_value_pointer = c_data[0]
+        c_sizet = c_data[1]
+
+        rc = C.zmq_setsockopt(self._zmq_socket,
+                               option,
+                               ffi.cast('void*', c_value_pointer),
+                               c_sizet)
+        _check_rc(rc)
+
+    def get(self, option):
+        c_data = new_pointer_from_opt(option, length=255)
+
+        c_value_pointer = c_data[0]
+        c_sizet_pointer = c_data[1]
+
+        rc = C.zmq_getsockopt(self._zmq_socket,
+                               option,
+                               c_value_pointer,
+                               c_sizet_pointer)
+        _check_rc(rc)
+        
+        sz = c_sizet_pointer[0]
+        v = value_from_opt_pointer(option, c_value_pointer, sz)
+        if option != zmq.IDENTITY and option in zmq.constants.bytes_sockopts and v.endswith(b'\0'):
+            v = v[:-1]
+        return v
+
+    def send(self, message, flags=0, copy=False, track=False):
+        if isinstance(message, unicode):
+            raise TypeError("Message must be in bytes, not an unicode Object")
+
+        if isinstance(message, Frame):
+            message = message.bytes
+
+        zmq_msg = ffi.new('zmq_msg_t*')
+        c_message = ffi.new('char[]', message)
+        rc = C.zmq_msg_init_size(zmq_msg, len(message))
+        C.memcpy(C.zmq_msg_data(zmq_msg), c_message, len(message))
+
+        rc = C.zmq_msg_send(zmq_msg, self._zmq_socket, flags)
+        C.zmq_msg_close(zmq_msg)
+        _check_rc(rc)
+
+        if track:
+            return zmq.MessageTracker()
+
+    def recv(self, flags=0, copy=True, track=False):
+        zmq_msg = ffi.new('zmq_msg_t*')
+        C.zmq_msg_init(zmq_msg)
+
+        rc = C.zmq_msg_recv(zmq_msg, self._zmq_socket, flags)
+
+        if rc < 0:
+            C.zmq_msg_close(zmq_msg)
+            _check_rc(rc)
+
+        _buffer = ffi.buffer(C.zmq_msg_data(zmq_msg), C.zmq_msg_size(zmq_msg))
+        value = _buffer[:]
+        C.zmq_msg_close(zmq_msg)
+
+        frame = Frame(value, track=track)
+        frame.more = self.getsockopt(RCVMORE)
+
+        if copy:
+            return frame.bytes
+        else:
+            return frame
+    
+    def monitor(self, addr, events=-1):
+        """s.monitor(addr, flags)
+
+        Start publishing socket events on inproc.
+        See libzmq docs for zmq_monitor for details.
+        
+        Note: requires libzmq >= 3.2
+        
+        Parameters
+        ----------
+        addr : str
+            The inproc url used for monitoring. Passing None as
+            the addr will cause an existing socket monitor to be
+            deregistered.
+        events : int [default: zmq.EVENT_ALL]
+            The zmq event bitmask for which events will be sent to the monitor.
+        """
+        
+        _check_version((3,2), "monitor")
+        if events < 0:
+            events = zmq.EVENT_ALL
+        if addr is None:
+            addr = ffi.NULL
+        rc = C.zmq_socket_monitor(self._zmq_socket, addr, events)
+
+
+__all__ = ['Socket', 'IPC_PATH_MAX_LEN']
diff --git a/src/console/zmq/backend/cffi/utils.py b/src/console/zmq/backend/cffi/utils.py
new file mode 100755
index 00000000..fde7827b
--- /dev/null
+++ b/src/console/zmq/backend/cffi/utils.py
@@ -0,0 +1,62 @@
+# coding: utf-8
+"""miscellaneous zmq_utils wrapping"""
+
+# Copyright (C) PyZMQ Developers
+# Distributed under the terms of the Modified BSD License.
+
+from ._cffi import ffi, C
+
+from zmq.error import ZMQError, _check_rc, _check_version
+from zmq.utils.strtypes import unicode
+
+def has(capability):
+    """Check for zmq capability by name (e.g. 'ipc', 'curve')
+    
+    .. versionadded:: libzmq-4.1
+    .. versionadded:: 14.1
+    """
+    _check_version((4,1), 'zmq.has')
+    if isinstance(capability, unicode):
+        capability = capability.encode('utf8')
+    return bool(C.zmq_has(capability))
+    
+def curve_keypair():
+    """generate a Z85 keypair for use with zmq.CURVE security
+    
+    Requires libzmq (≥ 4.0) to have been linked with libsodium.
+    
+    Returns
+    -------
+    (public, secret) : two bytestrings
+        The public and private keypair as 40 byte z85-encoded bytestrings.
+    """
+    _check_version((3,2), "monitor")
+    public = ffi.new('char[64]')
+    private = ffi.new('char[64]')
+    rc = C.zmq_curve_keypair(public, private)
+    _check_rc(rc)
+    return ffi.buffer(public)[:40], ffi.buffer(private)[:40]
+
+
+class Stopwatch(object):
+    def __init__(self):
+        self.watch = ffi.NULL
+
+    def start(self):
+        if self.watch == ffi.NULL:
+            self.watch = C.zmq_stopwatch_start()
+        else:
+            raise ZMQError('Stopwatch is already runing.')
+
+    def stop(self):
+        if self.watch == ffi.NULL:
+            raise ZMQError('Must start the Stopwatch before calling stop.')
+        else:
+            time = C.zmq_stopwatch_stop(self.watch)
+            self.watch = ffi.NULL
+            return time
+
+    def sleep(self, seconds):
+        C.zmq_sleep(seconds)
+
+__all__ = ['has', 'curve_keypair', 'Stopwatch']