queues

package
v0.0.1-2 Latest Latest
Warning

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

 Go to latest Published: Oct 24, 2023 License: MIT Imports: 14 Imported by: 0

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(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) error

Clear method are clears component state. Parameters:

  • ctx context.Context 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) (err error)

Closes component and frees used resources. Parameters:

  • ctx context.Context 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)

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

  • ctx context.Context 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, receiver cqueues.IMessageReceiver) error

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

  • ctx context.Context 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) (err error)

Opens the component. Parameters:

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

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

  • ctx context.Context 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, envelop *cqueues.MessageEnvelope) error

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

  • ctx context.Context 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.

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.