API Reference

Quick reference for all goflux types, interfaces, and operators.

Transport Support Matrix

Interface	Channel	NATS	JetStream	HTTP
`Publisher[T]`	yes	yes	yes	yes
`Subscriber[T]`	yes	yes	yes	yes
`Requester[Req, Resp]`	-	yes	-	yes
`Responder[Req, Resp]`	-	yes	-	yes

Messaging Patterns

Pattern	Core Interface(s)	Transport(s)
Fire and Forget	`Publisher` + `Subscriber`, acker=nil	channel, nats
At-Least-Once (push)	`Publisher` + `Subscriber`, auto-ack	jetstream
At-Least-Once (pull)	`Publisher` + `Subscriber` + `WithManualAck`, middleware ack	jetstream
Stream Processing	`bridge.ToStream` + goflow operators	any
Manual Ack/Nak/Term	`Subscriber` + `WithManualAck()`	jetstream
Exactly-Once	`Publisher` with `Nats-Msg-Id` header	jetstream
Request-Reply	`Requester` + `Responder`	nats, http
Queue Groups	`Subscriber` + `WithQueueGroup`	nats
Fan-out	`bridge.ToStream` + goflow `Tee` / `FanOut`	any
Fan-in	`bridge.ToStream` + goflow `FanIn`	any

Core Types

Interfaces

// Publisher sends encoded messages to a subject.
type Publisher[T any] interface {
    Publish(ctx context.Context, subject string, v T) error
    Close() error
}

// Subscriber listens on a subject and dispatches decoded messages to a Handler.
// Subscribe blocks until ctx is cancelled or a fatal error occurs.
type Subscriber[T any] interface {
    Subscribe(ctx context.Context, subject string, handler Handler[T]) error
    Close() error
}

// ReadySubscriber is an optional interface for subscribers that can signal
// when their subscription is established. All goflux transports implement it.
type ReadySubscriber[T any] interface {
    Subscriber[T]
    SubscribeWithReady(ctx context.Context, subject string, handler Handler[T], ready func()) error
}

// Requester sends a typed request and waits for a typed response.
type Requester[Req, Resp any] interface {
    Request(ctx context.Context, subject string, req Req) (Resp, error)
    Close() error
}

// Responder handles incoming requests and produces typed responses.
type Responder[Req, Resp any] interface {
    Serve(ctx context.Context, subject string, handler RequestHandler[Req, Resp]) error
    Close() error
}

Readiness

Subscribe blocks for the lifetime of the subscription, so it has no return value that can mean "registered". A publish sequenced after a go sub.Subscribe(...) is therefore a race. ReadySubscriber closes that gap.

Function / Method	Description
`SubscribeWithReady(ctx, sub, subject, handler, ready)`	Package-level helper: uses `ReadySubscriber` when `sub` implements it, otherwise calls `ready()` then `Subscribe`
`sub.SubscribeWithReady(ctx, subject, handler, ready)`	Transport method — `ready()` fires once the subscription is established
`bound.SubscribeWithReady(ctx, handler, ready)`	`BoundSubscriber` variant; delegates to the wrapped subscriber
`responder.ServeWithReady(ctx, subject, handler, ready)`	NATS `Responder` variant — `ready()` fires once the server acknowledged the subscription

Contract:

ready is called at most once, and never after SubscribeWithReady returns.
If registration fails, the error is returned and ready is not called.
ready must not block.
For network transports, "established" means the broker acknowledged the subscription — not merely that a frame was buffered locally.

The package-level fallback (for third-party Subscriber implementations that do not implement ReadySubscriber) calls ready() immediately before subscribing. That is best-effort and still racy; every goflux transport takes the non-racy path.

See Subscriber Readiness for the rationale and usage.

Handler and RequestHandler

// Handler is the callback for Subscriber.Subscribe.
type Handler[T any] func(ctx context.Context, msg Message[T]) error

// RequestHandler processes a request and returns a response.
type RequestHandler[Req, Resp any] func(ctx context.Context, req Req) (Resp, error)

Message[T]

type Message[T any] struct {
    Subject string
    Payload T
    Header  Header
}

Method	Description
`NewMessage(subject, payload)`	Create a message
`NewMessageWithHeader(subject, payload, header)`	Create a message with header
`msg.WithAcker(a)`	Attach an acker (transport use only)
`msg.HasAcker()`	Check if acker is attached
`msg.Ack()`	Acknowledge successful processing
`msg.Nak()`	Signal processing failure (redelivery)
`msg.NakWithDelay(d)`	Nak with redelivery delay hint
`msg.Term()`	Terminal rejection (no redelivery)

type Header map[string][]string

