pip_services3_messaging.queues.MessageQueue module

pip_services3_messaging.queues.MessageQeueue

Abstract 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.MessageQueue.MessageQueue(name: str = None, capabilities: pip_services3_messaging.queues.MessagingCapabilities.MessagingCapabilities = None)

Bases: pip_services3_commons.config.IConfigurable.IConfigurable, pip_services3_commons.refer.IReferenceable.IReferenceable, pip_services3_messaging.queues.IMessageQueue.IMessageQueue

Abstract message queue.

Abstract message queue that is used as a basis for specific message queue implementations.

### Configuration parameters ###

  • name: name of the message queue

  • connection(s):

    • discovery_key: key to retrieve parameters from discovery service

    • protocol: connection protocol like http, https, tcp, udp

    • host: host name or IP address

    • port: port number

    • uri: resource URI or connection string with all parameters in it

  • credential(s):

  • store_key: key to retrieve parameters from credential store

  • username: user name

  • password: user password

  • access_id: application access id

  • access_key: application secret key

### References ###

  • *:logger:*:*:1.0 (optional) ILogger components to pass log messages

  • *:counters:*:*:1.0 (optional) ICounters components to pass collected measurements

  • *:discovery:*:*:1.0 (optional) IDiscovery components to discover connection(s)

  • *:credential-store:*:*:1.0 (optional) ICredentialStore componetns to lookup credential(s)

abstract 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.

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

Listens for incoming messages without blocking the current thread.

Parameters

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

  • receiver – a receiver to receive incoming messages.

abstract clear(correlation_id: Optional[str])

Clears component state.

Parameters

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

abstract close(correlation_id: Optional[str])

Closes component and frees used resources.

Parameters

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

abstract 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. :param 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.

abstract end_listen(correlation_id: Optional[str])

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.

get_capabilities() → pip_services3_messaging.queues.MessagingCapabilities.MessagingCapabilities

Gets the queue capabilities

Returns

the queue’s capabilities object.

get_name() → str

Gets the queue name

Returns

the queue name.

abstract is_open() → bool

Checks if the component is opened.

Returns

true if the component has been opened and false otherwise.

abstract 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.

abstract 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.

open(correlation_id: Optional[str])

Opens the component.

Parameters

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

abstract 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 None.

Parameters

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

Returns

a peeked message or None.

abstract 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 peeked messages

abstract read_message_count() → int

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

Returns

a number of messages in the queue.

abstract receive(correlation_id: Optional[str], wait_timeout: int) → 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 received message or None.

abstract 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.

abstract send(correlation_id: Optional[str], envelop: pip_services3_messaging.queues.MessageEnvelope.MessageEnvelope)

Sends a message into the queue. :param correlation_id: (optional) transaction id to trace execution through call chain. :param envelop: a message envelop to be sent.

send_as_object(correlation_id: Optional[str], message_type: str, message: Any)

Sends an object into the queue. Before sending the object is converted into JSON string and wrapped in a MessageEnvelope.

Parameters

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

  • message_type – a message type

  • message – an object value to be sent

set_references(references: pip_services3_commons.refer.IReferences.IReferences)

Sets references to dependent components.

Parameters

references – references to locate the component dependencies.

to_string() → str

Gets a string representation of the object.

Returns

a string representation of the object.