### Install NATS Go Client

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Use 'go get' to install the latest released version or a specific version of the NATS Go client. Also shows how to get the latest major version of the NATS Server.

```bash
go get github.com/nats-io/nats.go@latest
```

```bash
go get github.com/nats-io/nats.go@v1.52.0
```

```bash
go get github.com/nats-io/nats-server/v2@latest
```

--------------------------------

### Initialize Legacy JetStream Context

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Examples of initializing the legacy JetStream context with default, domain, or custom API prefix options.

```go
js, _ := nc.JetStream()

// With domain
js, _ := nc.JetStream(nats.Domain("hub"))

// With custom API prefix
js, _ := nc.JetStream(nats.APIPrefix("myprefix"))
```

--------------------------------

### Legacy KeyValue Store Operations

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Demonstrates legacy KeyValue store operations including creation, putting a value, getting a value, and setting up a watcher.

```go
js, _ := nc.JetStream()
kv, _ := js.CreateKeyValue(&nats.KeyValueConfig{
    Bucket: "profiles",
})

kv.Put("sue.color", []byte("blue"))
entry, _ := kv.Get("sue.color")
fmt.Println(string(entry.Value()))

watcher, _ := kv.Watch("sue.*")
defer watcher.Stop()
```

--------------------------------

### Import nats.go micro package

Source: https://github.com/nats-io/nats.go/blob/main/micro/README.md

Import the micro package to start building microservices. No specific setup is required beyond this import.

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

--------------------------------

### Initialize New JetStream Context

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Examples of initializing the new JetStream context using the jetstream package with default, domain, or custom API prefix options.

```go
js, _ := jetstream.New(nc)

// With domain
js, _ := jetstream.NewWithDomain(nc, "hub")

// With custom API prefix
js, _ := jetstream.NewWithAPIPrefix(nc, "myprefix")
```

--------------------------------

### Manage Consumers via Stream Interface

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Shows how to manage consumers (create, get, delete) directly through the Stream interface. This is an alternative to using the top-level JetStream interface for consumer management.

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

// get stream handle
stream, _ := js.Stream(ctx, "ORDERS")

// create consumer
cons, _ := stream.CreateConsumer(ctx, jetstream.ConsumerConfig{
    Durable:   "foo",
    AckPolicy: jetstream.AckExplicitPolicy,
})

// get consumer handle
cons, _ = stream.Consumer(ctx, "foo")

// delete a consumer
stream.DeleteConsumer(ctx, "foo")
```

--------------------------------

### Connect with Custom Root CAs for TLS

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect using TLS with a self-signed certificate by providing a tls.Config with RootCAs setup using the RootCAs helper.

```go
nc, err = nats.Connect("tls://localhost:4443", nats.RootCAs("./configs/certs/ca.pem"))
```

--------------------------------

### Get JetStream Stream Information in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Shows how to fetch the latest stream information from the server or retrieve cached information without making an API call.

```go
// Fetches the latest stream info from server
info, _ := s.Info(ctx)
fmt.Println(info.Config.Name)

// Returns the most recently fetched StreamInfo, without making an API call to the server
cachedInfo := s.CachedInfo()
fmt.Println(cachedInfo.Config.Name)
```

--------------------------------

### Create, Update, Fetch, and Delete Pull Consumers

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates CRUD operations for pull consumers using the JetStream interface. Use CreateConsumer for new consumers and Consumer to get a handle to an existing one. CreateOrUpdateConsumer is also available.

```go
js, _ := jetstream.New(nc)

// create a consumer (this is an idempotent operation)
// an error will be returned if consumer already exists and has different configuration.
cons, _ := js.CreateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    Durable: "foo",
    AckPolicy: jetstream.AckExplicitPolicy,
})

// create an ephemeral pull consumer by not providing `Durable`
ephemeral, _ := js.CreateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    AckPolicy: jetstream.AckExplicitPolicy,
})


// consumer can also be created using CreateOrUpdateConsumer
// this method will either create a consumer if it does not exist
// or update existing consumer (if possible)
cons2 := js.CreateOrUpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    Name: "bar",
})

// consumers can be updated
// an error will be returned if consumer with given name does not exist
// or an illegal property is to be updated (e.g. AckPolicy)
updated, _ := js.UpdateConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    AckPolicy: jetstream.AckExplicitPolicy,
    Description: "updated consumer",
})

// get consumer handle
cons, _ = js.Consumer(ctx, "ORDERS", "foo")

