summaryrefslogtreecommitdiffstats
path: root/src/console/zmq/backend/cffi/socket.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/console/zmq/backend/cffi/socket.py')
-rwxr-xr-xsrc/console/zmq/backend/cffi/socket.py244
1 files changed, 0 insertions, 244 deletions
diff --git a/src/console/zmq/backend/cffi/socket.py b/src/console/zmq/backend/cffi/socket.py
deleted file mode 100755
index 3c427739..00000000
--- a/src/console/zmq/backend/cffi/socket.py
+++ /dev/null
@@ -1,244 +0,0 @@
-# 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']