Lesson 5 · Publishing, consuming & delivery

The golib & the connection

One shared client wraps all of NATS for the repo — and it connects lazily, the first time a service actually asks for it.

Your win: describe the JetStreamManagement wrapper, explain when and how the NATS connection opens, and say how a service opts into NATS at all.

One wrapper for the whole repo

No service talks to raw nats.go. Everything goes through a single house client, JetStreamManagement (internal/golibs/nats/jetstream.go) — an interface that bundles every NATS operation the repo needs:

type JetStreamManagement interface {
    ConnectToJS()                                         // open the connection
    UpsertStream(...) / UpsertConsumer(...)               // provision (Lessons 3–4)
    PublishContext / PublishAsyncContext / TracedPublish… // publish (Lesson 6)
    Subscribe / QueueSubscribe / PullSubscribe            // consume (Lesson 7)
    RegisterReconnectHandler / RegisterDisconnectErrHandler / Close
}

So the wrapper is where all the repo-specific behaviour lives — the tenancy envelope, the tracing, the metrics, the idempotent upserts — and services just call these methods. The rest of Part 2 is really "what these methods do."

Opening the connection

ConnectToJS dials the server and turns the raw connection into a JetStream context:

conn, _ := nats.Connect(url,
    nats.UserInfo(user, password),        // per-service credentials (NATS ACLs, Lesson 12)
    nats.ReconnectWait(...), nats.MaxReconnects(...),
    nats.DisconnectErrHandler(...), nats.ReconnectHandler(...),   // survive blips
    nats.Timeout(5*time.Second))
js, _ := conn.JetStream(nats.PublishAsyncMaxPending(200000))      // ← big async-publish buffer

Two details worth keeping: it registers reconnect/disconnect handlers so a brief network blip doesn't kill the service (and it flips a Prometheus disconnect gauge), and it sets PublishAsyncMaxPending(200000) — a large buffer of in-flight async publishes (Lesson 6).

Lazy connect — on first use

You never call ConnectToJS yourself. You ask the resource container for the client — rsc.NATS() — and it connects the first time, guarded by a sync.Once (exactly the lazy pattern from the Go-techstack course):

func (r *Resources) GetNATS() (nats.JetStreamManagement, error) {
    if r.natsjs == nil {
        r.natsjsOnce.Do(func() {              // ← runs once, ever
            natsjs, _ := r.natsjser.NewJetStreamManagement(l, c)  // build (doesn't connect)
            natsjs.ConnectToJS()                    // ← the real connection, on first call
            r.natsjs = natsjs
        })
    }
    return r.natsjs, nil
}

So the connection opens the first time any code calls rsc.NATS() — a publisher, a subscriber registration, or an interceptor. (rsc.NATS() panics if NATS config is missing — fail-fast at boot, like every Resources client.)

How a service opts into NATS

Check yourself (from memory)

1. nats.go. In-repo: internal/golibs/nats/jetstream.go:67-83,325-350, internal/golibs/bootstrap/resource.go:465-481.