// delete a consumer
js.DeleteConsumer(ctx, "ORDERS", "foo")
```

--------------------------------

### Get Stream Handle and Perform Operations

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Obtain a stream handle first, then use it to perform operations like purging or retrieving messages. This pattern is used for stream-specific actions.

```go
s, _ := js.Stream(ctx, "ORDERS")
s.Purge(ctx)
msg, _ := s.GetMsg(ctx, 100)
```

--------------------------------

### Create and Configure a Push Consumer

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Set up a push consumer with `CreateOrUpdatePushConsumer`. Configure `AckPolicy`, `FilterSubject`, and `IdleHeartbeat`. For push consumers, `IdleHeartbeat` must be set at the consumer level.

```go
cons, _ := js.CreateOrUpdatePushConsumer(ctx, "ORDERS", jetstream.ConsumerConfig{
    DeliverSubject: nats.NewInbox()
    AckPolicy: jetstream.AckExplicitPolicy,
    // receive messages from ORDERS.A subject only
    FilterSubject: "ORDERS.A",
    // unlike pull consumers, idle heartbeat is configured on the consumer level
    IdleHeartbeat: 30 * time.Second
})
```

--------------------------------

### Connect with Client Certificate for TLS

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect using TLS and provide a client certificate and key using the ClientCert helper function.

```go
cert := nats.ClientCert("./configs/certs/client-cert.pem", "./configs/certs/client-key.pem")
nc, err = nats.Connect("tls://localhost:4443", cert)
```

--------------------------------

### Basic JetStream Usage in Go

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Shows how to connect to a NATS server, create a JetStream context, and manage streams and consumers for persistent messaging. Includes consuming messages with acknowledgments.

```go
// connect to nats server
nc, _ := nats.Connect(nats.DefaultURL)

// create jetstream context from nats connection
js, _ := jetstream.New(nc)

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
deffer cancel()

// get existing stream handle
stream, _ := js.Stream(ctx, "foo")

// retrieve consumer handle from a stream
cons, _ := stream.Consumer(ctx, "cons")

// consume messages from the consumer in callback
cc, _ := cons.Consume(func(msg jetstream.Msg) {
    fmt.Println("Received jetstream message: ", string(msg.Data()))
    msg.Ack()
})
defer cc.Stop()
```

--------------------------------

### Get NATS JetStream Bucket Status

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

The `Status` operation provides current information about a bucket, including its name and total size in bytes. This is useful for monitoring storage usage.

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

os.PutString(ctx, "config-1", "cfg1")
os.PutString(ctx, "config-2", "cfg1")
os.PutString(ctx, "config-3", "cfg1")

status, _ := os.Status(ctx)

fmt.Println(status.Bucket()) // prints `configs`
fmt.Println(status.Size()) // prints the size of the bucket in bytes
```

--------------------------------

### Get KeyValue Bucket Status

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Retrieve the current status of a KeyValue bucket, including its name, number of values, and total size in bytes. Useful for monitoring bucket usage.

```go
js, _ := jetstream.New(nc)
ctx := context.Background()
kv, _ := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "profiles"})

kv.Put(ctx, "sue.color", []byte("blue"))
kv.Put(ctx, "sue.age", []byte("43"))
kv.Put(ctx, "bucket", []byte("profiles"))

status, _ := kv.Status(ctx)

fmt.Println(status.Bucket()) // prints `profiles`
fmt.Println(status.Values()) // prints `3`
fmt.Println(status.Bytes()) // prints the size of all values in bytes
```

--------------------------------

### Legacy Pull Subscribe with Fetch Loop

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Demonstrates the older method of setting up a pull subscription and continuously fetching messages in a loop. Ensure proper error handling for production use.

```go
sub, _ := js.PullSubscribe("ORDERS.*", "processor")

for {
    msgs, _ := sub.Fetch(10, nats.MaxWait(5*time.Second))
    for _, msg := range msgs {
        fmt.Printf("Received: %s\n", string(msg.Data))
        msg.Ack()
    }
}
```

--------------------------------

### Connect with Direct Nkey and Signature Callback

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect directly using a public Nkey and a custom signature callback function.

```go
nc, err := nats.Connect(serverUrl, nats.Nkey(pubNkey, sigCB))
```

--------------------------------

### Basic Object Store CRUD Operations in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates creating an object store, putting objects as strings, bytes, or files, retrieving objects, and deleting objects and the store. Ensure JetStream is initialized.

