messaging

Documentation

Overview

Package messaging provides a broker-agnostic pub/sub abstraction with typed JSON handlers and lifecycle-managed subscriptions.

Index

• Constants
• Variables
• func ContextWithCorrelationID(ctx context.Context, id string) context.Context
• func CorrelationID(headers map[string][]string) string
• func CorrelationIDFromContext(ctx context.Context) string
• func ExtractContext(ctx context.Context, headers map[string][]string) context.Context
• func HeaderValue(headers map[string][]string, key string) string
• func InjectContext(ctx context.Context, headers map[string][]string) map[string][]string
• func Module(client Client, name ...string) host.Module
• func SetCorrelationID(headers map[string][]string, id string) map[string][]string
• func SetHeader(headers map[string][]string, key, value string) map[string][]string
• func Subscribe(name, subject string, handler Handler, opts ...SubscriberOption) host.Module
• type Client
  • func Lookup(app *host.App, name ...string) (Client, bool)
  • func Of(app *host.App, name ...string) Client
• type Envelope
  • func (b *Envelope) ExtractBodyTo(v any) error
• type EnvelopeHandler
• type Handler
  • func HandleEnvelope[T any](fn EnvelopeHandler[T]) Handler
  • func HandleJSON[T any](fn TypedHandler[T]) Handler
• type Message
  • func NewJSONMessage[T any](subject string, payload T) (Message, error)
• type Publisher
• type Request
• type RequestHandler
• type Requester
• type Responder
• type Response
• type Subscriber
• type SubscriberOption
  • func WithQueueGroup(queue string) SubscriberOption
• type Subscription
• type TypedHandler

Constants

const CorrelationIDHeader = core.CorrelationIDHeader

CorrelationIDHeader is the header used to carry a correlation (request) id across the broker. It is the canonical correlation-id header defined in core, so a request id flows uniformly http -> messaging -> http.

Variables

var ErrTimeout = errors.New("messaging: request timed out")

ErrTimeout is returned by Requester.Request when a reply is not received in time. Transports normalise their broker-specific timeout errors into this sentinel so callers can detect timeouts with errors.Is without importing a broker package.

Functions

func ContextWithCorrelationID

func ContextWithCorrelationID(ctx context.Context, id string) context.Context

ContextWithCorrelationID returns a copy of ctx carrying the correlation id. Consumers call this after reading CorrelationIDHeader off an inbound message so downstream code (logging, nested calls) can recover the id from context. It delegates to core so the id shares one context key across http and messaging.

func CorrelationID

func CorrelationID(headers map[string][]string) string

CorrelationID returns the correlation id carried in headers, or "" if none.

func CorrelationIDFromContext

func CorrelationIDFromContext(ctx context.Context) string

CorrelationIDFromContext returns the correlation id stored in ctx, or "" if none is present.

func ExtractContext

func ExtractContext(ctx context.Context, headers map[string][]string) context.Context

ExtractContext returns ctx enriched with the metadata carried in headers: the remote W3C trace context and the correlation id. Broker clients call it before invoking a handler so logging and nested calls continue the trace.

func HeaderValue

func HeaderValue(headers map[string][]string, key string) string

HeaderValue returns the first value stored under key, or "" if the header is absent or empty.

func InjectContext

func InjectContext(ctx context.Context, headers map[string][]string) map[string][]string

InjectContext returns headers enriched with the cross-process metadata carried by ctx: the W3C trace context (traceparent/tracestate, via the global OpenTelemetry propagator — a no-op until one is installed, e.g. by the otelx module) and the correlation id. An existing correlation header is preserved. The (possibly newly allocated) map is returned so callers can write:

msg.Headers = messaging.InjectContext(ctx, msg.Headers)

func Module

func Module(client Client, name ...string) host.Module

Module registers a Client as the app's broker. The host drives the full lifecycle: it loads the client's config (if it implements configx.Configurable), dials during init, and disconnects on shutdown. Pass any implementation — there is no built-in broker:

host.MustNew().WithModule(messaging.Module(nats.New())).MustRun()

If the client implements SetLogger, it receives the host logger here. Reach the client from a Setup hook or handler with messaging.Of(app). Pass an optional name to register several brokers side by side, each retrieved with messaging.Of(app, name).

func SetCorrelationID

func SetCorrelationID(headers map[string][]string, id string) map[string][]string

SetCorrelationID returns headers with the correlation id set, allocating the map when it is nil.

func SetHeader

func SetHeader(headers map[string][]string, key, value string) map[string][]string

SetHeader returns headers with key set to a single value, allocating the map when it is nil. The (possibly new) map is returned so callers can write:

msg.Headers = messaging.SetHeader(msg.Headers, k, v)

func Subscribe

func Subscribe(name, subject string, handler Handler, opts ...SubscriberOption) host.Module

Subscribe installs a message subscription tied to the app lifecycle: it subscribes once the broker is connected and the app enters its start phase, and unsubscribes on shutdown. Requires messaging.Module. handler may be a raw Handler or one built with HandleJSON / HandleEnvelope for typed payloads:

host.MustNew().WithModules(
    messaging.Module(nats.New()),
    messaging.Subscribe("orders", "orders.created",
        messaging.HandleJSON(func(ctx context.Context, o Order) error { ... }),
        messaging.WithQueueGroup("order-workers")),
)

