NatsMessageQueue

Message queue that sends and receives messages via the NATS message broker.

Implements: NatsAbstractMessageQueue

Description

The NatsMessageQueue class allows you to create a message queue that sends and receives messages via the NATS message broker.

Configuration parameters

  • subject: name of NATS topic (subject) to subscribe
  • queue_group: name of NATS queue group
  • connection(s):
    • discovery_key: (optional) 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) key to retrieve the credentials from ICredentialStore
    • username: username
    • password: user’s password
  • options:
    • serialize_message: (optional) true to serialize entire message as JSON, false to send only the message payload (default: true)
    • autosubscribe: (optional) true to automatically subscribe on option (default: false)
    • retry_connect: (optional) turns on/off automated reconnect when connection is lost (default: true)
    • max_reconnect: (optional) maximum number of reconnection attempts (default: 3)
    • reconnect_timeout: (optional) number of milliseconds to wait on each reconnection attempt (default: 3000)
    • flush_timeout: (optional) number of milliseconds to wait on flushing messages (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:nats:*:1.0 - (optional) shared connection to NATS service

Constructors

NewNatsMessageQueue

Creates a new instance of the message queue.

NewNatsMessageQueue(name string) *NatsMessageQueue

  • name: string - (optional) queue name.

Fields

autoSubscribe

Auto-subscribe option

autoSubscribe: bool

messages

Messages

messages: []MessageEnvelope

receiver

Message receiver

receiver: IMessageReceiver

subscribed

Subscribe option

subscribed: bool

Methods

Clear

Clears a component’s state.

(c [*NatsMessageQueue[T]]) Clear(ctx context.Context, correlationId string) (err error)

  • ctx: context.Context - operation context.
  • correlationId: string - (optional) transaction id used to trace execution through the call chain.
  • returns: (err error) - error or nil if no errors occurred.

Close

Closes a component and frees used resources.

(c [*NatsMessageQueue[T]]) Close(ctx context.Context, correlationId string) error

  • ctx: context.Context - operation context.
  • correlationId: string - (optional) transaction id used to trace execution through the call chain.
  • returns: error - error or nil if no errors occurred.

Configure

Configures a component by passing its configuration parameters.

(c [*NatsMessageQueue[T]]) Configure(ctx context.Context, config *ConfigParams)

  • ctx: context.Context - operation context.
  • config:: *ConfigParams - configuration parameters to be set.

EndListen

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

(c [*NatsMessageQueue[T]]) EndListen(ctx context.Context, correlationId string)

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

Listen

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

See IMessageReceiver

(c [*NatsMessageQueue[T]]) Listen(ctx context.Context, correlationId string, receiver IMessageReceiver) error

  • ctx: context.Context - operation context.
  • correlationId: string - (optional) transaction id used to trace execution through the call chain.
  • receiver: IMessageReceiver - receiver used to receive incoming messages.
  • returns: error - error or nil no errors occured.

OnMessage

Checks if the message is not nil. If this is the case, it deserializes and sends the message to the receiver if it’s set. Otherwise, puts the message into the queue.

(c [*NatsMessageQueue[T]]) OnMessage(msg *nats.Msg)

  • msg: *nats.Msg - message

Open

Opens the component.

(c [*NatsMessageQueue[T]]) Open(ctx context.Context, correlationId string) error

  • ctx: context.Context - operation context.
  • correlationId: string - (optional) transaction id used to trace execution through the call chain.
  • returns: 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 [*NatsMessageQueue[T]]) Peek(ctx context.Context, correlationId string) (*MessageEnvelope, error)

  • ctx: context.Context - operation context.
  • correlationId: string - (optional) transaction id used to trace execution through the call chain.
  • returns: (*MessageEnvelope, error) - peeked message.

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 NATS.

(c [*NatsMessageQueue[T]]) PeekBatch(ctx context.Context, correlationId string, messageCount int64) ([]*MessageEnvelope, error)

  • ctx: context.Context - operation context.
  • correlationId: string - (optional) transaction id used to trace execution through the 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 [*NatsMessageQueue[T]]) ReadMessageCount() (count int64, err error)

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

Receive

Receives an incoming message and removes it from the queue.

(c [*NatsMessageQueue[T]]) Receive(ctx context.Context, correlationId string, waitTimeout time.Duration) (*MessageEnvelope, error)

  • ctx: context.Context - operation context.
  • correlationId: string - 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.
  • waitTimeout: time.Duration - timeout (milliseconds) to wait for a message to come.
  • returns: (*MessageEnvelope, error) - received message or nil if nothing was received.

Subscribe

Subscribes to a subject.

(c [*NatsMessageQueue[T]]) subscribe(correlationId string) error

  • correlationId: string - (optional) transaction id used to trace execution through the call chain.
  • returns: error - error or nil if no errors occurred.

Examples

ctx := context.Background()
queue := NewNatsMessageQueue("myqueue")
queue.Configure(ctx, cconf.NewConfigParamsFromTuples(
	"subject", "mytopic",
	"queue_group", "mygroup",
	"connection.protocol", "nats"
	"connection.host", "localhost"
	"connection.port", 1883
))

_ = queue.Open(ctx, "123")

_ = queue.Send(ctx, "123", NewMessageEnvelope("", "mymessage", "ABC"))

message, err := queue.Receive(ctx, "123", 10000*time.Milliseconds)
if (message != nil) {
	...
	queue.Complete(ctx, message)
}

See also