```go
js, _ := jetstream.New(nc)
ctx := context.Background()

// Create a new bucket. Bucket name is required and has to be unique within a JetStream account.
os, _ := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "configs"})

config1 := bytes.NewBufferString("first config")
// Put an object in a bucket. Put expects an object metadata and a reader
// to read the object data from.
os.Put(ctx, jetstream.ObjectMeta{Name: "config-1"}, config1)

// Objects can also be created using various helper methods

// 1. As raw strings
os.PutString(ctx, "config-2", "second config")

// 2. As raw bytes
os.PutBytes(ctx, "config-3", []byte("third config"))

// 3. As a file
os.PutFile(ctx, "config-4.txt")

// Get an object
// Get returns a reader and object info
// Similar to Put, Get can also be used with helper methods
// to retrieve object data as a string, bytes or to save it to a file
object, _ := os.Get(ctx, "config-1")
data, _ := io.ReadAll(object)
info, _ := object.Info()

// Prints `configs.config-1 -> "first config"`
fmt.Printf("%s.%s -> %q\n", info.Bucket, info.Name, string(data))

// Delete an object.
// Delete will remove object data from stream, but object metadata will be kept
// with a delete marker.
os.Delete(ctx, "config-1")

// getting a deleted object will return an error
_, err := os.Get(ctx, "config-1")
fmt.Println(err) // prints `nats: object not found`

// A bucket can be deleted once it is no longer needed
js.DeleteObjectStore(ctx, "configs")
```

--------------------------------

### Get and Delete JetStream Messages in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Illustrates how to retrieve specific messages (by sequence number or last message for a subject) and delete messages from a JetStream stream using their sequence numbers.

```go
// get message from stream with sequence number == 100
msg, _ := s.GetMsg(ctx, 100)

// get last message from "ORDERS.new" subject
msg, _ = s.GetLastMsgForSubject(ctx, "ORDERS.new")

// delete a message with sequence number == 100
_ = s.DeleteMsg(ctx, 100)
```

--------------------------------

### Connect with Nkey Seed File

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect using an Nkey seed file. The NkeyOptionFromSeed helper loads and decodes the seed for signing.

```bash
> cat seed.txt
# This is my seed nkey!
SUAGMJH5XLGZKQQWAWKRZJIGMOU4HPFUYLXJMXOO5NLFEO2OOQJ5LPRDPM
```

```go
opt, err := nats.NkeyOptionFromSeed("seed.txt")
nc, err := nats.Connect(serverUrl, opt)
```

--------------------------------

### New Push Consumer with Consume

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Demonstrates creating a push consumer using the new API and processing messages with `Consume()`. This requires `DeliverSubject` in the configuration.

```go
cons, _ := s.CreateOrUpdatePushConsumer(ctx, jetstream.ConsumerConfig{
    Durable:        "processor",
    FilterSubject:  "ORDERS.*",
    DeliverSubject: "deliver.orders",
    IdleHeartbeat:  30 * time.Second,
})

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

--------------------------------

### Iterate Over Incoming JetStream Messages

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Use `Messages()` to get an iterator for incoming JetStream messages. Handle potential errors from `iter.Next()` which can occur if the iterator is closed or heartbeats are missed. Remember to call `iter.Stop()` when done.

```go
iter, _ := cons.Messages()
for {
    msg, err := iter.Next()
    // Next can return error, e.g. when iterator is closed or no heartbeats were received
    if err != nil {
        //handle error
    }
    fmt.Printf("Received a JetStream message: %s\n", string(msg.Data()))
    msg.Ack()
}
iter.Stop()
```

--------------------------------

### Connect with Custom TLS Configuration

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect using a custom tls.Config, specifying server name, client certificates, root CAs, and minimum TLS version.

```go
certFile := "./configs/certs/client-cert.pem"
keyFile := "./configs/certs/client-key.pem"
cert, err := tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
    t.Fatalf("error parsing X509 certificate/key pair: %v", err)
}

config := &tls.Config{
    ServerName: 	opts.Host,
    Certificates: 	[]tls.Certificate{cert},
    RootCAs:    	pool,
    MinVersion: 	tls.VersionTLS12,
}

nc, err = nats.Connect("nats://localhost:4443", nats.Secure(config))
if err != nil {
	t.Fatalf("Got an error on Connect with Secure Options: %+v\n", err)
}
```

--------------------------------

### Purge JetStream Stream Messages in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Provides examples for purging messages from a JetStream stream, including purging all messages, messages for a specific subject, up to a sequence number, or keeping a certain number of the newest messages.

```go
// remove all messages from a stream
_ = s.Purge(ctx)

// remove all messages from a stream that are stored on a specific subject
_ = s.Purge(ctx, jetstream.WithPurgeSubject("ORDERS.new"))

// remove all messages up to specified sequence number
_ = s.Purge(ctx, jetstream.WithPurgeSequence(100))

