MqttMessageQueue

Message queue that sends and receives messages via an MQTT message broker.

Implements: MessageQueue

Description

The MqttMessageQueue class allows you to create message queues that send and receive messages via an MQTT message broker.

Configuration parameters

  • topic: name of MQTT topic to subscribe
  • connection(s):
    • discovery_key: (optional) key to retrieve the connection from [IDiscovery]../../../config/connect/idiscovery)
    • host: host name or IP address
    • port: port number
    • uri: resource URI or connection string with all parameters in it
  • credential(s):
    • store_key: (optional) key to retrieve the credentials from ICredentialStore
    • username: username
    • password: user’s password
  • options:
    • serialize_envelope: (optional) true to serialize entire message as JSON, false to send only message payload (default: true)
    • autosubscribe: (optional) true to automatically subscribe on option (default: false)
    • qos: (optional) quality of service level aka QOS (default: 0)
    • retain: (optional) retention flag for published messages (default: false)
    • retry_connect: (optional) turns on/off automated reconnect when connection is lost (default: true)
    • connect_timeout: (optional) number of milliseconds to wait for connection (default: 30000)
    • reconnect_timeout: (optional) number of milliseconds to wait on each reconnection attempt (default: 1000)
    • keepalive_timeout: (optional) number of milliseconds to ping the broker while inactive (default: 3000)

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 services
  • *:credential-store:*:*:1.0 - (optional) ICredentialStore to resolve credentials
  • *:connection:mqtt:*:1.0 - (optional) shared connection to MQTT service

Constructors

NewMqttMessageQueue

Creates a new instance of the message queue.

NewMqttMessageQueue(name string) *MqttMessageQueue

  • name: string - (optional) queue’s name.

Fields

autoSubscribe

Auto-subscribe option

autoSubscribe: bool

Connection

MQTT connection component

Connection: *MqttConnection

DependencyResolver

Dependency resolver

DependencyResolver: *DependencyResolver

Logger

Logger

Logger: *CompositeLogger

config

Configuration options

config: *ConfigParams

_messages

Message

_messages: MessageEnvelope[] = []

qos

Quality of service

qos: byte

receiver

Message receiver

receiver: IMessageReceiver

serializeEnvelope

Serialization option

serializeEnvelope: bool

subscribed

Subscribe option

subscribed: bool

topic

Topic

topic: string

Methods

Abandon

Returnes 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 send to the dead letter queue.

  • Important: This method is not supported by MQTT.

(c *MqttMessageQueue) Abandon(ctx context.Context, message *MessageEnvelope) error

Clear

Clears a component’s state.

(c *MqttMessageQueue) Clear(ctx context.Context, context IContext) error

  • ctx: context.Context - operaton context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • returns: error - error or nil if no errors occurred.

Close

Closes a component and frees used resources.

(c *MqttMessageQueue) Close(ctx context.Context context IContext) (err error)

  • ctx: context.Context - operation context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • returns: (err error) - error or nil if no errors occurred.

Complete

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

  • Important: This method is not supported by MQTT.

(c *MqttMessageQueue) Complete(ctx context.Context, message *MessageEnvelope) error

Configure

Configures a component by passing its configuration parameters.

(c *MqttMessageQueue) Configure(ctx context.Context, config *ConfigParams)

EndListen

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

(c *MqttMessageQueue) EndListen(ctx context.Context, context IContext)

FromMessage

Returns the topic and the data of a message.

(c *MqttMessageQueue) fromMessage(message *MessageEnvelope) ([]byte, error)

GetTopic

Obtains the topic.

(c *MqttMessageQueue) getTopic() string

  • returns: string - topic

IsOpen

Checks if the component is open.

(c *MqttMessageQueue) IsOpen() bool

  • returns: bool - true if the component is open and false otherwise.

Listen

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

See IMessageReceiver

(c *MqttMessageQueue) Listen(ctx context.Context, context IContext, receiver IMessageReceiver) error

  • ctx: context.Context - operation context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • receiver: IMessageReceiver - receiver used to receive incoming messages.
  • returns: error - error or nil if no errors occurred.

