nats

README

Forge NATS Transport

contrib/message/nats adapts core NATS pub/sub to github.com/sylphylabs/forge/transport/message.

Example:

client, err := nats.New(nats.WithURL("nats://127.0.0.1:4222"))
if err != nil {
    return err
}
defer client.Close()

server := message.NewServer(client)
if err := server.Handle("accounts.created", handleAccountCreated); err != nil {
    return err
}

app := forge.New(forge.WithServer(server))
return app.Run()

The adapter implements message.Publisher and message.Subscriber. It does not add NATS to the root Forge module.

Semantics

• Publish sends a core NATS message and then calls FlushWithContext. A successful return means the NATS server responded to the flush after the publish. It does not imply JetStream durability. A timeout or cancellation after the local publish is an ambiguous outcome: the server may already have received the message, so callers need an idempotency strategy before retrying.
• Subscribe binds the NATS subscription lifetime to the context passed by message.Server. Cancellation unsubscribes and stops later delivery.
• Handler errors are reported through WithErrorHandler; core NATS has no acknowledgement decision to return to the server.
• Request is provided as a NATS-specific helper. Request/reply is not part of the broker-neutral transport/message contract.
• Message.ID and Message.Key are carried in Forge-owned NATS headers. Other headers use the normalized multi-value metadata.Metadata model.

Subject Semantics

The two packages in this module treat destination differently, and the difference is not a configuration detail.

Core NATS: a wire address

A core NATS destination is the subject, passed to conn.Subscribe untouched. Tokens are separated by ..

	Syntax	Position
Single token	*	any token, and must occupy the whole token
Multi token	>	last token only

orders.* matches orders.created but not orders.created.eu; orders.> matches both. Matching is done entirely by the NATS server — the adapter neither parses, validates, nor rewrites the subject. The destination the core Handler receives is the concrete published subject, not the pattern the subscription was registered with:

// handler receives "orders.created"
client.Subscribe(ctx, "orders.*", handler)

A subject in another broker's syntax is passed through as-is, so the server decides: orders/# is a single token containing slashes and simply matches nothing.

JetStream: a logical name

The jetstream subpackage destination is not a subject. It is a key into the bindings map given to NewSubscriber, resolving to the {Stream, Consumer} pair to attach to:

jetstream.NewSubscriber(js, map[string]jetstream.Binding{
    "orders.created": {Stream: "ORDERS", Consumer: "order-worker"},
})

Wildcards are therefore not declared here at all — the subjects that reach a subscription come from the consumer's FilterSubject, which is external deployment state the adapter never reads. Passing orders.* is a missing map key, and Subscribe fails with ErrBindingNotFound. See the jetstream README.

This package remains the ephemeral core NATS adapter. Applications that need durable storage, explicit acknowledgement, redelivery, and duplicate detection use the jetstream subpackage. JetStream is not only a server configuration switch: publishers must wait for PubAck, consumers must bind to named durable consumers, and handlers need explicit ack/nack/term behavior.

Documentation

Overview

Package nats adapts core NATS publish/subscribe to transport/message.

Index

• Constants
• Variables
• type Client
  • func New(opts ...Option) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Publish(ctx context.Context, subject string, msg *message.Message) error
  • func (c *Client) Request(ctx context.Context, subject string, msg *message.Message) (*message.Message, error)
  • func (c *Client) Subscribe(ctx context.Context, subject string, handler message.Handler) (message.Subscription, error)
• type ErrorHandler
• type Option
  • func WithConn(conn *natsgo.Conn) Option
  • func WithConnectOptions(opts ...natsgo.Option) Option
  • func WithErrorHandler(handler ErrorHandler) Option
  • func WithFlushTimeout(timeout time.Duration) Option
  • func WithQueue(queue string) Option
  • func WithURL(url string) Option

Constants

const (
	// HeaderMessageID carries message.Message.ID through core NATS headers.
	HeaderMessageID = "forge-message-id"
	// HeaderMessageKey carries message.Message.Key through core NATS headers.
	HeaderMessageKey = "forge-message-key"
)

Variables

var (
	// ErrNilContext reports an operation started without a context.
	ErrNilContext = errors.New("nats: nil context")
	// ErrEmptySubject reports an invalid NATS subject.
	ErrEmptySubject = errors.New("nats: empty subject")
	// ErrEmptyURL reports an invalid NATS server URL.
	ErrEmptyURL = errors.New("nats: empty url")
	// ErrNilMessage reports an invalid message.
	ErrNilMessage = errors.New("nats: nil message")
	// ErrNilHandler reports an invalid subscription handler.
	ErrNilHandler = errors.New("nats: nil handler")
	// ErrNilConn reports an invalid connection option.
	ErrNilConn = errors.New("nats: nil connection")
	// ErrClosed reports an adapter closed by its owner.
	ErrClosed = errors.New("nats: client closed")
)

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client adapts one NATS connection to Forge message transport.

func New

func New(opts ...Option) (*Client, error)

New creates a NATS message adapter. Without WithConn, the returned client owns the connection and closes it from Close.

func (*Client) Close

func (c *Client) Close() error

Close closes an adapter-owned connection. Application-owned connections supplied with WithConn are left open.

func (*Client) Publish

func (c *Client) Publish(ctx context.Context, subject string, msg *message.Message) error

Publish sends one core NATS message and flushes the connection with the caller's context. A successful return means the server responded to the flush after the publish; it does not imply durable JetStream storage.

func (*Client) Request

func (c *Client) Request(ctx context.Context, subject string, msg *message.Message) (*message.Message, error)

Request sends a NATS request/reply message. It is intentionally adapter specific; request/reply is not part of the broker-neutral message contract.

func (*Client) Subscribe

func (c *Client) Subscribe(ctx context.Context, subject string, handler message.Handler) (message.Subscription, error)

Subscribe registers a core NATS subscription. The subscription lifetime is bound to ctx; cancellation unsubscribes and stops later deliveries.

type ErrorHandler

type ErrorHandler func(context.Context, string, *message.Message, error)

ErrorHandler observes handler failures from asynchronous NATS callbacks. Core NATS has no acknowledgement decision to return to the server, so the adapter reports the error to the application instead of logging globally. Different subscriptions may call the handler concurrently.

type Option

type Option func(*options)

Option configures a Client.

func WithConn

func WithConn(conn *natsgo.Conn) Option

WithConn uses an application-owned NATS connection. Client.Close will not close a connection supplied this way.

func WithConnectOptions

func WithConnectOptions(opts ...natsgo.Option) Option

WithConnectOptions appends options used by nats.Connect.

func WithErrorHandler

func WithErrorHandler(handler ErrorHandler) Option

WithErrorHandler observes asynchronous handler failures.

func WithFlushTimeout

func WithFlushTimeout(timeout time.Duration) Option

WithFlushTimeout sets the default FlushWithContext deadline used by Publish and Subscribe when the caller's context has no deadline. A non-positive value leaves the caller's context unchanged.

func WithQueue

func WithQueue(queue string) Option

WithQueue subscribes through a NATS queue group.

func WithURL

func WithURL(url string) Option

WithURL sets the NATS server URL used when the adapter owns the connection.