// remove messages, but keep 10 newest
_ = s.Purge(ctx, jetstream.WithPurgeKeep(10))
```

--------------------------------

### Basic JetStream Usage in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates publishing and consuming messages using JetStream with explicit acknowledgments and callbacks. Requires a running NATS server with JetStream enabled.

```go
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)
    }
}
```

--------------------------------

### Create a NATS Microservice

Source: https://github.com/nats-io/nats.go/blob/main/micro/README.md

Create a new microservice instance using `micro.AddService`. This requires a NATS connection and service configuration, including name, version, and base endpoint.

```go
nc, _ := nats.Connect(nats.DefaultURL)

// request handler
echoHandler := func(req micro.Request) {
    req.Respond(req.Data())
}

srv, err := micro.AddService(nc, micro.Config{
    Name:        "EchoService",
    Version:     "1.0.0",
    // base handler
    Endpoint: &micro.EndpointConfig{
        Subject: "svc.echo",
        Handler: micro.HandlerFunc(echoHandler),
    },
})
```

--------------------------------

### Legacy NATS Object Store Usage

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Demonstrates the legacy pattern for creating and interacting with a NATS Object Store without context.

```go
js, _ := nc.JetStream()
os, _ := js.CreateObjectStore(&nats.ObjectStoreConfig{
    Bucket: "configs",
})

os.PutString("config-1", "data")
result, _ := os.Get("config-1")
data, _ := io.ReadAll(result)
```

--------------------------------

### List JetStream Streams and Names in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates how to list all available JetStream streams and their names using the Go client. Handles potential errors during the listing process.

```go
// list streams
streams := js.ListStreams(ctx)
for s := range streams.Info() {
    fmt.Println(s.Config.Name)
}
if streams.Err() != nil {
    fmt.Println("Unexpected error occurred")
}

// list stream names
names := js.StreamNames(ctx)
for name := range names.Name() {
    fmt.Println(name)
}
if names.Err() != nil {
    fmt.Println("Unexpected error occurred")
}
```

--------------------------------

### Basic NATS Client Operations in Go

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Demonstrates fundamental NATS client operations including connecting, publishing, asynchronous and synchronous subscribing, request/reply patterns, and connection management (drain/close).

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

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

// Simple Publisher
nc.Publish("foo", []byte("Hello World"))

// Simple Async Subscriber
nc.Subscribe("foo", func(m *nats.Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
})

// Responding to a request message
nc.Subscribe("request", func(m *nats.Msg) {
    m.Respond([]byte("answer is 42"))
})

// Simple Sync Subscriber
sub, err := nc.SubscribeSync("foo")
m, err := sub.NextMsg(timeout)

// Channel Subscriber
ch := make(chan *nats.Msg, 64)
sub, err := nc.ChanSubscribe("foo", ch)
msg := <- ch

// Unsubscribe
sub.Unsubscribe()

// Drain
sub.Drain()

// Requests
msg, err := nc.Request("help", []byte("help me"), 10*time.Millisecond)

// Replies
nc.Subscribe("help", func(m *nats.Msg) {
    nc.Publish(m.Reply, []byte("I can help!"))
})

// Drain connection (Preferred for responders)
// Close() not needed if this is called.
nc.Drain()

// Close connection
nc.Close()
```

--------------------------------

### Connect with User Credentials

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Use the UserCredentials helper to connect using a user credentials file. This method handles JWT and nonce signing automatically.

```go
nc, err := nats.Connect(url, nats.UserCredentials("user.creds"))
```

```go
nc, err := nats.Connect(url, nats.UserCredentials("user.jwt", "user.nk"))
```

--------------------------------

### New Ordered Consumer with Consume

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Demonstrates creating an ordered consumer using the new API and processing messages with the `Consume()` method. This ensures strictly ordered, gap-free delivery.

```go
cons, _ := js.OrderedConsumer(ctx, "ORDERS", jetstream.OrderedConsumerConfig{
    FilterSubjects: []string{"ORDERS.*"},
})

// Use the same consumption methods as regular consumers
cc, _ := cons.Consume(func(msg jetstream.Msg) {
    fmt.Printf("Received: %s\n", string(msg.Data()))
})
deffer cc.Stop()
```

--------------------------------

### Initialize NATS Connection

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Standard initialization for a NATS connection. Ensure the NATS server is running.

```go
import (
    "github.com/nats-io/nats.go"
    "github.com/nats-io/nats.go/jetstream"
)

nc, _ := nats.Connect(nats.DefaultURL)
```

--------------------------------

### Build NATS.go Project

Source: https://github.com/nats-io/nats.go/blob/main/CLAUDE.md

Compiles all Go packages within the project. Ensure you are in the project's root directory.

```bash
go build ./...
```

--------------------------------