MoveToDeadLetter

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

  • Important: This method is not supported by MQTT.

(c *MqttMessageQueue) MoveToDeadLetter(ctx context.Context, message *MessageEnvelope) error

OnMessage

Checks if the message comes from the right topic. If this is the case, it deserializes and sends the message to the receiver if it’s set. Otherwise, it puts the message into the queue.

(c *MqttMessageQueue) OnMessage(msg mqtt.Message)

  • msg: mqtt.Message - MQTT message with data and topic

Open

Opens the component.

(c *MqttMessageQueue) Open(ctx context.Context, context IContext) (err error)

  • ctx: context.Context - operation context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • returns: (err error) - error or nil if no errors occurred.

Peek

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

(c *MqttMessageQueue) Peek(ctx context.Context, context IContext) (*MessageEnvelope, error)

PeekBatch

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

  • Important: This method is not supported by MQTT.

(c *MqttMessageQueue) PeekBatch(ctx context.Context, context IContext, messageCount int64) ([]*MessageEnvelope, error)

  • ctx: context.Context - operation context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • messageCount: int64 - maximum number of messages to peek.
  • returns: ([]*MessageEnvelope, error) - list with peeked messages.

ReadMessageCount

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

(c *MqttMessageQueue) ReadMessageCount() (int64, error)

  • *returns: (int64, error) - number of messages in the queue.

Receive

Receives an incoming message and removes it from the queue.

(c *MqttMessageQueue) Receive(ctx context.Context, context IContext, waitTimeout time.Duration) (*MessageEnvelope, error)

  • ctx: context.Context - operation context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • waitTimeout: time.Duration - timeout in milliseconds to wait for a message to come.
  • returns: (*MessageEnvelope, error) - received message or nil if nothing was received.

RenewLock

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.

  • Important: This method is not supported by MQTT.

(c *MqttMessageQueue) RenewLock(ctx context.Context, message *MessageEnvelope, lockTimeout time.Duration) (err error)

  • ctx: context.Context - operation context.
  • message: *MessageEnvelope - message to extend its lock.
  • lockTimeout: time.Duration - locking timeout in milliseconds.
  • returns: (err error) - error or nil if no errors occurred.

Send

Sends a message into the queue.

(c *MqttMessageQueue) Send(ctx context.Context, context IContext, envelop *MessageEnvelope) error

  • ctx: context.Context - operation context.
  • context: IContext - (optional) a context to trace execution through a call chain.
  • message: *MessageEnvelope - message envelop to be sent.
  • returns: error - error or nil if no errors occurred.

SetReferences

Sets references to dependent components.

(c *MqttMessageQueue) SetReferences(ctx context.Context, references IReferences)

Subscribe

Subscribes to a topic.

(c *MqttMessageQueue) subscribe(context IContext) error

  • context: IContext - (optional) a context to trace execution through a call chain.
  • returns: error - error or nil if no errors occurred.

ToMessage

If the message has no data, it returns nil. Otherwise, it returns the message.

(c *MqttMessageQueue) toMessage(msg mqtt.Message) (*MessageEnvelope, error)

  • msg: mqtt.Message - MQTT message with data and topic
  • returns: (*MessageEnvelope, error) - Null if the message has no data. Otherwise, it returns the message.

UnsetReferences

Unsets (clears) previously set references to dependent components.

(c *MqttMessageQueue) UnsetReferences()

Examples

queue := NewMqttMessageQueue("myqueue")
queue.Configure(context.Background(), cconf.NewConfigParamsFromTuples(
	"subject", "mytopic",
	"connection.protocol", "mqtt"
	"connection.host", "localhost"
	"connection.port", 1883
))

queue.Open(context.Background(), "123")

queue.Send(context.Background(), "123", NewMessageEnvelope("", "mymessage", "ABC"))

message, err := queue.Receive(context.Background(), "123")
if (message != nil) {
	...
	queue.Complete(context.Background(), "123", message);
}

See also