Core API¶

create_zmq_connection¶

aiozmq.create_zmq_connection(protocol_factory, zmq_type, *, bind=None, connect=None, zmq_sock=None, loop=None)[source]¶

Create a ZeroMQ connection.

This method is a coroutine.

If you don’t use bind or connect params you can do it later by ZmqTransport.bind() and ZmqTransport.connect() calls.

Parameters:

protocol_factory (callable) – a factory that instantiates ZmqProtocol object.
zmq_type (int) – a type of ZeroMQ socket (zmq.REQ, zmq.REP, zmq.PUB, zmq.SUB, zmq.PAIR*, zmq.DEALER, zmq.ROUTER, zmq.PULL, zmq.PUSH, etc.)
bind (str or iterable of strings) –
endpoints specification.

Every endpoint generates call to ZmqTransport.bind() for accepting connections from specified endpoint.

Other side should use connect parameter to connect to this transport.
connect (str or iterable of strings) –
endpoints specification.

Every endpoint generates call to ZmqTransport.connect() for connecting transport to specified endpoint.

Other side should use bind parameter to wait for incoming connections.
zmq_sock (zmq.Socket) – a preexisting zmq socket that will be passed to returned transport.
loop (asyncio.AbstractEventLoop) – optional event loop instance, None for default event loop.

Returns:

a pair of (transport, protocol) where transport supports ZmqTransport interface.

Return type:

tuple

New in version 0.5.

ZmqTransport¶

class aiozmq.ZmqTransport[source]¶

Transport for ZeroMQ connections. Implements asyncio.BaseTransport interface.

End user should never create ZmqTransport objects directly, he gets it by await aiozmq.create_zmq_connection() call.

get_extra_info(key, default=None)¶

Return optional transport information if name is present otherwise return default.

ZmqTransport supports the only valid key: "zmq_socket". The value is zmq.Socket instance.

Parameters:

name (str) – name of info record.
default – default value

close()¶

Close the transport.

Buffered data will be flushed asynchronously. No more data will be received. After all buffered data is flushed, the protocol’s connection_lost() method will (eventually) called with None as its argument.

write(data)[source]¶

Write message to the transport.

Parameters:: data – iterable to send as multipart message.

This does not block; it buffers the data and arranges for it to be sent out asynchronously.

abort()[source]¶

Close the transport immediately.

Buffered data will be lost. No more data will be received. The protocol’s connection_lost() method will (eventually) be called with None as it’s argument.

getsockopt(option)[source]¶

Get ZeroMQ socket option.

Parameters:

option (int) –

a constant like zmq.SUBSCRIBE, zmq.UNSUBSCRIBE, zmq.TYPE etc.

For list of available options please see: http://api.zeromq.org/master:zmq-getsockopt

Returns:

option value

Raises:

OSError – if call to ZeroMQ was unsuccessful.

setsockopt(option, value)[source]¶

Set ZeroMQ socket option.

param int option:

a constant like zmq.SUBSCRIBE, zmq.UNSUBSCRIBE, zmq.TYPE etc.

param value:

a new option value, it’s type depend of option name.

For list of available options please see: http://api.zeromq.org/master:zmq-setsockopt

get_write_buffer_limits()[source]¶

Get the high- and low-water limits for write flow control. Return a tuple (low, high) where low and high are positive number of bytes.

Use set_write_buffer_limits() to set the limits.

New in version 0.6.

set_write_buffer_limits(high=None, low=None)[source]¶

Set the high- and low-water limits for write flow control.

Parameters:

high (int or None) – high-water limit
low (int or None) – low-water limit

These two values control when to call the protocol’s pause_writing() and resume_writing() methods. If specified, the low-water limit must be less than or equal to the high-water limit. Neither value can be negative.

The defaults are implementation-specific. If only the high-water limit is given, the low-water limit defaults to a implementation-specific value less than or equal to the high-water limit. Setting high to zero forces low to zero as well, and causes pause_writing() to be called whenever the buffer becomes non-empty. Setting low to zero causes resume_writing() to be called only once the buffer is empty. Use of zero for either limit is generally sub-optimal as it reduces opportunities for doing I/O and computation concurrently.