### New JetStream KeyValue Store Operations

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Illustrates new KeyValue store operations using the jetstream package, requiring context for all methods and using updated configuration and creation methods.

```go
js, _ := jetstream.New(nc)
kv, _ := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{
    Bucket: "profiles",
})

kv.Put(ctx, "sue.color", []byte("blue"))
entry, _ := kv.Get(ctx, "sue.color")
fmt.Println(string(entry.Value()))

watcher, _ := kv.Watch(ctx, "sue.*")
defer watcher.Stop()
```

--------------------------------

### Basic JetStream Usage

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates the fundamental steps for connecting to NATS, creating a JetStream context, creating a stream, publishing messages, and consuming messages using both fetch and callback methods.

```APIDOC
## Basic Usage

This section covers the essential steps for using the NATS JetStream client, including connection, stream creation, message publishing, and consumption.

### Connection and JetStream Initialization

Connect to a NATS server and create a JetStream management interface.

```go
package main

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

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

func main() {
    // Use context for timeouts and cancellation
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

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

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

    // ... rest of the code
}
```

### Stream Creation

Create a new stream to organize messages.

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

### Publishing Messages

Publish messages to a subject within the stream.

```go
    // 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)
    }
```

### Consuming Messages (Fetch)

Create a durable consumer and fetch messages in batches.

```go
    // 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())
    }
```

### Consuming Messages (Callback)

Consume messages continuously using a callback function.

```go
    // 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()
```

### Consuming Messages (Iterator)

Iterate over messages using an iterator.

```go
    // 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)
    }
}
```
```

--------------------------------

### Create an Ordered Consumer

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates the creation of an ordered consumer, which guarantees strict message processing order. It supports filtering messages by subject.

```go
js, _ := jetstream.New(nc)

// create a consumer (this is an idempotent operation)
cons, _ := js.OrderedConsumer(ctx, "ORDERS", jetstream.OrderedConsumerConfig{
    // Filter results from "ORDERS" stream by specific subject
    FilterSubjects: []string{"ORDERS.A"},
})
```

--------------------------------

### Connect with Custom JWT and Signature Callbacks

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Manually provide callback handlers for JWT and signature challenges when connecting.

```go
nc, err := nats.Connect(url, nats.UserJWT(jwtCB, sigCB))
```

--------------------------------

### Legacy Push Consumer Subscription

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Shows the older method for setting up a push consumer, including durable name, deliver subject, and idle heartbeat. Push consumers only support `Consume()`.

```go
sub, _ := js.Subscribe("ORDERS.*", handler,
    nats.Durable("processor"),
    nats.DeliverSubject("deliver.orders"),
    nats.IdleHeartbeat(30*time.Second),
)
```

--------------------------------

### Manage Multiple NATS Connections

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Demonstrates establishing and using multiple independent NATS connections to different servers.

```go
// Multiple connections
nc1 := nats.Connect("nats://host1:4222")
nc2 := nats.Connect("nats://host2:4222")

nc1.Subscribe("foo", func(m *Msg) {
    fmt.Printf("Received a message: %s\n", string(m.Data))
})

nc2.Publish("foo", []byte("Hello World!"))
```

--------------------------------

### Legacy Ordered Consumer Subscription

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Illustrates the older way to subscribe to an ordered consumer. This method is less flexible than the new approach.

```go
sub, _ := js.Subscribe("ORDERS.*", handler, nats.OrderedConsumer())
```

--------------------------------

### New NATS Object Store Usage with Context

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Illustrates the new pattern for NATS Object Store operations, which includes context.Context for all methods.

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

os.PutString(ctx, "config-1", "data")
result, _ := os.Get(ctx, "config-1")
data, _ := io.ReadAll(result)
```

--------------------------------

### Run Static Analysis with Staticcheck

Source: https://github.com/nats-io/nats.go/blob/main/CLAUDE.md

Performs static analysis using the staticcheck tool, similar to how the CI pipeline does it. Requires the test-specific go_test.mod file.

```bash
staticcheck -modfile=go_test.mod ./...
```

--------------------------------

### New Fetch API for One-Off Batch

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Shows how to use the `Fetch()` method directly on a consumer for one-off batch retrieval. This method is non-blocking and returns a `FetchResult`. Prefer `Consume()` or `Messages()` for continuous processing.

```go
cons, _ := s.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
    Durable:       "processor",
    FilterSubject: "ORDERS.*",
})

