nats.go Client Library

The jetstream package is a modern API for interacting with NATS JetStream, designed to replace the older JetStream implementation in the nats package. It provides simpler interfaces for managing streams and consumers, uses a more predictable approach to message consumption (favoring pull consumers over the complex Subscribe() method), and separates JetStream context from the core NATS client.

Key Interfaces:

• JetStream: Top-level interface for creating/managing streams, consumers, and publishing messages.
• Stream: Manages consumers for a specific stream and performs stream-specific operations (purging, fetching/deleting messages, getting info).
• Consumer: Used to retrieve consumer information and consume messages.
• Msg: Used for message-specific operations like reading data/headers/metadata and performing acknowledgements.
• KeyValue (KV) and Object Stores: Abstraction layers built on top of JetStream for simplified key-value and large data storage.

Requirement: jetstream requires nats-server version 2.9.0 or higher.

The legacy js.Subscribe() (which used push consumers) should be replaced with pull consumers using Consume() or Messages(). Pull consumers are recommended because they provide better flow control and prevent slow consumer issues.

Using Consume() (Callback approach)

Consume() is the closest equivalent to the legacy callback-based js.Subscribe(). It delivers messages to a callback function continuously.

Using Messages() (Iterator approach)

Messages() provides an iterator-based approach, giving you explicit control over when the next message is fetched via iter.Next().

Handling Queue Semantics

In the new API, you do not need an explicit queue group for pull consumers. Multiple instances calling Consume() or Messages() on the same durable consumer will naturally distribute messages among themselves. If you specifically require push-based queue semantics, use CreateOrUpdatePushConsumer and set the DeliverGroup field.

// New: callback with Consume()
s, _ := js.Stream(ctx, "ORDERS")
cons, _ := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
    Durable:       "processor",
    FilterSubject: "ORDERS.*",
})

cc, _ := cons.Consume(func(msg jetstream.Msg) {
    fmt.Printf("Received: %s\n", string(msg.Data()))
    msg.Ack()
})
defer cc.Stop()

Object Stores support Watchers to notify you of changes in a bucket. A Watcher sends metadata updates (name, bucket, size, etc.) via a channel.

Important: Watchers do not retrieve the actual object data; use the Get method if you need the content.

By default, a watcher sends the latest information for all objects in the bucket, followed by a nil value on the channel to signal that the initial state has been sent. Subsequent updates are sent as changes occur.

Watcher configuration options:

• IncludeHistory: Sends historical updates for each object.
• IgnoreDeletes: Does not pass objects with delete markers.
• UpdatesOnly: Only passes updates for objects that were not already present when the watcher started.

js, _ := jetstream.New(nc)
ctx := context.Background()
os, _ := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "configs"})

os.PutString(ctx, "config-1", "first config")

// Watch for changes
watcher, _ := os.Watch(ctx)
defer watcher.Stop()

// create a second object
os.PutString(ctx, "config-2", "second config")

// update metadata of the first object
os.UpdateMeta(ctx, "config-1", jetstream.ObjectMeta{Name: "config-1", Description: "updated config"})

// Receive initial values
object := <-watcher.Updates()

// Receive nil after initial values are sent
object = <-watcher.Updates()
if object != nil {
    fmt.Println("Unexpected object received")
}

// Receive updates
object = <-watcher.Updates()
// ...

The legacy JetStream API provides methods to manage the lifecycle of streams and consumers.

• Create a Stream: Use js.AddStream(&nats.StreamConfig{...}) specifying the Name and Subjects.
• Update a Stream: Use js.UpdateStream(&nats.StreamConfig{...}) to modify existing stream properties like MaxBytes.
• Create a Consumer: Use js.AddConsumer(streamName, &nats.ConsumerConfig{...}) to define a consumer, often using a Durable name.
• Delete a Consumer: Use js.DeleteConsumer(streamName, consumerName).
• Delete a Stream: Use js.DeleteStream(streamName).

import "github.com/nats-io/nats.go"

// Connect to NATS
nc, _ := nats.Connect(nats.DefaultURL)

// Create JetStream Context
js, _ := nc.JetStream()

// Create a Stream
js.AddStream(&nats.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"ORDERS.*"},
})

// Update a Stream
js.UpdateStream(&nats.StreamConfig{
    Name:     "ORDERS",
    MaxBytes: 8,
})

// Create a Consumer
js.AddConsumer("ORDERS", &nats.ConsumerConfig{
    Durable: "MONITOR",
})

// Delete Consumer
js.DeleteConsumer("ORDERS", "MONITOR")

// Delete Stream
js.DeleteStream("ORDERS")

The legacy JetStream API allows for synchronous and asynchronous publishing, as well as various subscription models (ephemeral, durable, and pull-based).

To use it, connect to NATS using nats.Connect and create a JetStream context via nc.JetStream().

Key patterns include:

• Synchronous Publishing: Use js.Publish(subject, data).
• Asynchronous Publishing: Use js.PublishAsync(subject, data) and wait for completion using js.PublishAsyncComplete().
• Ephemeral Subscriptions: Use js.Subscribe(subject, callback).
• Durable Sync Subscriptions: Use js.SubscribeSync(subject, ...opts) to create a consumer with a specific name (e.g., nats.Durable("NAME")).
• Pull Consumers: Use js.PullSubscribe(subject, durableName) and retrieve messages using sub.Fetch(batchSize).

import "github.com/nats-io/nats.go"

// Connect to NATS
nc, _ := nats.Connect(nats.DefaultURL)

// Create JetStream Context
js, _ := nc.JetStream(nats.PublishAsyncMaxPending(256))