Method	Description
`h.Get(key)`	First value for key, or `""`
`h.Set(key, value)`	Replace all values for key
`h.Add(key, value)`	Append a value for key
`h.Del(key)`	Remove key
`h.Clone()`	Deep copy

Acknowledgment Interfaces

// Acker -- minimal acknowledgment (Ack + Nak).
type Acker interface {
    Ack() error
    Nak() error
}

// DelayedNaker -- extends Acker with delayed negative acknowledgment.
type DelayedNaker interface {
    Acker
    NakWithDelay(d time.Duration) error
}

// Terminator -- extends Acker with terminal rejection.
type Terminator interface {
    Acker
    Term() error
}

Convenience Types

Type	Description
`Topic[T]`	Bundles `Publisher[T]` + `Subscriber[T]`
`BoundPublisher[T]`	Wraps `Publisher[T]` with a fixed subject
`Middleware[T]`	`func(Handler[T]) Handler[T]`
`PublisherMiddleware[T]`	`func(Publisher[T]) Publisher[T]`
`pipe.Filter[T]`	`func(ctx, msg Message[T]) bool`
`pipe.MapFunc[T, U]`	`func(ctx, msg Message[T]) (U, error)`
`pipe.FlatMapFunc[T, U]`	`func(ctx, msg Message[T]) ([]U, error)`
`pipe.DeadLetterFunc[T]`	`func(ctx, msg Message[T], err error)`

Context Helpers

Function	Description
`WithMessageID(ctx, id)`	Attach business-level message ID
`MessageID(ctx)`	Read message ID from context
`WithHeader(ctx, h)`	Attach header to context (read by transports on Publish)
`HeaderFromContext(ctx)`	Read header from context

Middleware

All middleware have the signature Middleware[T] = func(Handler[T]) Handler[T].

Built-in implementations live in the github.com/foomo/goflux/middleware package. The Middleware[T] type and Chain[T] remain in the root package.

Middleware	Package	Signature	Description
`AutoAck[T]()`	`middleware`	`-> Middleware[T]`	Ack on nil error, Nak on non-nil
`RetryAck[T](policy)`	`middleware`	`RetryPolicy -> Middleware[T]`	Classify errors into ack/nak/term via policy
`InjectMessageID[T]()`	`middleware`	`-> Middleware[T]`	Header message ID → context
`InjectHeader[T]()`	`middleware`	`-> Middleware[T]`	Message header → context
`ForwardMessageID[T]()`	`middleware`	`-> Middleware[T]`	Forward message ID from context through pipe stages
`Chain[T](mws...)`	`goflux`	`...Middleware[T] -> Middleware[T]`	Compose left-to-right

Stream Processing Operators

For concurrency limiting, deduplication, throttling, skip/take, and peek, use goflow operators via bridge.ToStream.

See Middleware for details and examples.

Pipeline Operators

Operator	Description
`pipe.New[T](pub, opts...)`	Forward messages to a publisher
`pipe.NewMap[T, U](pub, mapFn, opts...)`	Transform and forward
`pipe.NewFlatMap[T, U](pub, fn, opts...)`	Expand and forward (1→N)
`BindPublisher[T](pub, subject)`	Fix subject on a publisher
`ToChan[T](ctx, sub, subject, bufSize)`	Bridge subscriber to `<-chan Message[T]`
`bridge.ToStream[T](ctx, sub, subject, bufSize)`	Bridge subscriber to `goflow.Stream[Message[T]]`
`bridge.FromStream[T](stream, pub)`	Consume goflow stream and publish each message
`RetryPublisher[T](pub, maxAttempts, backoff, opts...)`	Wrap publisher with retry logic

Pipe Options

Option	Description
`pipe.WithFilter[T](f)`	Skip messages where filter returns false
`pipe.WithDeadLetter[T](fn)`	Observe failed messages (does not swallow error)
`pipe.WithMiddleware[T](mw...)`	Wrap pipe handler with middleware
`pipe.WithMapFilter[T, U](f)`	Filter for map/flatmap pipes
`pipe.WithMapDeadLetter[T, U](fn)`	Dead-letter observer for map/flatmap pipes
`pipe.WithMapMiddleware[T, U](mw...)`	Middleware for map/flatmap pipes

See Pipeline Operators for details and examples.

Group Coordination

// Group coordinates the lifecycle of multiple message handlers.
// Fail-fast: first error cancels all sibling tasks.
type Group struct{ /* ... */ }

