queues

package
v1.0.8 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jan 12, 2023 License: MIT Imports: 12 Imported by: 2

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

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(cconf.NewConfigParamsFromTuples(
      "subject", "mytopic",
      "connection.protocol", "mqtt"
      "connection.host", "localhost"
      "connection.port", 1883
    ))

    queue.open("123")

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

    message, err := queue.Receive("123")
	if (message != nil) {
		...
		queue.Complete("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(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:

  • message *cqueues.MessageEnvelope a message to return.

Returns: error 

error or nil for success.

func (*MqttMessageQueue) Clear 

func (c *MqttMessageQueue) Clear(correlationId string) error

Clear method are clears component state. Parameters:

  • 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(correlationId string) (err error)

Closes component and frees used resources.

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

func (*MqttMessageQueue) Complete 

func (c *MqttMessageQueue) Complete(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:

  • message *cqueues.MessageEnvelope a message to remove.

Returns: error error or nil for success.

func (*MqttMessageQueue) Configure added in v1.0.2 

func (c *MqttMessageQueue) Configure(config *cconf.ConfigParams)

Configures component by passing configuration parameters.

  • config configuration parameters to be set.

func (*MqttMessageQueue) EndListen 

func (c *MqttMessageQueue) EndListen(correlationId string)

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

  • 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(correlationId string, receiver cqueues.IMessageReceiver) error

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

  • 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(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:

  • message *cqueues.MessageEnvelope a message to be removed.

Returns: error 

error or nil for success.

func (*MqttMessageQueue) OnMessage added in v1.0.2 

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

func (*MqttMessageQueue) Open added in v1.0.2 

func (c *MqttMessageQueue) Open(correlationId string) (err error)

Opens the component.

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

func (*MqttMessageQueue) Peek 

func (c *MqttMessageQueue) Peek(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:

  • 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(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:

  • 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(correlationId string, waitTimeout time.Duration) (*cqueues.MessageEnvelope, error)

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

  • 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(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:

  • 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(correlationId string, envelop *cqueues.MessageEnvelope) error

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

  • 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 added in v1.0.2 

func (c *MqttMessageQueue) SetReferences(references cref.IReferences)

Sets references to dependent components.

  • references references to locate the component dependencies.

func (*MqttMessageQueue) UnsetReferences added in v1.0.2 

func (c *MqttMessageQueue) UnsetReferences()

Unsets (clears) previously set references to dependent components.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.