queues

Documentation

Index

• type MqttMessageQueue
  • func NewMqttMessageQueue(name string) *MqttMessageQueue
  • func (c *MqttMessageQueue) Abandon(ctx context.Context, message *cqueues.MessageEnvelope) error
  • func (c *MqttMessageQueue) Clear(ctx context.Context, correlationId string) error
  • func (c *MqttMessageQueue) Close(ctx context.Context, correlationId string) (err error)
  • func (c *MqttMessageQueue) Complete(ctx context.Context, message *cqueues.MessageEnvelope) error
  • func (c *MqttMessageQueue) Configure(ctx context.Context, config *cconf.ConfigParams)
  • func (c *MqttMessageQueue) EndListen(ctx context.Context, correlationId string)
  • func (c *MqttMessageQueue) IsOpen() bool
  • func (c *MqttMessageQueue) Listen(ctx context.Context, correlationId string, receiver cqueues.IMessageReceiver) error
  • func (c *MqttMessageQueue) MoveToDeadLetter(ctx context.Context, message *cqueues.MessageEnvelope) error
  • func (c *MqttMessageQueue) OnMessage(msg mqtt.Message)
  • func (c *MqttMessageQueue) Open(ctx context.Context, correlationId string) (err error)
  • func (c *MqttMessageQueue) Peek(ctx context.Context, correlationId string) (*cqueues.MessageEnvelope, error)
  • func (c *MqttMessageQueue) PeekBatch(ctx context.Context, correlationId string, messageCount int64) ([]*cqueues.MessageEnvelope, error)
  • func (c *MqttMessageQueue) ReadMessageCount() (int64, error)
  • func (c *MqttMessageQueue) Receive(ctx context.Context, correlationId string, waitTimeout time.Duration) (*cqueues.MessageEnvelope, error)
  • func (c *MqttMessageQueue) RenewLock(ctx context.Context, message *cqueues.MessageEnvelope, ...) (err error)
  • func (c *MqttMessageQueue) Send(ctx context.Context, correlationId string, envelop *cqueues.MessageEnvelope) error
  • func (c *MqttMessageQueue) SetReferences(ctx context.Context, references cref.IReferences)
  • func (c *MqttMessageQueue) UnsetReferences()

Types

type MqttMessageQueue

type MqttMessageQueue struct {
	*cqueues.MessageQueue

	//The dependency resolver.
	DependencyResolver *cref.DependencyResolver
	//The logger.
	Logger *clog.CompositeLogger
	//The MQTT connection component.
	Connection *connect.MqttConnection
	// contains filtered or unexported fields
}

MqttMessageQueue are message queue that sends and receives messages via MQTT message broker.

Configuration parameters:

• topic: name of MQTT topic to subscribe
• connection(s):
• discovery_key: (optional) a key to retrieve the connection from 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) a key to retrieve the credentials from ICredentialStore
• username: user name
• password: user 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 log (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 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 to resolve connections
• *:credential-store:*:*:1.0 (optional) Credential stores to resolve credentials
• *:connection:mqtt:*:1.0 (optional) Shared connection to MQTT service

See MessageQueue See MessagingCapabilities

Example:

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);
}

func NewMqttMessageQueue

func NewMqttMessageQueue(name string) *MqttMessageQueue

Creates a new instance of the queue component.

• name (optional) a queue name.

func (*MqttMessageQueue) Abandon

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

Abandon method are 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. Important: This method is not supported by MQTT. Parameters:

• ctx context.Context operation context.
• message *cqueues.MessageEnvelope a message to return.

Returns: error

error or nil for success.

func (*MqttMessageQueue) Clear

func (c *MqttMessageQueue) Clear(ctx context.Context, correlationId string) error

Clear method are clears component state. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.

Returns error or nil no errors occured.

func (*MqttMessageQueue) Close

func (c *MqttMessageQueue) Close(ctx context.Context, correlationId string) (err error)

Closes component and frees used resources. Parameters:

• ctx context.Context operation context.
• correlationId (optional) transaction id to trace execution through call chain.
• Returns error or nil no errors occured.