Function / Method	Description
`NewGroup(opts...)`	Create a lifecycle coordinator
`WithGroupOptions(opts...)`	Pass `gofuncy.GroupOption` to the underlying group
`g.Go(name, fn)`	Register a named blocking task
`g.GoWithReady(name, fn)`	Register a task that signals readiness via `ready()`
`g.Run(ctx)`	Start all tasks, block until done (fail-fast)
`g.RunWithReady(ctx, onReady)`	Start all tasks, call `onReady` when all are ready
`GroupSubscribe[T](g, name, sub, subject, handler, mws...)`	Register a Subscriber handler on the group

g := goflux.NewGroup()

goflux.GroupSubscribe(g, "orders", orderSub, "orders.new", orderHandler)

err := g.Run(ctx) // blocks, fail-fast on first error

RetryPublisher

func RetryPublisher[T any](pub Publisher[T], maxAttempts int, backoff BackoffFunc, opts ...RetryPublisherOption) Publisher[T]

type BackoffFunc func(attempt int) time.Duration
type RetryableFunc func(err error) bool

Wraps a Publisher[T] with retry logic. On publish failure, retries until maxAttempts publish calls have been made, with delays from backoff. Context cancellation aborts immediately.

pub := goflux.RetryPublisher[Event](innerPub, 3, func(attempt int) time.Duration {
    return time.Duration(attempt+1) * time.Second // linear backoff
})

Option	Description
`WithRetryable(fn)`	Retry only the errors for which `fn` returns true (default: retry all)

pub := goflux.RetryPublisher[Event](innerPub, 3, backoff,
    goflux.WithRetryable(func(err error) bool {
        return errors.Is(err, goflux.ErrTransport)
    }),
)

Sentinel Errors

Error	Description
`ErrPublish`	Failure in the publish path
`ErrSubscribe`	Failure in the subscribe path
`ErrEncode`	Serialization failure
`ErrDecode`	Deserialization failure
`ErrTransport`	Transport-level failure (network, protocol)

Transports join these with causal errors via errors.Join, enabling:

if errors.Is(err, goflux.ErrEncode) { /* handle codec problem */ }

Retry Policy (`middleware` package)

type RetryPolicy func(err error) RetryDecision

type RetryDecision struct {
    Action RetryAction
    Delay  time.Duration
}

Value	Package	Description
`RetryNak`	`middleware`	Immediate redelivery via `Nak()`
`RetryNakWithDelay`	`middleware`	Delayed redelivery via `NakWithDelay(d)`
`RetryTerm`	`middleware`	Terminal rejection via `Term()`

Function	Package	Description
`NewRetryPolicy(delay)`	`middleware`	Term non-retryable, nak-with-delay otherwise
`RetryAck[T](policy)`	`middleware`	Middleware that applies retry policy to ack decisions
`ErrNonRetryable`	`goflux`	Sentinel for permanent failures
`NonRetryable(err)`	`goflux`	Wrap error as non-retryable
`IsNonRetryable(err)`	`goflux`	Check if error is non-retryable

See Stream Processing for details and examples.

Telemetry

Function / Method	Description
`NewTelemetry(opts...)`	Create instrumentation instance
`DefaultTelemetry(tel)`	Return `tel` if non-nil, else create from OTel globals
`NewNoopTelemetry()`	Create noop instance (safe calls, no output)
`WithTracerProvider(tp)`	Set tracer provider
`WithMeterProvider(mp)`	Set meter provider
`WithPropagator(p)`	Set text-map propagator
`WithDestinationTemplate(fn)`	Map subjects to templates to bound cardinality in span names and metrics
`RecordPublish(ctx, subject, system, fn)`	Producer span
`RecordProcess(ctx, subject, system, fn, opts...)`	Consumer span
`RecordFetch(ctx, subject, system, count, fn)`	Consumer span (pull fetch)
`RecordRequest(ctx, subject, system, fn)`	Request-reply client span
`RecordAckOutcome(ctx, action, subject, err)`	Record ack/nak/term outcome
`RegisterLag(subject, lagFn)`	Observable gauge
`InjectContext(ctx, carrier)`	Inject span context (publish side)
`ExtractContext(ctx, carrier)`	Extract as parent (sync transports)
`ExtractSpanContext(ctx, carrier)`	Extract as link (async transports)
`WithRemoteSpanContext(sc)`	Attach producer span as link

See Telemetry for details.

API Reference ​

Transport Support Matrix ​

Messaging Patterns ​

Core Types ​

Interfaces ​

Readiness ​

Handler and RequestHandler ​

Message[T] ​

Header ​

Acknowledgment Interfaces ​

Convenience Types ​

Context Helpers ​

Middleware ​

Pipeline Operators ​

Pipe Options ​

Group Coordination ​

RetryPublisher ​

Sentinel Errors ​

Retry Policy (middleware package) ​

Telemetry ​