MessageQueue

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

Implements: IMessageQueue, IConfigurable, IReferenceable

Description

The MessageQueue class allows you to create a message queue that is used as a basis for specific message queue implementions.

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: username
  • password: user’s 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)

Constructors

Creates a new instance of the message queue.

MessageQueue(name: str = None, capabilities: MessagingCapabilities = None)

  • name: str - (optional) queue name
  • capabilities: MessagingCapabilities - (optional) capabilities of this message queue

Fields

_lock

Threading lock.

_lock: threading.Lock

_event

Threading event.

_event = threading.Event

_logger

Component used to pass log messages.

_logger: CompositeLogger

_counters

Component to pass collected measurements.

_counters: CompositeCounters

_connectionResolver

Component used to resolve connections.

_connectionResolver: ConnectionResolver

_credentialResolver

Component used to resolve credentials.

_credentialResolver: CredentialResolver

_name

Name of the message queue.

_name: str

_capabilities

Component used to store the message queue.

_capabilities: MessagingCapabilities

Abstract methods

abandon

Returns a 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 sent to dead letter queue.

abstractmethod abandon(message: MessageEnvelope)

clear

Clears a component’s state.

abstractmethod clear(correlation_id: Optional[str])

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.

close

Closes a component and frees the used resources.

abstractmethod close(correlation_id: Optional[str])

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.

complete

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

abstractmethod complete(message: MessageEnvelope)

end_listen

Ends listening for incoming messages. When this method is called, listen unblocks the thread and execution continues.

abstractmethod end_listen(correlation_id: Optional[str])

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.

is_open

Checks if the component is opened.

abstractmethod is_open(): bool

  • returns: bool - True if the component has been opened and False otherwise.

listen

Listens for incoming messages and blocks the current thread until the queue is closed.
See also IMessageReceiver, receive

abstractmethod listen(correlation_id: Optional[str], receiver: IMessageReceiver)

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • receiver: IMessageReceiver - receiver used to receive incoming messages.

move_to_dead_letter

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

abstractmethod move_to_dead_letter(message: MessageEnvelope)

peek

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

abstractmethod peek(correlation_id: Optional[str]): MessageEnvelope

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • returns: MessageEnvelope - peeked message or None.

peek_batch

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

abstractmethod peek_batch(correlation_id: Optional[str], message_count: int): List[MessageEnvelope]

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • message_count: int - maximum number of messages to peek.
  • returns: List[MessageEnvelope] - list of peeked messages

read_message_count

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

abstractmethod read_message_count(): int

  • returns: int - number of messages in the queue.

receive

Receives an incoming message and removes it from the queue.

abstractmethod receive(correlation_id: Optional[str], wait_timeout: int): MessageEnvelope

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • wait_timeout: int - timeout in milliseconds to wait for a message to come.
  • returns: MessageEnvelope - received message or None.

renew_lock

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.

abstractmethod renew_lock(message: MessageEnvelope, lock_timeout: int)

  • message: MessageEnvelope - message to extend its lock.
  • lock_timeout: int - locking timeout in milliseconds.

send

Sends a message into the queue.

abstractmethod send(correlation_id: Optional[str], envelope: MessageEnvelope)

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • envelope: MessageEnvelope - message envelop to be sent.

send_as_object

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

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

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • message_type: str - a message type
  • message: Any - an object value to be sent

Instance methods

begin_listen

Listens for incoming messages without blocking the current thread.
See also listen, IMessageReceiver

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

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • receiver: IMessageReceiver - receiver used to receive incoming messages.

configure

Configures the component by passing its configuration parameters.

configure(config: ConfigParams)

  • config: ConfigParams - configuration parameters to be set.

get_capabilities

Gets the queue capabilities

get_capabilities(): MessagingCapabilities

get_name

Gets the queue name

get_name(): str

  • returns: str - queue name.

open

Opens the component.

open(correlation_id: Optional[str])

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.

send_as_object

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

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

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • message_type: str - message type
  • message: Any - object value to be sent

set_references

Sets references to dependent components.

set_references(references: IReferences)

  • references: IReferences - references to locate the component dependencies.

to_string

Gets a string representation of the object.

to_string(): str

  • returns: str - string representation of the object.

_open_with_params

Opens the component with the given connection and credential parameters.

_open_with_params(correlation_id: Optional[str], connections: List[ConnectionParams], credentials: CredentialParams)

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • connections: List[ConnectionParams] - connection parameters
  • credentials: CredentialParams - credential parameters

_check_open

Checks if the queue has been opened.

_check_open(correlation_id: Optional[str])

  • correlation_id: Optional[str] - (optional) transaction id used to trace execution through the call chain.
  • returns: Exception - an exception if queue wasn’t opened or None otherwise