pip_services3_messaging.queues.MemoryMessageQueue module¶

pip_services3_messaging.queues.MemoryMessageQueue¶

Memory message queue implementation.

copyright: Conceptual Vision Consulting LLC 2018-2019, see AUTHORS for more details.
license: MIT, see LICENSE for more details.

class pip_services3_messaging.queues.MemoryMessageQueue.MemoryMessageQueue(name: str = None)¶

Bases: pip_services3_messaging.queues.MessageQueue.MessageQueue, pip_services3_commons.run.ICleanable.ICleanable

Message queue that sends and receives messages within the same process by using shared memory. This queue is typically used for testing to mock real queues.

### Configuration parameters ###

name: name of the message queue

### References ###

*:logger:*:*:1.0 (optional) ILogger components to pass log messages
*:counters:*:*:1.0 (optional) ICounters components to pass collected measurements

Example:

queue = MessageQueue("myqueue")
queue.send("123", MessageEnvelope(None, "mymessage", "ABC"))

message = queue.receive("123", 0)
if message != None:
    # ...
    queue.complete("123", message)

abandon(message: pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope)¶

Returnes message into the queue and makes it available for all subscribers to receive it again. This method is usually used to return a message which could not be processed at the moment to repeat the attempt. Messages that cause unrecoverable errors shall be removed permanently or/and send to dead letter queue.

Parameters: message – a message to return.

clear(correlation_id: Optional[str])¶

Clears component state.

Parameters: correlation_id – (optional) transaction id to trace execution through call chain.

close(correlation_id: Optional[str])¶

Closes component and frees used resources.

Parameters: correlation_id – (optional) transaction id to trace execution through call chain.

complete(message: pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope)¶

Permanently removes a message from the queue. This method is usually used to remove the message after successful processing.

Parameters: message – a message to remove.

configure(config: pip_services3_commons.config.ConfigParams.ConfigParams)¶

Configures component by passing configuration parameters.

Parameters: config – configuration parameters to be set.

end_listen(correlation_id)¶

Ends listening for incoming messages. When this method is call listen() unblocks the thread and execution continues.

Parameters: correlation_id – (optional) transaction id to trace execution through call chain.

is_open()¶

Checks if the component is opened.

Returns: true if the component has been opened and false otherwise.

listen(correlation_id: Optional[str], receiver: pip_services3_messaging.queues.IMessageReceiver.IMessageReceiver)¶

Listens for incoming messages and blocks the current thread until queue is closed.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
receiver – a receiver to receive incoming messages.

move_to_dead_letter(message: pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope)¶

Permanently removes a message from the queue and sends it to dead letter queue.

Parameters: message – a message to be removed.

peek(correlation_id: Optional[str]) → pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope ¶

Peeks a single incoming message from the queue without removing it. If there are no messages available in the queue it returns null.

Parameters: correlation_id – (optional) transaction id to trace execution through call chain.
Returns: a message object.

peek_batch(correlation_id: Optional[str], message_count: int) → List[pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope]¶

Peeks multiple incoming messages from the queue without removing them. If there are no messages available in the queue it returns an empty list.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
message_count – a maximum number of messages to peek.

Returns

a list of message objects.

read_message_count() → int ¶

Reads the current number of messages in the queue to be delivered.

Returns: a number of messages

receive(correlation_id: Optional[str], wait_timeout: int) → Optional[pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope]¶

Receives an incoming message and removes it from the queue.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
wait_timeout – a timeout in milliseconds to wait for a message to come.

Returns

a message object.

renew_lock(message: pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope, lock_timeout: int)¶

Renews a lock on a message that makes it invisible from other receivers in the queue. This method is usually used to extend the message processing time.

Parameters

message – a message to extend its lock.
lock_timeout – a locking timeout in milliseconds.

send(correlation_id, message)¶

Sends a message into the queue.

Parameters

correlation_id – (optional) transaction id to trace execution through call chain.
message – a message envelop to be sent.