NATS Development Reference

Docs: https://docs.nats.io | Examples: https://natsbyexample.com

NATS is a subject-based connective layer for distributed systems. Messages route by subject string, not hostname:port. Server binary is ~20MB, runs on Raspberry Pi to cloud. CNCF project, Apache 2.0 licensed, 40+ client libraries. Max message payload: 1MB default (configurable up to 64MB; keep under ~8MB in practice).

Two planes — choose before designing:

Deep reference files in references/:

• jetstream.md — full StreamConfig/ConsumerConfig fields, async publish, KV/object-store ops
• server-deployment.md — full config options, Docker, Kubernetes, Go embedding struct
• security.md — NKey, JWT/operator model, nsc CLI, TLS, auth callout

Subjects

→ https://docs.nats.io/nats-concepts/subjects

service.orders.created      # exact (publishers always use exact subjects)
service.orders.*            # * = one token (subscribers only)
service.orders.>            # > = one or more tokens at end (subscribers only)

• Dot-separated hierarchy; max 16 tokens, <256 chars recommended
• Alphanumeric, -, _ only (avoid other special chars)
• $ prefix reserved for system use ($SYS.*, $JS.*, $KV.*, $SRV.*)
• Multiple overlapping subs on one connection → duplicate delivery per matching sub

Core NATS Patterns

→ https://docs.nats.io/nats-concepts/core-nats

Pub/Sub

Fan-out to all subscribers. Zero config — subjects are ephemeral.

nc.Publish("orders.created", data)
nc.Subscribe("orders.*", func(msg *nats.Msg) { /* handle */ })

→ https://docs.nats.io/nats-concepts/core-nats/pubsub

Request/Reply

Requester sends to a subject with a temp reply-to inbox (_INBOX.<nonce>). First responder wins.

msg, err := nc.Request("svc.lookup", payload, 2*time.Second)
// err == nats.ErrNoResponders when no subscriber (immediate 503)

nc.Subscribe("svc.lookup", func(msg *nats.Msg) { msg.Respond(result) })

→ https://docs.nats.io/nats-concepts/core-nats/reqreply

Queue Groups

Competitive consumers — one random member gets each message. Scale horizontally with zero reconfiguration. Geo-affinity: local consumers served first.

nc.QueueSubscribe("orders.created", "order-processors", handler)

→ https://docs.nats.io/nats-concepts/core-nats/queue

JetStream

→ https://docs.nats.io/nats-concepts/jetstream

Enable on server: add jetstream {} block or pass --jetstream flag.

Streams

Streams persistently capture Core NATS subjects. Configuration is separate from consumption.

js, _ := nc.JetStream()
js.AddStream(&nats.StreamConfig{
    Name:      "ORDERS",
    Subjects:  []string{"orders.>"},
    Storage:   nats.FileStorage,    // or MemoryStorage
    Replicas:  3,                   // clustered only; 1, 2, 3, or 5
    Retention: nats.LimitsPolicy,   // default
    MaxAge:    24 * time.Hour,
    MaxBytes:  1 << 30,
})

Retention policies (see references/jetstream.md for decision tree):

→ https://docs.nats.io/nats-concepts/jetstream/streams

Publishing to JetStream

Always use js.Publish() — not nc.Publish() — to receive server ack confirming storage.

ack, err := js.Publish("orders.created", data)
// ack.Stream, ack.Sequence confirm exactly where it was stored

// Exactly-once: include Nats-Msg-Id header (dedup window: 2 min default)
js.PublishMsg(&nats.Msg{
    Subject: "orders.created",
    Header:  nats.Header{"Nats-Msg-Id": []string{uniqueID}},
    Data:    data,
})

Consumers

Prefer pull consumers for new projects. Use push only for ordered replay with a single subscriber.

// Durable pull consumer
js.AddConsumer("ORDERS", &nats.ConsumerConfig{
    Durable:       "order-worker",
    FilterSubject: "orders.created",
    AckPolicy:     nats.AckExplicitPolicy,
    DeliverPolicy: nats.DeliverAllPolicy,
    MaxDeliver:    5,
    AckWait:       30 * time.Second,
})

sub, _ := js.PullSubscribe("orders.created", "order-worker")
msgs, _ := sub.Fetch(10, nats.MaxWait(5*time.Second))
for _, msg := range msgs {
    msg.Ack()  // or .Nak(), .InProgress(), .Term()
}

// Ordered push consumer (single subscriber, no ack, for replay/inspection)
sub, _ := js.SubscribeSync("orders.>", nats.OrderedConsumer())

Delivery policies: DeliverAllPolicy · DeliverLastPolicy · DeliverLastPerSubjectPolicy · DeliverNewPolicy · DeliverByStartSequencePolicy · DeliverByStartTimePolicy

Ack policies: AckExplicitPolicy (default) · AckNonePolicy · AckAllPolicy

→ https://docs.nats.io/nats-concepts/jetstream/consumers

Full consumer config fields → references/jetstream.md

Key/Value Store

Built on JetStream streams (prefix KV_). Immediately consistent; no read-your-writes guarantee on direct gets (use Watch for consistency). Valid key chars: alphanumeric + _, -, ., =, /.

kv, _ := js.CreateKeyValue(&nats.KeyValueConfig{
    Bucket:  "config",
    TTL:     1 * time.Hour,
    History: 5,     // keep last 5 revisions per key (default: 1)
})

kv.Put("flags.dark-mode", data)
entry, _ := kv.Get("flags.dark-mode")   // entry.Value(), .Revision()
kv.Delete("flags.dark-mode")
kv.Create("lock", data)               // compare-to-null-and-set; fails if exists
kv.Update("lock", newData, revision)  // CAS

watcher, _ := kv.Watch("flags.*")
for entry := range watcher.Updates() { /* nil = end of initial snapshot */ }

→ https://docs.nats.io/nats-concepts/jetstream/key-value-store

Full KV API → references/jetstream.md

Object Store

Chunked file storage on JetStream. Not a distributed filesystem — all objects must fit on the target node.

obs, _ := js.CreateObjectStore(&nats.ObjectStoreConfig{Bucket: "artifacts"})
obs.PutFile("model.bin", "/local/path/model.bin")
obs.GetFile("model.bin", "/dest/model.bin")
obs.Delete("model.bin")
watcher, _ := obs.Watch()

→ https://docs.nats.io/nats-concepts/jetstream/obj_store

Connection & Reconnection

→ https://docs.nats.io/using-nats/developer/connecting

nc, err := nats.Connect(
    "nats://s1:4222,nats://s2:4222",     // comma-separated cluster seeds
    nats.UserCredentials("app.creds"),   // JWT+NKey creds file
    // nats.Token("secret")             // or token
    // nats.NkeyOptionFromSeed("user.nk")
    nats.MaxReconnects(-1),             // -1 = infinite (default: 60 attempts)
    nats.ReconnectWait(2*time.Second),
    nats.ReconnectJitter(100*time.Millisecond, time.Second),
    nats.ReconnectBufSize(8<<20),       // 8MB buffer during reconnect
    nats.DisconnectErrHandler(onDisconnect),
    nats.ReconnectHandler(onReconnect),
    nats.ClosedHandler(onClose),
    nats.ErrorHandler(onAsyncError),
)
defer nc.Drain()  // flush pending, then close — never use nc.Close() in production

URL schemes: nats:// (opportunistic TLS) · tls:// (mandatory TLS) · ws:// · wss://

→ https://docs.nats.io/using-nats/developer/connecting/reconnect