Docs / Backends / NATS

NATS

NATS Core (at-most-once pub/sub with queue groups) and JetStream (streams, durable consumers, acks)

import "go.digitalxero.dev/mq-nats/v2"

Index

func NewMessageBuilder() types.MessageBuilder
func NewNatsConsumer(config *viper.Viper) (types.Consumer, error)
func NewNatsProducer(config *viper.Viper) (types.Producer, error)
func NewNatsTransport() types.Transport

funcNewMessageBuilder

func NewMessageBuilder() types.MessageBuilder

NewMessageBuilder creates a new NATS message builder. The builder provides a fluent interface for constructing NATS messages with all supported properties and attributes.

Example:

msg := nats.NewMessageBuilder().
    WithBody([]byte(`{"sensor": "temp-01", "value": 23.5}`)).
    WithContentType("application/json").
    WithCorrelationId("reading-456").
    WithHeaders(map[string]interface{}{
        "location": "warehouse-1",
        "timestamp": time.Now().Unix(),
    }).
    WithReplyTo("sensor.responses"). // For request-reply pattern
    Build()

funcNewNatsConsumer

func NewNatsConsumer(config *viper.Viper) (types.Consumer, error)

NewNatsConsumer creates a new NATS/JetStream consumer with the provided configuration. This function is typically called by mqutils.NewConsumer when it detects a NATS URL.

key	default	description
url	required	NATS server URLs e.g., “nats://localhost:4222”
subject	required	Subject to consume from
handler	“natsLogger”	Name of registered handler function
queue_group	optional	Queue group for load balancing
use_jetstream	false	Enable JetStream mode
stream_name	—	JetStream stream name (required when use_jetstream is true)
consumer_name	“mqutils-consumer”	JetStream consumer name
durable	false	Create a durable JetStream consumer
max_deliver	50	Max JetStream delivery attempts before dropping
ack_wait	30000	JetStream ack wait in ms
auto_reconnect	false	Reconnect the consumer loop on connection loss
skip_verify	false	Skip TLS certificate verification
max_reconnects	60	Max reconnection attempts
reconnect_wait	2000	Wait between reconnects in ms
reconnect_buf_size	0 = 8MB library default; -1 disables	Client-side buffer (bytes) for publishes while disconnected
fetch_batch_size	10	JetStream pull-fetch batch size
fetch_max_wait_ms	1000	JetStream pull-fetch max wait in ms
message_channel_buffer	10	Buffer size for message channel
batch_size	5	Number of messages per batch
batch_timeout	100ms	Batch collection timeout as a duration
enable_batch_processing	false	Enable batch message processing
max_concurrent_handlers	effective message_channel_buffer; must be positive	Maximum in-flight handlers or batches

The canonical keys destination, max_retries, and consumer_group are accepted as aliases for subject, max_deliver, and queue_group respectively; the native key wins when both are set.

Returns an error if configuration validation fails or transport creation fails.

funcNewNatsProducer

func NewNatsProducer(config *viper.Viper) (types.Producer, error)

NewNatsProducer creates a new NATS/JetStream producer with the provided configuration. This function is typically called by mqutils.NewProducer when it detects a NATS URL.

key	default	description
url	required	NATS server URLs e.g., “nats://localhost:4222”
use_jetstream	false	Enable JetStream mode for persistence
stream_name	—	JetStream stream to declare on Start (required when use_jetstream is true)
skip_verify	false	Skip TLS certificate verification
max_reconnects	60	Max reconnection attempts
reconnect_wait	2000	Wait between reconnects in ms
reconnect_buf_size	0 = 8MB library default; -1 disables buffering	Client-side buffer (bytes) that queues outgoing publishes while the connection is down; beyond it publishes fail with “nats: outbound buffer limit exceeded”
request_timeout	5000	Timeout for request-reply in ms

The canonical key destination is accepted as an alias for stream_name; the native key wins when both are set.

Returns an error if configuration validation fails or transport creation fails. The producer must be started with Start() before publishing messages.

funcNewNatsTransport

func NewNatsTransport() types.Transport

NewNatsTransport creates a new NATS transport implementation with default options. The transport provides low-level NATS operations including connection management, publishing, and subscription handling for both Core NATS and JetStream.

The transport manages:

NATS connections with automatic reconnection
JetStream context for persistent messaging
Subscription lifecycle management
Authentication (credentials, TLS)
Health monitoring

The transport supports both standard NATS (at-most-once delivery) and JetStream (at-least-once with persistence) messaging patterns.

Generated by gomarkdoc