A structured, Go library for working with NATS and JetStream.

loafer-natsx provides a clean abstraction layer for:

• Core NATS publishing
• JetStream publishing with deduplication
• Route-based message consumption
• Durable consumers
• Retry and redelivery handling
• Dead Letter Queue (DLQ)
• Request--Reply patterns
• Historical replay
• Graceful shutdown handling
• Concurrent multi-route broker orchestration

The library is designed around explicit configuration, clear separation of concerns, and production-safe defaults.

The project follows these principles:

• Explicit configuration over hidden behavior
• Clear separation between Core NATS and JetStream concerns
• Route-driven consumption model
• Functional options pattern
• Sentinel errors for validation
• Context-aware shutdown
• Fail-fast orchestration
• Concurrency safety by design
• Production-grade resilience

go get github.com/silviolleite/loafer-natsx

Requirements:

• Go 1.26+
• NATS Server
• JetStream enabled for persistence features

The project is organized into focused packages:

• conn → Connection management
• producer → Core and JetStream producers
• router → Route definitions
• consumer → Message consumption engine
• broker → Multi-route concurrent orchestration
• logger → Logging abstraction
• typed → Generic type-safe wrappers for producers and handlers

                    ┌─────────────────────┐
                    │     Application     │
                    └──────────┬──────────┘
                               │
                      ┌────────▼────────┐
                      │      Broker     │
                      │  (Orchestrator) │
                      └────────┬────────┘
                               │
        ┌──────────────────────┼──────────────────────┐
        │                      │                      │
 ┌──────▼──────┐       ┌───────▼───────┐       ┌──────▼───────┐
 │   Router    │       │   Router      │       │   Router     │
 │ (Route A)   │       │ (Route B)     │       │ (Route N)    │
 └──────┬──────┘       └───────┬───────┘       └──────┬───────┘
        │                      │                      │
 ┌──────▼──────┐       ┌───────▼───────┐       ┌──────▼───────┐
 │  Consumer   │       │   Consumer    │       │   Consumer   │
 │ (Workers)   │       │  (Workers)    │       │  (Workers)   │
 └──────┬──────┘       └───────┬───────┘       └──────┬───────┘
        │                      │                      │
        └──────────────┬───────┴──────────────┬───────┘
                       │                      │
                 ┌─────▼─────┐          ┌─────▼──────┐
                 │   NATS    │          │ JetStream  │
                 │  (Core)   │          │ Persistence│
                 └───────────┘          └────────────┘

The broker package allows running multiple routes concurrently within a single service process.

It provides:

• Registration of multiple validated routes with handlers
• Configurable worker concurrency
• Coordinated startup of all routes
• Fail-fast behavior (if one route fails, all are stopped)
• Context propagation across all routes
• Global cancellation control
• Safe shutdown without partial execution states

The broker guarantees:

• Concurrency safety
• No goroutine leaks
• No silent route failures
• Coordinated lifecycle management
• Deterministic shutdown behavior

This enables building services that consume multiple subjects safely without risking inconsistent runtime states.

The broker supports Prometheus metrics out of the box via the WithMetrics option.

Observability is built on a small, composable middleware layer in the middleware package. A middleware wraps a handler with cross-cutting behavior while keeping the handler signature unchanged:

type Handler func(ctx context.Context, data []byte) (any, error)
type Middleware func(Handler) Handler

Middlewares are composed with middleware.Chain using first-is-outermost semantics and wired into the broker in two scopes:

• Global, applied to every route, via broker.WithGlobalMiddleware(...)
• Per route, applied to a single registration, via the optional variadic argument of broker.NewRouteRegistration(route, handler, mws...)

Global middlewares run outermost (first in, last out), then per-registration middlewares, then the user handler.

The package ships two backends out of the box, and any custom middleware.Middleware can be plugged in the same way — the library is not limited to Prometheus and OpenTelemetry.

Instruments processing with the loafer_* collectors listed above, labeled by subject. It registers collectors idempotently, so it is safe to build for multiple routes on the same registerer.

br := broker.New(nc, log,
    broker.WithGlobalMiddleware(
        middleware.Metrics(middleware.WithMetricsRegisterer(prometheus.DefaultRegisterer)),
    ),
)

broker.WithMetrics(reg) remains available as convenience sugar over WithGlobalMiddleware(middleware.Metrics(middleware.WithMetricsRegisterer(reg))).

Creates a SpanKindConsumer span named loafer.process/<subject> per message, extracts any trace context propagated through the NATS message headers, and sets the span status from the handler outcome.

br := broker.New(nc, log,
    broker.WithGlobalMiddleware(
        middleware.OTel(),                    // continue the incoming trace, or
        // middleware.OTel(middleware.WithLinkFromContext()), // start a new root linked to it
        middleware.Metrics(),
    ),
)

Options: WithTracerProvider, WithPropagator, and WithLinkFromContext (useful for long-lived consumers to avoid inheriting an unbounded producer trace while preserving causality through a span link).

See the middleware example.

The typed package provides compile-time type safety for producers and consumers using Go generics. It wraps the existing API with zero breaking changes.

It provides:

• Codec[T] interface for pluggable serialization (JSON, Protobuf, etc.)
• JSONCodec[T] built-in implementation using encoding/json
• Producer[T] typed wrapper with Publish method
• Requester[T, R] typed request-reply with automatic response decoding
• WrapHandler adapter from typed handler to consumer.HandlerFunc
• WrapReply adapter from typed ReplyFunc[R] to router.ReplyFunc

Applications opt-in gradually — existing raw []byte usage continues to work unchanged.

Typed example

When enabled for JetStream routes:

• Messages exceeding MaxDeliver are published to dlq.<subject>
• Headers include:
  • X-Error
  • X-Retry-Count

Deduplication occurs during publish when a MsgID is provided.

If another message with the same MsgID is published within the stream's duplicate window:

• The message is not stored again
• The server acknowledges the original sequence
• ack.Duplicate is set to true

All consumers and brokers respect context.Context.

When the context is canceled:

• Core subscriptions are drained
• JetStream consumers are stopped
• Broker cancels all routes
• Connections can be gracefully drained

See the examples directory:

https://github.com/silviolleite/loafer-natsx/tree/main/examples

We welcome contributions! Follow the steps below to set up your development environment.

• Go 1.26+
• Node.js (for commit linting via husky)
• Docker & Docker Compose (for local NATS server)

1. Clone the repository

   git clone https://github.com/silviolleite/loafer-natsx.git
   cd loafer-natsx

2. Set up the development environment

   This installs Go tools, Node dependencies, and configures git hooks for commit validation:

   make setup-dev

3. Run tests

   make test

4. Run linter

   make lint

This project uses Conventional Commits. All commits must follow this format:

type(scope?): subject

Examples:

• feat: add new consumer option
• fix(router): handle nil pointer on shutdown
• docs: update README
• chore: bump dependencies
• test: add coverage for producer

The git hook will reject commits that don't follow this convention.

1. Create a feature branch from main
2. Make your changes with properly formatted commit messages
3. Ensure all tests pass (make test)
4. Ensure linter passes (make lint)
5. Open a Pull Request

MIT