// non-blocking, returns a `FetchResult` that provides messages and error
msgs, _ := cons.Fetch(10, jetstream.FetchMaxWait(5*time.Second))
for msg := range msgs.Messages() {
    fmt.Printf("Received: %s\n", string(msg.Data()))
    msg.Ack()
}
if msgs.Error() != nil {
    // handle error
}
```

--------------------------------

### Lint NATS.go JetStream Package

Source: https://github.com/nats-io/nats.go/blob/main/CLAUDE.md

Runs golangci-lint for linting specifically on the jetstream package, with a 5-minute timeout.

```bash
golangci-lint run --timeout 5m0s ./jetstream/...
```

--------------------------------

### Connect to NATS Cluster with Options

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect to a NATS cluster using a comma-separated list of server URLs. Options like MaxReconnects and ReconnectWait can be configured.

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

nc, err := nats.Connect(servers)

// Optionally set ReconnectWait and MaxReconnect attempts.
// This example means 10 seconds total per backend.
nc, err = nats.Connect(servers, nats.MaxReconnects(5), nats.ReconnectWait(2 * time.Second))
```

--------------------------------

### Add NATS Microservices

Source: https://github.com/nats-io/nats.go/blob/main/micro/README.md

Adds multiple instances of a NATS microservice with a defined configuration. Ensure the NATS server is running and accessible.

```go
nc, _ := nats.Connect("nats://localhost:4222")
echoHandler := func(req micro.Request) {
    req.Respond(req.Data())
}

config := micro.Config{
    Name:    "EchoService",
    Version: "1.0.0",
    Endpoint: &micro.EndpointConfig{
        Subject: "svc.echo",
        Handler: micro.HandlerFunc(echoHandler),
    },
}
for i := 0; i < 3; i++ {
    srv, err := micro.AddService(nc, config)
    if err != nil {
        log.Fatal(err)
    }
    defer srv.Stop()
}
```

--------------------------------

### Synchronous Publish (New API)

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Publish messages synchronously using the new API, which includes `context.Context` for cancellation and timeouts.

```go
ack, _ := js.Publish(ctx, "ORDERS.new", []byte("hello"))
```

```go
ack, _ = js.PublishMsg(ctx, &nats.Msg{
    Subject: "ORDERS.new",
    Data:    []byte("hello"),
})
```

--------------------------------

### Asynchronous Publish (New API)

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Publish messages asynchronously using the new API. This method returns immediately and does not accept a context.

```go
ackF, _ := js.PublishAsync("ORDERS.new", []byte("hello"))

select {
case ack := <-ackF.Ok():
    fmt.Println(ack.Sequence)
case err := <-ackF.Err():
    fmt.Println(err)
}

<-js.PublishAsyncComplete()
```

--------------------------------

### Run NATS.go Tests for Micro Package

Source: https://github.com/nats-io/nats.go/blob/main/CLAUDE.md

Executes all tests within the micro package with the race detector enabled.

```bash
go test -modfile=go_test.mod -race ./micro/... --failfast
```

--------------------------------

### Fetch Consumer Information

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Retrieves consumer configuration details. Use Info() to fetch the latest information from the server, or CachedInfo() for the most recently fetched data without a server call.

```go
// Fetches latest consumer info from server
info, _ := cons.Info(ctx)
fmt.Println(info.Config.Durable)

// Returns the most recently fetched ConsumerInfo, without making an API call to the server
cachedInfo := cons.CachedInfo()
fmt.Println(cachedInfo.Config.Durable)
```

--------------------------------

### Watching for Object Store Changes in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Sets up a watcher on an object store to receive notifications for object changes. By default, it sends current values, then nil, then updates. Use options like UpdatesOnly to filter.

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

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

// By default, watcher will return most recent values for all objects in a bucket.
// Watcher can be configured to only return updates by using jetstream.UpdatesOnly() option.
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"})

// First, the watcher sends most recent values for all matching objects.
// In this case, it will send a single entry for `config-1`.
object := <-watcher.Updates()
// Prints `configs.config-1 -> ""`
fmt.Printf("%s.%s -> %q\n", object.Bucket, object.Name, object.Description)

// After all current values have been sent, watcher will send nil on the channel.
object = <-watcher.Updates()
if object != nil {
    fmt.Println("Unexpected object received")
}

// After that, watcher will send updates when changes occur
// In this case, it will send an entry for `config-2` and `config-1`.
object = <-watcher.Updates()
// Prints `configs.config-2 -> ""`
fmt.Printf("%s.%s -> %q\n", object.Bucket, object.Name, object.Description)

object = <-watcher.Updates()
// Prints `configs.config-1 -> "updated config"`
fmt.Printf("%s.%s -> %q\n", object.Bucket, object.Name, object.Description)
```

--------------------------------

### Basic KeyValue Store Operations with NATS.go JetStream

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Demonstrates fundamental CRUD operations on a JetStream KeyValue store (bucket). Use for simple key-value data storage. Operations like `Update` and `Create` have specific revision and existence requirements.

```go
js, _ := jetstream.New(nc)
ctx := context.Background()

