FM / cffi

/opt/imh-python/lib/python3.9/site-packages/zmq/backend/cffi/

/opt/imh-python/lib/python3.9/site-packages/zmq/backend/cffi

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
constants.py	357	0644	edit dl rm
context.py	1916	0644	edit dl rm
devices.py	1644	0644	edit dl rm
error.py	353	0644	edit dl rm
message.py	1388	0644	edit dl rm
socket.py	9126	0644	edit dl rm
utils.py	2167	0644	edit dl rm
_cdefs.h	2216	0644	edit dl rm
_cffi.py	3882	0644	edit dl rm
_poll.py	2832	0644	edit dl rm
_verify.c	105	0644	edit dl rm
__init__.py	616	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/zmq/backend/cffi/devices.py (1644B)

# coding: utf-8
"""zmq device functions"""

# Copyright (C) PyZMQ Developers
# Distributed under the terms of the Modified BSD License.

from ._cffi import C, ffi
from .socket import Socket
from .utils import _retry_sys_call

def device(device_type, frontend, backend):
    return proxy(frontend, backend)

def proxy(frontend, backend, capture=None):
    if isinstance(capture, Socket):
        capture = capture._zmq_socket
    else:
        capture = ffi.NULL

_retry_sys_call(
            C.zmq_proxy,
            frontend._zmq_socket,
            backend._zmq_socket,
            capture
        )

def proxy_steerable(frontend, backend, capture=None, control=None):
    """proxy_steerable(frontend, backend, capture, control)

Start a zeromq proxy with control flow.

.. versionadded:: libzmq-4.1
    .. versionadded:: 18.0

Parameters
    ----------
    frontend : Socket
        The Socket instance for the incoming traffic.
    backend : Socket
        The Socket instance for the outbound traffic.
    capture : Socket (optional)
        The Socket instance for capturing traffic.
    control : Socket (optional)
        The Socket instance for control flow.
    """
    if isinstance(capture, Socket):
        capture = capture._zmq_socket
    else:
        capture = ffi.NULL

if isinstance(control, Socket):
        control = control._zmq_socket
    else:
        control = ffi.NULL

_retry_sys_call(
            C.zmq_proxy_steerable,
            frontend._zmq_socket,
            backend._zmq_socket,
            capture,
            control
        )

__all__ = ['device', 'proxy', 'proxy_steerable']

# coding: utf-8 """zmq device functions""" # Copyright (C) PyZMQ Developers # Distributed under the terms of the Modified BSD License. from ._cffi import C, ffi from .socket import Socket from .utils import _retry_sys_call def device(device_type, frontend, backend): return proxy(frontend, backend) def proxy(frontend, backend, capture=None): if isinstance(capture, Socket): capture = capture._zmq_socket else: capture = ffi.NULL _retry_sys_call( C.zmq_proxy, frontend._zmq_socket, backend._zmq_socket, capture ) def proxy_steerable(frontend, backend, capture=None, control=None): """proxy_steerable(frontend, backend, capture, control) Start a zeromq proxy with control flow. .. versionadded:: libzmq-4.1 .. versionadded:: 18.0 Parameters ---------- frontend : Socket The Socket instance for the incoming traffic. backend : Socket The Socket instance for the outbound traffic. capture : Socket (optional) The Socket instance for capturing traffic. control : Socket (optional) The Socket instance for control flow. """ if isinstance(capture, Socket): capture = capture._zmq_socket else: capture = ffi.NULL if isinstance(control, Socket): control = control._zmq_socket else: control = ffi.NULL _retry_sys_call( C.zmq_proxy_steerable, frontend._zmq_socket, backend._zmq_socket, capture, control ) __all__ = ['device', 'proxy', 'proxy_steerable']