// Simple Stream Publisher
js.Publish("ORDERS.scratch", []byte("hello"))

// Simple Async Stream Publisher
for i := 0; i < 500; i++ {
    js.PublishAsync("ORDERS.scratch", []byte("hello"))
}
select {
case <-js.PublishAsyncComplete():
case <-time.After(5 * time.Second):
    fmt.Println("Did not resolve in time")
}

// Simple Async Ephemeral Consumer
js.Subscribe("ORDERS.*", func(m *nats.Msg) {
    fmt.Printf("Received a JetStream message: %s\n", string(m.Data))
})

// Simple Sync Durable Consumer
sub, err := js.SubscribeSync("ORDERS.*", nats.Durable("MONITOR"), nats.MaxDeliver(3))
m, err := sub.NextMsg(timeout)

// Simple Pull Consumer
sub, err := js.PullSubscribe("ORDERS.*", "MONITOR")
msgs, err := sub.Fetch(10)

// Unsubscribe/Drain
sub.Unsubscribe()
sub.Drain()

When using a cluster of NATS servers, use nats.Connect with a comma-separated list of server URLs. You can configure reconnection behavior using several options to ensure high availability:

• nats.MaxReconnects(n): Sets the maximum number of reconnection attempts.
• nats.ReconnectWait(d): Sets the duration to wait between reconnection attempts.
• nats.ReconnectJitter(nonTLS, tls): Adds random jitter to reconnection attempts to prevent thundering herds. Use different values for non-TLS and TLS connections.
• nats.CustomReconnectDelay(func(attempts int) time.Duration): Provides a custom backoff function that receives the number of attempts and returns the sleep duration.
• nats.DontRandomize(): Disables the randomization of the server pool selection.

You can also set lifecycle callbacks to handle connection events:

• nats.DisconnectErrHandler(func(nc *nats.Conn, err error)): Triggered when the connection is lost.
• nats.ReconnectHandler(func(nc *nats.Conn)): Triggered when a reconnection is successful.
• nats.ClosedHandler(func(nc *nats.Conn)): Triggered when the connection is permanently closed.

var servers = "nats://localhost:1222, nats://localhost:1223, nats://localhost:1224"

// Example with reconnection settings and callbacks
nc, err := nats.Connect(servers, 
    nats.MaxReconnects(5), 
    nats.ReconnectWait(2 * time.Second),
    nats.ReconnectJitter(500*time.Millisecond, 2*time.Second),
    nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
        fmt.Printf("Got disconnected! Reason: %q\n", err)
    }),
    nats.ReconnectHandler(func(nc *nats.Conn) {
        fmt.Printf("Got reconnected to %v!\n", nc.ConnectedUrl())
    }),
    nats.ClosedHandler(func(nc *nats.Conn) {
        fmt.Printf("Connection closed. Reason: %q\n", nc.LastError())
    }),
)

To use the jetstream package, first connect to NATS using the standard nats.Connect method, then initialize the JetStream management interface using jetstream.New(nc). Most API calls in this package require a context.Context for handling timeouts and cancellations.

Common workflow:

1. Create a JetStream management interface.
2. Create or update a stream using js.CreateStream or js.UpdateStream.
3. Publish messages using js.Publish.
4. Create a consumer (e.g., durable) using s.CreateOrUpdateConsumer.
5. Consume messages using c.Fetch(), c.Consume() (callback-based), or c.Messages() (iterator-based).

package main

import (
    "context"
    "fmt"
    "strconv"
    "time"

    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

func main() {
    // In the `jetstream` package, almost all API calls rely on `context.Context` for timeout/cancellation handling
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    nc, _ := nats.Connect(nats.DefaultURL)

    // Create a JetStream management interface
    js, _ := jetstream.New(nc)

    // Create a stream
    s, _ := js.CreateStream(ctx, jetstream.StreamConfig{
        Name:     "ORDERS",
        Subjects: []string{"ORDERS.*"},
    })

    // Publish some messages
    for i := 0; i < 100; i++ {
        js.Publish(ctx, "ORDERS.new", []byte("hello message "+strconv.Itoa(i)))
        fmt.Printf("Published hello message %d\n", i)
    }

    // Create durable consumer
    c, _ := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
        Durable:   "CONS",
        AckPolicy: jetstream.AckExplicitPolicy,
    })

    // Get 10 messages from the consumer
    messageCounter := 0
    msgs, err := c.Fetch(10)
    if err != nil {
        // handle error
    }

    for msg := range msgs.Messages() {
        msg.Ack()
        fmt.Printf("Received a JetStream message via fetch: %s\n", string(msg.Data()))
        messageCounter++
    }

    fmt.Printf("Received %d messages\n", messageCounter)

    if msgs.Error() != nil {
        fmt.Println("Error during Fetch(): ", msgs.Error())
    }

    // Receive messages continuously in a callback
    cons, _ := c.Consume(func(msg jetstream.Msg) {
        msg.Ack()
        fmt.Printf("Received a JetStream message via callback: %s\n", string(msg.Data()))
        messageCounter++
    })
    defer cons.Stop()

    // Iterate over messages continuously
    it, _ := c.Messages()
    for i := 0; i < 10; i++ {
        msg, _ := it.Next()
        msg.Ack()
        fmt.Printf("Received a JetStream message via iterator: %s\n", string(msg.Data()))
        messageCounter++
    }
    it.Stop()

    // block until all 100 published messages have been processed
    for messageCounter < 100 {
        time.Sleep(10 * time.Millisecond)
    }
}