// Create a new bucket. Bucket name is required and has to be unique within a JetStream account.
kv, _ := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "profiles"})
```

```go
// Set a value for a given key
// Put will either create or update a value for a given key
kv.Put(ctx, "sue.color", []byte("blue"))
```

```go
// Get an entry for a given key
// Entry contains key/value, but also metadata (revision, timestamp, etc.))
entry, _ := kv.Get(ctx, "sue.color")

// Prints `sue.color @ 1 -> "blue"`
fmt.Printf("%s @ %d -> %q\n", entry.Key(), entry.Revision(), string(entry.Value()))
```

```go
// Update a value for a given key
// Update will fail if the key does not exist or the revision has changed
kv.Update(ctx, "sue.color", []byte("red"), 1)
```

```go
// Create will fail if the key already exists
_, err := kv.Create(ctx, "sue.color", []byte("purple"))
fmt.Println(err) // prints `nats: key exists`
```

```go
// Delete a value for a given key.
// Delete is not destructive, it will add a delete marker for a given key
// and all previous revisions will still be available
kv.Delete(ctx, "sue.color")
```

```go
// getting a deleted key will return an error
_, err = kv.Get(ctx, "sue.color")
fmt.Println(err) // prints `nats: key not found`
```

```go
// A bucket can be deleted once it is no longer needed
js.DeleteKeyValue(ctx, "profiles")
```

--------------------------------

### New Callback Consumption with Consume()

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

The recommended replacement for legacy callback subscriptions. It uses pull consumers for continuous message delivery with better flow control. Manual acknowledgment is implicit.

```go
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()
```

--------------------------------

### Run All NATS.go Tests

Source: https://github.com/nats-io/nats.go/blob/main/CLAUDE.md

Executes all tests with the race detector enabled, internal testing tags, and sequential execution. Uses the test-specific go_test.mod file.

```bash
go test -modfile=go_test.mod -race -v -p=1 ./... --failfast -vet=off -tags=internal_testing
```

--------------------------------

### Add Services with Custom Queue Groups

Source: https://github.com/nats-io/nats.go/blob/main/micro/README.md

Adds multiple services with unique queue groups to handle responses for a subject. Useful for distributing load or implementing fanout patterns.

```go
for i := 0; i < 5; i++ {
  srv, _ := micro.AddService(nc, micro.Config{
    Name:        "EchoService",
    Version:     "1.0.0",
    QueueGroup:  fmt.Sprintf("q-%d", i),
    // base handler
    Endpoint: &micro.EndpointConfig{
        Subject: "svc.echo",
        Handler: micro.HandlerFunc(echoHandler),
    },
  })
}
```

--------------------------------

### Subscribe with Wildcard Tokens

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Subscribe to subjects using wildcard tokens. '*' matches any single token, while '>' matches any length of the tail.

```go
// "*" matches any token, at any level of the subject.
nc.Subscribe("foo.*.baz", func(m *Msg) {
    fmt.Printf("Msg received on [%s] : %s\n", m.Subject, string(m.Data))
})

nc.Subscribe("foo.bar.*", func(m *Msg) {
    fmt.Printf("Msg received on [%s] : %s\n", m.Subject, string(m.Data))
})
```

```go
// ">" matches any length of the tail of a subject, and can only be the last token
// E.g. 'foo.>' will match 'foo.bar', 'foo.bar.baz', 'foo.foo.bar.bax.22'
nc.Subscribe("foo.>"), func(m *Msg) {
    fmt.Printf("Msg received on [%s] : %s\n", m.Subject, string(m.Data))
})
```

--------------------------------

### Generate Code Coverage Report

Source: https://github.com/nats-io/nats.go/blob/main/CLAUDE.md

Generates a code coverage report for the NATS.go project using the provided script.

```bash
./scripts/cov.sh
```

--------------------------------

### NATS Authentication with User Info and Token

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Connect to a NATS server that may require either username/password or token authentication by providing both options.

```go
// You can even pass the two at the same time in case one of the servers
// in the mesh requires token instead of user name and password.
nc, err = nats.Connect("nats://localhost:4222",
    nats.UserInfo("foo", "bar"),
    nats.Token("S3cretT0ken"))
```

--------------------------------

### Watch for KeyValue Bucket Changes

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Create a watcher to monitor changes in a KeyValue bucket. By default, it returns current values and then updates. Use jetstream.UpdatesOnly() to receive only new updates.

```go
js, _ := jetstream.New(nc)
ctx := context.Background()
kv, _ := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "profiles"})

