loafernatsx

README

loafer-natsx

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.

---

Philosophy

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

---

Installation

go get github.com/silviolleite/loafer-natsx

Requirements:

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

---

Architecture

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

High-Level Architecture Diagram

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

---

Broker

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.

Available Metrics

Metric	Type	Labels	Description
loafer_requests_total	Counter	subject	Total number of processed messages
loafer_errors_total	Counter	subject	Total number of handler errors
loafer_request_duration_seconds	Histogram	subject	Duration of message handler execution
loafer_inflight	Gauge	subject	Number of handlers currently being executed

---

Middleware

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.

Prometheus (middleware.Metrics)

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))).

OpenTelemetry (middleware.OTel)

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.

---

Typed Package

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.

Usage

Typed example

---

Dead Letter Queue (DLQ)

When enabled for JetStream routes:

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

---

Deduplication

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

---

Graceful Shutdown

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

---

Examples

See the examples directory:

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

---

Contributing

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

Prerequisites

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

Getting Started

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
   

Commit Message Convention

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.

Available Make Targets

Target	Description
make configure	Install all dev tools and git hooks
make test	Run tests with race detection and coverage
make lint	Format code and run golangci-lint
make cover	Generate coverage report
make cover-html	Generate HTML coverage report

Pull Request Process

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

---

License

MIT

Documentation

Index

• Constants
• type Err
  • func (e Err) Error() string
• type NakWithDelayError
  • func (e NakWithDelayError) Error() string

Constants

const (
	// ErrUnsupportedType indicates an error when an unsupported router type is encountered.
	ErrUnsupportedType = Err("unsupported router type")

	// ErrMissingURL indicates that a connection URL is required but was not provided.
	ErrMissingURL = Err("connection URL is required")

	// ErrMissingSubject indicates an error when the required subject is not provided.
	ErrMissingSubject = Err("subject is required")

	// ErrMissingMessage indicates that a request operation received a nil message.
	ErrMissingMessage = Err("message is required")

	// ErrMissingQueueGroup indicates an error when a queue group is required but not provided for the router.
	ErrMissingQueueGroup = Err("queue group is required for the router")

	// ErrMissingStream indicates an error when a stream is required but not provided for a jetstream router.
	ErrMissingStream = Err("stream is required for jetstream router")

	// ErrMissingDurable indicates an error when a durable name is required but not provided for a jetstream router.
	ErrMissingDurable = Err("durable name is required for jetstream router")

	// ErrNilRoute indicates that the provided route instance is nil, which is invalid for route registration.
	ErrNilRoute = Err("route cannot be nil")

	// ErrNilHandler indicates that the provided handler instance is nil, which is invalid for route registration.
	ErrNilHandler = Err("handler cannot be nil")

	// ErrNoRoutes indicates that no routes were provided when attempting to configure or run the broker.
	ErrNoRoutes = Err("no routes provided")

	// ErrNilRouteRegistration indicates that a route registration provided to the broker is nil, which is not allowed.
	ErrNilRouteRegistration = Err("route registration cannot be nil")

	// ErrRequestNotSupported indicates that request-reply routes are not supported for JetStream producers.
	ErrRequestNotSupported = Err("request-reply routes are not supported for JetStream producers")

	// ErrRequestTimeout indicates that a request-reply operation exceeded its deadline.
	ErrRequestTimeout = Err("request timeout: consumer did not reply in time")

	// ErrPermanentFailure indicates that the handler failed with a permanent (non-retryable) error.
	// When a handler returns an error wrapping ErrPermanentFailure, the message is acknowledged
	// (Ack) to prevent further redelivery attempts. Use this when the failure cannot be fixed
	// by retrying (e.g. malformed payload, business rule violation, missing precondition).
	ErrPermanentFailure = Err("permanent failure: message acknowledged without retry")

	// ErrSendToDLQ indicates that the handler explicitly requests the message to be routed
	// directly to the Dead Letter Queue, bypassing the normal retry flow. This only has effect
	// on JetStream routes with DLQ enabled; on other route types it behaves like a regular error.
	ErrSendToDLQ = Err("send to dead letter queue: skip retries")
)

Types

type Err

type Err string

Err represents an error as a string type and implements the error interface.

func (Err) Error

func (e Err) Error() string

Error returns an error message as a string

type NakWithDelayError

type NakWithDelayError struct {
	// Delay is the duration the server should wait before redelivering the message.
	Delay time.Duration
}

NakWithDelayError instructs the JetStream consumer to negatively acknowledge the message and request redelivery after the specified delay. This enables progressive backoff strategies without exposing the underlying jetstream.Msg to the handler.

Usage:

return nil, loafernatsx.NakWithDelayError{Delay: 30 * time.Second}
return nil, fmt.Errorf("context: %w", loafernatsx.NakWithDelayError{Delay: 5 * time.Minute})

On non-JetStream routes this error is treated as a regular error (logged, no special ack behavior).

func (NakWithDelayError) Error

func (e NakWithDelayError) Error() string

Error implements the error interface.