devices#

Functions#

zmq.device(device_type: C.int, frontend: Socket, backend: Socket = None)

Start a zeromq device.

Deprecated since version libzmq-3.2: Use zmq.proxy

Parameters:

  • device_type ((QUEUE, FORWARDER, STREAMER)) – The type of device to start.

  • frontend (Socket) – The Socket instance for the incoming traffic.

  • backend (Socket) – The Socket instance for the outbound traffic.

zmq.proxy(frontend: Socket, backend: Socket, capture: Socket = None)

Start a zeromq proxy (replacement for device).

New in version libzmq-3.2.

New in version 13.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.

zmq.proxy_steerable(frontend: Socket, backend: Socket, capture: Socket = None, control: Socket = None)

Start a zeromq proxy with control flow.

New in version libzmq-4.1.

New in version 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.

Module: zmq.devices#

0MQ Device classes for running in background threads or processes.

Base Devices#

Device#

class zmq.devices.Device(device_type: int = DeviceType.QUEUE, in_type: int | None = None, out_type: int | None = None)#

A 0MQ Device to be run in the background.

You do not pass Socket instances to this, but rather Socket types:

Device(device_type, in_socket_type, out_socket_type)

For instance:

dev = Device(zmq.QUEUE, zmq.DEALER, zmq.ROUTER)

Similar to zmq.device, but socket types instead of sockets themselves are passed, and the sockets are created in the work thread, to avoid issues with thread safety. As a result, additional bind_{in|out} and connect_{in|out} methods and setsockopt_{in|out} allow users to specify connections for the sockets.

Parameters:

  • device_type (int) – The 0MQ Device type

  • {in|out}_type (int) – zmq socket types, to be passed later to context.socket(). e.g. zmq.PUB, zmq.SUB, zmq.REQ. If out_type is < 0, then in_socket is used for both in_socket and out_socket.

bind_{in_out}(iface)

passthrough for {in|out}_socket.bind(iface), to be called in the thread

connect_{in_out}(iface)

passthrough for {in|out}_socket.connect(iface), to be called in the thread

setsockopt_{in_out}(opt,value)

passthrough for {in|out}_socket.setsockopt(opt, value), to be called in the thread

daemon#

sets whether the thread should be run as a daemon Default is true, because if it is false, the thread will not exit unless it is killed

Type:

bool

context_factory#

Function for creating the Context. This will be Context.instance in ThreadDevices, and Context in ProcessDevices. The only reason it is not instance() in ProcessDevices is that there may be a stale Context instance already initialized, and the forked environment should never try to use it.

Type:

callable (class attribute)

bind_in(addr: str) → None#

Enqueue ZMQ address for binding on in_socket.

See zmq.Socket.bind for details.

bind_in_to_random_port(addr: str, *args, **kwargs) → int#

Enqueue a random port on the given interface for binding on in_socket.

See zmq.Socket.bind_to_random_port for details.

New in version 18.0.

bind_out(addr: str) → None#

Enqueue ZMQ address for binding on out_socket.

See zmq.Socket.bind for details.

bind_out_to_random_port(addr: str, *args, **kwargs) → int#

Enqueue a random port on the given interface for binding on out_socket.

See zmq.Socket.bind_to_random_port for details.

New in version 18.0.

connect_in(addr: str) → None#

Enqueue ZMQ address for connecting on in_socket.

See zmq.Socket.connect for details.

connect_out(addr: str)#

Enqueue ZMQ address for connecting on out_socket.

See zmq.Socket.connect for details.

join(timeout: float | None = None) → None#

wait for me to finish, like Thread.join.

Reimplemented appropriately by subclasses.

setsockopt_in(opt: int, value: Any) → None#

Enqueue setsockopt(opt, value) for in_socket

See zmq.Socket.setsockopt for details.

setsockopt_out(opt: int, value: Any)#

Enqueue setsockopt(opt, value) for out_socket

See zmq.Socket.setsockopt for details.

start() → None#

Start the device. Override me in subclass for other launchers.

ThreadDevice#

class zmq.devices.ThreadDevice(device_type: int = DeviceType.QUEUE, in_type: int | None = None, out_type: int | None = None)#

A Device that will be run in a background Thread.

See Device for details.

ProcessDevice#

class zmq.devices.ProcessDevice(device_type: int = DeviceType.QUEUE, in_type: int | None = None, out_type: int | None = None)#

A Device that will be run in a background Process.

See Device for details.

context_factory#

alias of Context

Proxy Devices#

Proxy#

class zmq.devices.Proxy(in_type, out_type, mon_type=SocketType.PUB)#

Threadsafe Proxy object.

See zmq.devices.Device for most of the spec. This subclass adds a <method>_mon version of each <method>_{in|out} method, for configuring the monitor socket.

