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

---

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

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

	// 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")
)

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