Several Subscribe modules share one underlying consumer, so they start and stop together.

Types

type Client

type Client interface {
	Publisher
	Subscriber
	Requester
	Responder
	core.HealthChecker

	Connect() error
	Disconnect() error
	IsConnected() bool
}

Client is the messaging abstraction used by the host. Publish and Subscribe take a context for cancellation/deadlines, and messages carry headers so metadata (correlation ids, trace context) survives across the broker.

func Lookup

func Lookup(app *host.App, name ...string) (Client, bool)

Lookup returns the messaging client installed under the optional name and whether one was installed. Use it where absence is an expected, recoverable condition; prefer Of when the client must exist.

func Of

func Of(app *host.App, name ...string) Client

Of returns the messaging client installed by Module under the optional name, panicking if none was installed.

type Envelope

type Envelope struct {
	Type      string         `json:"type"`
	ID        string         `json:"id"`
	Name      string         `json:"name"`
	Source    string         `json:"source"`
	Target    *string        `json:"target"`
	Headers   map[string]any `json:"headers"`
	Body      any            `json:"body"`
	CreatedAt int64          `json:"createdAt"`
}

Envelope is a message envelope carrying metadata alongside an arbitrary Body. Use ExtractBodyTo to decode Body into a concrete type.

func (*Envelope) ExtractBodyTo

func (b *Envelope) ExtractBodyTo(v any) error

type EnvelopeHandler

type EnvelopeHandler[T any] func(ctx context.Context, env *Envelope, body T) error

EnvelopeHandler processes a decoded Envelope together with its body extracted into T. Pair it with HandleEnvelope.

type Handler

type Handler func(ctx context.Context, msg *Message) error

Handler processes a received message. The context is derived from the subscription context and is cancelled when the subscription ends.

func HandleEnvelope

func HandleEnvelope[T any](fn EnvelopeHandler[T]) Handler

HandleEnvelope adapts an EnvelopeHandler[T] into a raw Handler for messages carrying an Envelope JSON payload: it decodes the envelope, extracts its body into T, and calls fn with both. Decode/extract failures are returned as BadRequest *errorx.Error values and fn is not invoked.

func HandleJSON

func HandleJSON[T any](fn TypedHandler[T]) Handler

HandleJSON adapts a TypedHandler[T] into a raw Handler by JSON-decoding the message data into T before calling fn. A decode failure is returned as a BadRequest *errorx.Error (so it is distinguishable from a handler-side failure) and fn is not invoked.

sub, _ := client.Subscribe(ctx, "orders.created",
    messaging.HandleJSON(func(ctx context.Context, o Order) error { ... }))

type Message

type Message struct {
	Subject string
	Data    []byte
	Headers map[string][]string
}

Message is a published or received message. Headers carry metadata such as a correlation id (e.g. "X-Request-ID") so it can be propagated across services.

func NewJSONMessage

func NewJSONMessage[T any](subject string, payload T) (Message, error)

NewJSONMessage builds a Message whose data is the JSON encoding of payload, for publishing a typed value without hand-marshaling. Headers are left nil; the broker client adds trace/correlation headers at publish time.

type Publisher

type Publisher interface {
	// Publish sends a message to the specified subject
	Publish(ctx context.Context, msg Message) error
}

Publisher defines a generic interface for message publishing

type Request

type Request struct {
	Subject string
	Data    []byte
	Headers map[string][]string
}

type RequestHandler

type RequestHandler func(ctx context.Context, req *Request) (*Response, error)

Handler processes a received message. The context is derived from the subscription context and is cancelled when the subscription ends.

type Requester

type Requester interface {
	// Request sends a request and waits for a reply
	Request(ctx context.Context, req Request) (*Response, error)
}

Requester defines a generic interface for request-reply pattern

type Responder

type Responder interface {
	// Respond registers a handler for requests on the specified subject
	Respond(command string, handler RequestHandler) (Subscription, error)
	// QueueRespond registers a handler for requests on the specified subject and a queue group
	QueueRespond(command string, queue string, handler RequestHandler) (Subscription, error)
}

Responder defines a generic interface for handling request-reply pattern

type Response

type Response struct {
	Subject string
	Data    []byte
	Reply   string
	Headers map[string][]string
}

type Subscriber

type Subscriber interface {
	// Subscribe registers a handler for the specified subject
	Subscribe(ctx context.Context, subject string, handler Handler) (Subscription, error)
	// QueueSubscribe registers a handler for the specified subject and queue group
	QueueSubscribe(ctx context.Context, subject string, queue string, handler Handler) (Subscription, error)
}

Subscriber defines a generic interface for message subscription

type SubscriberOption

type SubscriberOption func(*subscriberBinding)

SubscriberOption configures a subscription registered with Subscriber.

func WithQueueGroup

func WithQueueGroup(queue string) SubscriberOption

WithQueueGroup makes the subscription join a queue group, so each message is delivered to only one member of the group — use it to load-balance a subject across replicas of the same service.

type Subscription

type Subscription interface {
	// Unsubscribe cancels the subscription
	Unsubscribe() error

	// Subject returns the subject of the subscription
	Subject() string
}

Subscription represents an active subscription that can be cancelled.

type TypedHandler

type TypedHandler[T any] func(ctx context.Context, payload T) error

TypedHandler processes a message payload already decoded into T. Pair it with HandleJSON to get a raw Handler the broker can dispatch.