A Proxy is a 3-socket ZMQ Device that functions just like a QUEUE, except each message is also sent out on the monitor socket.

A PUB socket is the most logical choice for the mon_socket, but it is not required.

bind_mon(addr)#

Enqueue ZMQ address for binding on mon_socket.

See zmq.Socket.bind for details.

connect_mon(addr)#

Enqueue ZMQ address for connecting on mon_socket.

See zmq.Socket.connect for details.

setsockopt_mon(opt, value)#

Enqueue setsockopt(opt, value) for mon_socket

See zmq.Socket.setsockopt for details.

ThreadProxy#

class zmq.devices.ThreadProxy(in_type, out_type, mon_type=SocketType.PUB)#

Proxy in a Thread. See Proxy for more.

ProcessProxy#

class zmq.devices.ProcessProxy(in_type, out_type, mon_type=SocketType.PUB)#

Proxy in a Process. See Proxy for more.

ProxySteerable#

class zmq.devices.ProxySteerable(in_type, out_type, mon_type=SocketType.PUB, ctrl_type=None)#

Class for running a steerable proxy in the background.

See zmq.devices.Proxy for most of the spec. If the control socket is not NULL, the proxy supports control flow, provided by the socket.

If PAUSE is received on this socket, the proxy suspends its activities. If RESUME is received, it goes on. If TERMINATE is received, it terminates smoothly. If the control socket is NULL, the proxy behave exactly as if zmq.devices.Proxy had been used.

This subclass adds a <method>_ctrl version of each <method>_{in|out} method, for configuring the control socket.

New in version libzmq-4.1.

New in version 18.0.

bind_ctrl(addr)#

Enqueue ZMQ address for binding on ctrl_socket.

See zmq.Socket.bind for details.

connect_ctrl(addr)#

Enqueue ZMQ address for connecting on ctrl_socket.

See zmq.Socket.connect for details.

setsockopt_ctrl(opt, value)#

Enqueue setsockopt(opt, value) for ctrl_socket

See zmq.Socket.setsockopt for details.

ThreadProxySteerable#

class zmq.devices.ThreadProxySteerable(in_type, out_type, mon_type=SocketType.PUB, ctrl_type=None)#

ProxySteerable in a Thread. See ProxySteerable for details.

ProcessProxySteerable#

class zmq.devices.ProcessProxySteerable(in_type, out_type, mon_type=SocketType.PUB, ctrl_type=None)#

ProxySteerable in a Process. See ProxySteerable for details.

MonitoredQueue Devices#

zmq.devices.monitored_queue(in_socket: Socket, out_socket: Socket, mon_socket: Socket, in_prefix: bytes = b'in', out_prefix: bytes = b'out')#

Start a monitored queue device.

A monitored queue is very similar to the zmq.proxy device (monitored queue came first).

Differences from zmq.proxy:

  • monitored_queue supports both in and out being ROUTER sockets (via swapping IDENTITY prefixes).

  • monitor messages are prefixed, making in and out messages distinguishable.

Parameters:

  • in_socket (Socket) – One of the sockets to the Queue. Its messages will be prefixed with ‘in’.

  • out_socket (Socket) – One of the sockets to the Queue. Its messages will be prefixed with ‘out’. The only difference between in/out socket is this prefix.

  • mon_socket (Socket) – This socket sends out every message received by each of the others with an in/out prefix specifying which one it was.

  • in_prefix (str) – Prefix added to broadcast messages from in_socket.

  • out_prefix (str) – Prefix added to broadcast messages from out_socket.

MonitoredQueue#

class zmq.devices.MonitoredQueue(in_type, out_type, mon_type=SocketType.PUB, in_prefix=b'in', out_prefix=b'out')#

Class for running monitored_queue in the background.

See zmq.devices.Device for most of the spec. MonitoredQueue differs from Proxy, only in that it adds a prefix to messages sent on the monitor socket, with a different prefix for each direction.

MQ also supports ROUTER on both sides, which zmq.proxy does not.

If a message arrives on in_sock, it will be prefixed with in_prefix on the monitor socket. If it arrives on out_sock, it will be prefixed with out_prefix.

A PUB socket is the most logical choice for the mon_socket, but it is not required.

ThreadMonitoredQueue#

class zmq.devices.ThreadMonitoredQueue(in_type, out_type, mon_type=SocketType.PUB, in_prefix=b'in', out_prefix=b'out')#

Run zmq.monitored_queue in a background thread.

See MonitoredQueue and Proxy for details.

ProcessMonitoredQueue#

class zmq.devices.ProcessMonitoredQueue(in_type, out_type, mon_type=SocketType.PUB, in_prefix=b'in', out_prefix=b'out')#

Run zmq.monitored_queue in a separate process.

See MonitoredQueue and Proxy for details.