Use get_write_buffer_limits() to get the limits.

get_write_buffer_size()[source]¶: Return the current size of the write buffer.

pause_reading()[source]¶

Pause the receiving end.

No data will be passed to the protocol’s ZmqProtocol.msg_received() method until ZmqTransport.resume_reading() is called.

ZmqProtocol¶

class aiozmq.ZmqProtocol[source]¶

Protocol for ZeroMQ connections. Derives from asyncio.BaseProtocol.

connection_made(transport)¶

Called when a connection is made.

Parameters:: transport (ZmqTransport) – representing the pipe connection. To receive data, wait for msg_received() calls. When the connection is closed, connection_lost() is called.

connection_lost(exc)¶

Called when the connection is lost or closed.

Parameters:: exc (instance of Exception or derived class) – an exception object or None (the latter meaning the connection was aborted or closed).

pause_writing()¶

Called when the transport’s buffer goes over the high-water mark.

Pause and resume calls are paired – pause_writing() is called once when the buffer goes strictly over the high-water mark (even if subsequent writes increases the buffer size even more), and eventually resume_writing() is called once when the buffer size reaches the low-water mark.

Note that if the buffer size equals the high-water mark, pause_writing() is not called – it must go strictly over. Conversely, resume_writing() is called when the buffer size is equal or lower than the low-water mark. These end conditions are important to ensure that things go as expected when either mark is zero.

Note

This is the only Protocol callback that is not called through asyncio.AbstractEventLoop.call_soon() – if it were, it would have no effect when it’s most needed (when the app keeps writing without awaiting until pause_writing() is called).

resume_writing()¶

Called when the transport’s buffer drains below the low-water mark.

See pause_writing() for details.

msg_received(data)[source]¶

Called when some ZeroMQ message is received.

Parameters:: data (list) – the multipart list of bytes with at least one item.

event_received(event)[source]¶

Called when a ZeroMQ socket event is received.

This method is only called when a socket monitor is enabled.

Parameters:: event (namedtuple) – a SocketEvent namedtuple containing 3 items: event, value, and endpoint.

New in version 0.7.

Exception policy¶

Every call to zmq.Socket method can raise zmq.ZMQError exception. But all methods of ZmqEventLoop and ZmqTransport translate ZMQError into OSError (or descendat) with errno and strerror borrowed from underlying ZMQError values.

The reason for translation is that Python 3.3 implements PEP 3151 — Reworking the OS and IO Exception Hierarchy which gets rid of exceptions zoo and uses OSError and descendants for all exceptions generated by system function calls.

aiozmq implements the same pattern. Internally it looks like:

try:
    return self._zmq_sock.getsockopt(option)
except zmq.ZMQError as exc:
    raise OSError(exc.errno, exc.strerror)

Also public methods of aiozmq will never raise InterruptedError (aka EINTR), they process interruption internally.

Getting aiozmq version¶

aiozmq.version¶

a text version of the library:

'0.1.0 , Python 3.3.2+ (default, Feb 28 2014, 00:52:16) \n[GCC 4.8.1]'

aiozmq.version_info¶

a named tuple with version information, useful for comparison:

VersionInfo(major=0, minor=1, micro=0, releaselevel='alpha', serial=0)

The Python itself uses the same schema (sys.version_info).

Installing ZeroMQ event loop¶

Deprecated since version 0.5: aiozmq works with any asyncio event loop, it doesn’t require dedicated event loop policy.

To use ZeroMQ layer you may install proper event loop first.

The recommended way is to setup global event loop policy:

import asyncio
import aiozmq

asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())

That installs ZmqEventLoopPolicy globally. After installing you can get event loop instance from main thread by asyncio.get_event_loop() call:

loop = asyncio.get_event_loop()

If you need to execute event loop in your own (not main) thread you have to set it up first:

import threading

def thread_func():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop()

    loop.run_forever()