kv.Put(ctx, "sue.color", []byte("blue"))

// A watcher can be created to watch for changes on a given key or the whole bucket
// By default, watcher will return most recent values for all matching keys.
// Watcher can be configured to only return updates by using jetstream.UpdatesOnly() option.
watcher, _ := kv.Watch(ctx, "sue.*")
defer watcher.Stop()

kv.Put(ctx, "sue.age", []byte("43"))
kv.Put(ctx, "sue.color", []byte("red"))

// First, the watcher sends most recent values for all matching keys.
// In this case, it will send a single entry for `sue.color`.
entry := <-watcher.Updates()
// Prints `sue.color @ 1 -> "blue"`
fmt.Printf("%s @ %d -> %q\n", entry.Key(), entry.Revision(), string(entry.Value()))

// After all current values have been sent, watcher will send nil on the channel.
entry = <-watcher.Updates()
if entry != nil {
    fmt.Println("Unexpected entry received")
}

// After that, watcher will send updates when changes occur
// In this case, it will send an entry for `sue.color` and `sue.age`.

entry = <-watcher.Updates()
// Prints `sue.age @ 2 -> "43"`
fmt.Printf("%s @ %d -> %q\n", entry.Key(), entry.Revision(), string(entry.Value()))

entry = <-watcher.Updates()
// Prints `sue.color @ 3 -> "red"`
fmt.Printf("%s @ %d -> %q\n", entry.Key(), entry.Revision(), string(entry.Value()))
```

--------------------------------

### Run tests with go_test.mod

Source: https://github.com/nats-io/nats.go/blob/main/CONTRIBUTING.md

Execute tests using a specific modfile for testing dependencies. This can be done by passing the -modfile flag directly to go test or by setting the GOFLAGS environment variable.

```shell
go test ./... -modfile=go_test.mod
```

--------------------------------

### Retrieve Specific Service Instance Info

Source: https://github.com/nats-io/nats.go/blob/main/micro/README.md

Fetches detailed information for a specific NATS microservice instance using its service name and ID. The output is piped to `jq` for pretty-printing.

```sh
nats req '$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7' '' | jq

13:04:19 Sending request on "$SRV.INFO.EchoService.x3Yuiq7g7MoxhXdxk7i4K7"
13:04:19 Received with rtt 318.875µs
{
  "name": "EchoService",
  "id": "x3Yuiq7g7MoxhXdxk7i4K7",
  "version": "1.0.0",
  "metadata": {},
  "type": "io.nats.micro.v1.info_response",
  "description": "",
  "endpoints": [
    {
      "name": "default",
      "subject": "svc.echo",
      "queue_group": "q",
      "metadata": null
    }
  ]
}
```

--------------------------------

### NATS Connection Event Handlers

Source: https://github.com/nats-io/nats.go/blob/main/README.md

Set up handlers for disconnect, reconnect, and closed connection events to monitor and react to connection state changes.

```go
// Setup callbacks to be notified on disconnects, reconnects and connection closed.
nc, err = nats.Connect(servers,
	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())
	})
)
```

--------------------------------

### Create, Update, and Delete JetStream Streams in Go

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/README.md

Shows how to create, update, retrieve, and delete JetStream streams using the Go client. These operations are idempotent where applicable.

```go
js, _ := jetstream.New(nc)

// create a stream (this is an idempotent operation)
s, _ := js.CreateStream(ctx, jetstream.StreamConfig{
    Name:     "ORDERS",
    Subjects: []string{"ORDERS.*"},
})

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

// get stream handle
s, _ = js.Stream(ctx, "ORDERS")

// delete a stream
js.DeleteStream(ctx, "ORDERS")
```

--------------------------------

### Asynchronous Publish (Legacy API)

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

Publish messages asynchronously using the legacy API. This returns a future acknowledgment that can be polled for success or error.

```go
ackF, _ := js.PublishAsync("ORDERS.new", []byte("hello"))

select {
case ack := <-ackF.Ok():
    fmt.Println(ack.Sequence)
case err := <-ackF.Err():
    fmt.Println(err)
}

// Wait for all pending acks
<-js.PublishAsyncComplete()
```

--------------------------------

### Legacy Callback Subscription

Source: https://github.com/nats-io/nats.go/blob/main/jetstream/MIGRATION.md

This is the legacy method for subscribing to messages using a callback function. It requires manual acknowledgment.

```go
sub, _ := js.Subscribe("ORDERS.*", func(msg *nats.Msg) {
    fmt.Printf("Received: %s\n", string(msg.Data))
    msg.Ack()
}, nats.Durable("processor"), nats.ManualAck)
defer sub.Unsubscribe()
```