func (*MqttMessageQueue) Complete

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

Complete method are 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. Parameters:

• ctx context.Context operation context.
• message *cqueues.MessageEnvelope a message to remove.

Returns: error error or nil for success.

func (*MqttMessageQueue) Configure

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

Configures component by passing configuration parameters. Parameters:

• ctx context.Context operation context.
• config configuration parameters to be set.

func (*MqttMessageQueue) EndListen

func (c *MqttMessageQueue) EndListen(ctx context.Context, correlationId string)

EndListen method are ends listening for incoming messages. When this method is call listen unblocks the thread and execution continues. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.

func (*MqttMessageQueue) IsOpen

func (c *MqttMessageQueue) IsOpen() bool

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

func (*MqttMessageQueue) Listen

func (c *MqttMessageQueue) Listen(ctx context.Context, correlationId string, receiver cqueues.IMessageReceiver) error

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

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.
• receiver cqueues.IMessageReceiver a receiver to receive incoming messages.

See IMessageReceiver See receive

func (*MqttMessageQueue) MoveToDeadLetter

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

Permanently removes a message from the queue and sends it to dead letter queue. Important: This method is not supported by MQTT. Parameters:

• ctx context.Context operation context.
• message *cqueues.MessageEnvelope a message to be removed.

Returns: error

error or nil for success.

func (*MqttMessageQueue) OnMessage

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

func (*MqttMessageQueue) Open

func (c *MqttMessageQueue) Open(ctx context.Context, correlationId string) (err error)

Opens the component. Parameters:

• ctx context.Context operation context.
• correlationId (optional) transaction id to trace execution through call chain.
• Returns error or nil no errors occured.

func (*MqttMessageQueue) Peek

func (c *MqttMessageQueue) Peek(ctx context.Context, correlationId string) (*cqueues.MessageEnvelope, error)

Peek method are peeks a single incoming message from the queue without removing it. If there are no messages available in the queue it returns nil. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.

Returns: result *cqueues.MessageEnvelope, err error message or error.

func (*MqttMessageQueue) PeekBatch

func (c *MqttMessageQueue) PeekBatch(ctx context.Context, correlationId string, messageCount int64) ([]*cqueues.MessageEnvelope, error)

PeekBatch method are 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. Parameters:

• ctx context.Context operation context.
• correlationId (optional) transaction id to trace execution through call chain.
• messageCount a maximum number of messages to peek.

Returns: callback function that receives a list with messages or error.

func (*MqttMessageQueue) ReadMessageCount

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

ReadMessageCount method are reads the current number of messages in the queue to be delivered. Returns number of messages or error.

func (*MqttMessageQueue) Receive

func (c *MqttMessageQueue) Receive(ctx context.Context, correlationId string, waitTimeout time.Duration) (*cqueues.MessageEnvelope, error)

Receive method are receives an incoming message and removes it from the queue. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.
• waitTimeout time.Duration a timeout in milliseconds to wait for a message to come.

Returns: result *cqueues.MessageEnvelope, err error receives a message or error.

func (*MqttMessageQueue) RenewLock

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

RenewLock method are 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. Parameters:

• ctx context.Context operation context.
• message *cqueues.MessageEnvelope a message to extend its lock.
• lockTimeout time.Duration a locking timeout in milliseconds.

Returns: error receives an error or nil for success.

func (*MqttMessageQueue) Send

func (c *MqttMessageQueue) Send(ctx context.Context, correlationId string, envelop *cqueues.MessageEnvelope) error

Send method are sends a message into the queue. Parameters:

• ctx context.Context operation context.
• correlationId string (optional) transaction id to trace execution through call chain.
• envelope *cqueues.MessageEnvelope a message envelop to be sent.

Returns: error or nil for success.

func (*MqttMessageQueue) SetReferences

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

Sets references to dependent components. Parameters:

• ctx context.Context operation context.
• references references to locate the component dependencies.

func (*MqttMessageQueue) UnsetReferences

func (c *MqttMessageQueue) UnsetReferences()

Unsets (clears) previously set references to dependent components.