thread = threading.Thread(target=thread_func)
thread.start()

ZmqEventLoopPolicy¶

Deprecated since version 0.5: aiozmq works with any asyncio event loop, it doesn’t require dedicated event loop policy.

ZeroMQ policy implementation for accessing the event loop.

In this policy, each thread has its own event loop. However, we only automatically create an event loop by default for the main thread; other threads by default have no event loop.

ZmqEventLoopPolicy implements an asyncio.AbstractEventLoopPolicy interface.

class aiozmq.ZmqEventLoopPolicy[source]¶

Create policy for ZeroMQ event loops.

Note

policy should be installed, see Installing ZeroMQ event loop.

get_event_loop()[source]¶

Get the event loop.

If current thread is the main thread and there are no registered event loop for current thread then the call creates new event loop and registers it.

Returns:: Return an instance of ZmqEventLoop.
Raises:: RuntimeError – if there is no registered event loop for current thread.

new_event_loop()[source]¶

Create a new event loop.

You must call ZmqEventLoopPolicy.set_event_loop() to make this the current event loop.

set_event_loop(loop)[source]¶

Set the event loop.

As a side effect, if a child watcher was set before, then calling .set_event_loop() from the main thread will call asyncio.AbstractChildWatcher.attach_loop() on the child watcher.

Parameters:: loop – an asyncio.AbstractEventLoop instance or None
Raises:: TypeError – if loop is not instance of asyncio.AbstractEventLoop

get_child_watcher()[source]¶

Get the child watcher

If not yet set, a asyncio.SafeChildWatcher object is automatically created.

Returns:: Return an instance of asyncio.AbstractChildWatcher.

set_child_watcher(watcher)[source]¶

Set the child watcher.

Parameters:: watcher – an asyncio.AbstractChildWatcher instance or None
Raises:: TypeError – if watcher is not instance of asyncio.AbstractChildWatcher

ZmqEventLoop¶

Deprecated since version 0.5: aiozmq works with any asyncio event loop, it doesn’t require dedicated event loop object.

Event loop with ZeroMQ support.

Follows asyncio.AbstractEventLoop specification and has create_zmq_connection() method for ZeroMQ sockets layer.

class aiozmq.ZmqEventLoop(*, zmq_context=None)[source]¶

Parameters:: zmq_context (zmq.Context) – explicit context to use for ZeroMQ socket creation inside ZmqEventLoop.create_zmq_connection() calls. aiozmq shares global context returned by zmq.Context.instance() call if zmq_context parameter is None.

create_zmq_connection(protocol_factory, zmq_type, *, bind=None, connect=None, zmq_sock=None)[source]¶

Create a ZeroMQ connection.

If you don’t use bind or connect params you can do it later by ZmqTransport.bind() and ZmqTransport.connect() calls.

Parameters:

protocol_factory (callable) – a factory that instantiates ZmqProtocol object.
zmq_type (int) – a type of ZeroMQ socket (zmq.REQ, zmq.REP, zmq.PUB, zmq.SUB, zmq.PAIR*, zmq.DEALER, zmq.ROUTER, zmq.PULL, zmq.PUSH, etc.)
bind (str or iterable of strings) –
endpoints specification.

Every endpoint generates call to ZmqTransport.bind() for accepting connections from specified endpoint.

Other side should use connect parameter to connect to this transport.
connect (str or iterable of strings) –
endpoints specification.

Every endpoint generates call to ZmqTransport.connect() for connecting transport to specified endpoint.

Other side should use bind parameter to wait for incoming connections.
zmq_sock (zmq.Socket) – a preexisting zmq socket that will be passed to returned transport.

Returns:

a pair of (transport, protocol) where transport supports ZmqTransport interface.

Return type:

tuple

Core API¶

create_zmq_connection¶

ZmqTransport¶

ZmqProtocol¶

Exception policy¶

Getting aiozmq version¶

Installing ZeroMQ event loop¶

ZmqEventLoopPolicy¶

ZmqEventLoop¶

Table of Contents

Previous topic

Next topic

This Page