kiwipy.rmq.threadcomms module

class kiwipy.rmq.threadcomms.RmqThreadCommunicator(connection_params: str | dict | None = None, connection_factory=<function connect_robust>, message_exchange: str = 'kiwipy.messages', queue_expires: int = 60000, task_exchange='kiwipy.tasks', task_queue: str = 'kiwipy.tasks', task_prefetch_size=0, task_prefetch_count=0, encoder=functools.partial(<function dump>, encoding='utf-8'), decoder=functools.partial(<function load>, Loader=<class 'yaml.loader.FullLoader'>), testing_mode=False, async_task_timeout=5.0)

Bases: Communicator

RabbitMQ communicator that runs an event loop on a separate thread to do communication. This also means that heartbeats are not missed and the main program is free to block for as long as it wants.

TASK_TIMEOUT = 5.0

_communicator: RmqCommunicator

_ensure_open()

_wrap_future(kiwi_future: Future)

_wrap_subscriber(subscriber)

” We need to convert any kiwipy.Futures we get from a subscriber call into asyncio ones for the event loop based communicator. Do this by wrapping any subscriber methods and intercepting the return values.

add_broadcast_subscriber(subscriber, identifier=None)

Add a broadcast subscriber that will receive all broadcast messages

Parameters:

• subscriber – the subscriber function to be called

• identifier – an optional identifier for the subscriber

Returns:

an identifier for the subscriber and can be subsequently used to remove it

add_close_callback(callback: Callable[[AbstractConnection, BaseException | None], Any], weak: bool = False) → None

Add a callable to be called each time (after) the connection is closed.

Parameters:

weak – If True, the callback will be added to a WeakSet

add_rpc_subscriber(subscriber, identifier=None)

Add an RPC subscriber to the communicator with an optional identifier. If an identifier is not provided the communicator will generate a unique one. In all cases the identifier will be returned.

add_task_subscriber(subscriber, identifier=None)

Add a task subscriber to the communicator’s default queue. Returns the identifier.

Parameters:

• subscriber – The task callback function

• identifier – the subscriber identifier

broadcast_send(body, sender=None, subject=None, correlation_id=None)

Broadcast a message to all subscribers

close()

Close down the communicator and clean up all resources.

After this call it cannot be used again.

classmethod connect(connection_params: str | dict | None = None, connection_factory=<function connect_robust>, message_exchange='kiwipy.messages', task_exchange='kiwipy.tasks', task_queue='kiwipy.tasks', task_prefetch_size=0, task_prefetch_count=0, encoder=functools.partial(<function dump>, encoding='utf-8'), decoder=functools.partial(<function load>, Loader=<class 'yaml.loader.FullLoader'>), testing_mode=False, async_task_timeout=5.0)

is_closed() → bool

Return True if the communicator was closed

loop()

remove_broadcast_subscriber(identifier)

Remove a broadcast subscriber

Parameters:

identifier – the identifier of the subscriber to remove

remove_rpc_subscriber(identifier)

Remove an RPC subscriber given the identifier. Raises a ValueError if there is no such subscriber.

Parameters:

identifier – The RPC subscriber identifier

remove_task_subscriber(identifier)

Remove a task subscriber from the communicator’s default queue.

Parameters:

identifier – the subscriber to remove

Raises:

ValueError if identifier does not correspond to a known subscriber

rpc_send(recipient_id, msg)

Initiate a remote procedure call on a recipient. This method returns a future representing the outcome of the call.

Parameters:

• recipient_id – The recipient identifier

• msg – The body of the message

Returns:

A future corresponding to the outcome of the call

Return type:

kiwipy.Future

property server_properties: Dict

A dictionary containing server properties as returned by the RMQ server at connection time.

The details are defined by the RMQ standard and can be found here:

https://www.rabbitmq.com/amqp-0-9-1-reference.html#connection.start.server-properties

The protocol states that this dictionary SHOULD contain at least:

‘host’ - specifying the server host name or address ‘product’ - giving the name of the server product ‘version’ - giving the name of the server version ‘platform’ - giving the name of the operating system ‘copyright’ - if appropriate, and, ‘information’ - giving other general information

Note

In testing it seems like ‘host’ is not always returned. Host information may be found in ‘cluster_name’ but clients shouldn’t rely on this.

Returns:

the server properties dictionary

task_queue(queue_name: str, prefetch_size=0, prefetch_count=0)

task_send(task, no_reply=False)

Send a task messages, this will be queued and picked up by a worker at some point in the future. The method returns a future representing the outcome of the task.

Parameters:

• task – The task message

• no_reply (bool) – Do not send a reply containing the result of the task

Returns:

A future corresponding to the outcome of the task

class kiwipy.rmq.threadcomms.RmqThreadTaskQueue(task_queue: RmqTaskQueue, loop_scheduler: LoopScheduler, wrap_subscriber)

Bases: object

Thread task queue.

add_task_subscriber(subscriber)

next_task(timeout=5.0, fail=True)

remove_task_subscriber(subscriber)

task_send(task, no_reply=False)

kiwipy.rmq.threadcomms.connect(connection_params: str | dict | None = None, connection_factory=<function connect_robust>, message_exchange='kiwipy.messages', task_exchange='kiwipy.tasks', task_queue='kiwipy.tasks', task_prefetch_size=0, task_prefetch_count=0, encoder=functools.partial(<function dump>, encoding='utf-8'), decoder=functools.partial(<function load>, Loader=<class 'yaml.loader.FullLoader'>), testing_mode=False) → RmqThreadCommunicator

Establish a RabbitMQ communicator connection