README
¶
Modulex NATS EventBus Adapter
This package provides a modulex.EventBus implementation backed by
NATS.
Usage
import (
"github.com/nats-io/nats.go"
natsadapter "github.com/mediusfy/modulex/nats"
)
conn, err := nats.Connect(nats.DefaultURL)
if err != nil {
return err
}
defer conn.Close()
eb := natsadapter.NewEventBus(conn)
manager, err := modulex.NewManager(modulex.WithEventBus(eb), modulex.WithLogger(logger))
if err != nil {
return err
}
Behavior
Publishmaps directly toconn.Publish(topic, payload).Subscribecreates a NATS subscription and adapts incoming messages to the genericmodulex.EventHandlersignature.Closeunsubscribes all registered subscriptions.
Testing
The adapter tests start an embedded NATS server using
github.com/nats-io/nats-server/v2/test. Run them with:
go test ./nats/...
Documentation
¶
Overview ¶
Package nats provides a Modulex EventBus adapter backed by NATS.
Index ¶
- Variables
- type EventBus
- type JetStreamEventBus
- func (j *JetStreamEventBus) Close(ctx context.Context) error
- func (j *JetStreamEventBus) Publish(ctx context.Context, topic string, payload []byte) error
- func (j *JetStreamEventBus) Subscribe(context.Context, string, modulex.EventHandler) error
- func (j *JetStreamEventBus) SubscribeDurable(ctx context.Context, topic string, handler modulex.DurableHandler, ...) error
- type JetStreamOption
- func WithDurableAckWait(d time.Duration) JetStreamOption
- func WithDurableBatchSize(n int) JetStreamOption
- func WithDurableDeadLetterSuffix(suffix string) JetStreamOption
- func WithDurableFetchWait(d time.Duration) JetStreamOption
- func WithDurableMaxDeliver(n int) JetStreamOption
- func WithJetStreamLogger(logger *slog.Logger) JetStreamOption
- type Option
Constants ¶
This section is empty.
Variables ¶
var ErrDurableConsumerNameRequired = errors.New("nats: SubscribeDurable requires a non-empty ConsumerName")
ErrDurableConsumerNameRequired is returned by SubscribeDurable when called without modulex.WithConsumerName (or with an empty/whitespace-only name).
var ErrJetStreamSubscribeUnsupported = errors.New("nats: JetStreamEventBus does not support Subscribe; use nats.EventBus or SubscribeDurable")
ErrJetStreamSubscribeUnsupported is returned by JetStreamEventBus.Subscribe. JetStream consumption requires substantially more configuration (durable vs ephemeral consumers, ack policies, delivery subjects, replay policy) than the core NATS EventBus's fire-and-forget Subscribe can express, so JetStreamEventBus is deliberately publish-only via the plain modulex.EventBus/Subscriber capability. Use EventBus.Subscribe (core NATS, no durability) for fire-and-forget consumption, or SubscribeDurable (modulex.DurableConsumer) on this same type for durable, acknowledged consumption.
Functions ¶
This section is empty.
Types ¶
type EventBus ¶
type EventBus struct {
// contains filtered or unexported fields
}
EventBus implements modulex.EventBus by wrapping a concrete NATS connection.
func NewEventBus ¶
NewEventBus instantiates the NATS event bus driver.
The EventBus does not take ownership of conn: the caller creates and closes the underlying *nats.Conn, typically after modulex.Manager.StopModules has closed the EventBus. This lets a single connection be shared across multiple concerns outside the module lifecycle if desired.
func (*EventBus) Close ¶
Close implements modulex.EventBus. It unsubscribes all registered NATS subscriptions but does not close the underlying *nats.Conn, which the caller owns.
func (*EventBus) Subscribe ¶
Subscribe implements modulex.EventBus. It registers a NATS subscription, adapting the incoming message to the generic EventHandler signature.
The subscriber's context is propagated into the handler. If the incoming NATS message carries W3C trace context headers, they are extracted and merged so OpenTelemetry span continuity is preserved across the broker.
NATS core has no acknowledgement semantics, so a failing handler cannot be redelivered or retried by the broker. The error is logged so failures are visible instead of silently discarded; this mirrors the acknowledge-and-log policy used by the other EventBus adapters in this module (see rabbitmq.EventBus.Subscribe and watermill.EventBus.Subscribe).
type JetStreamEventBus ¶ added in v0.5.1
type JetStreamEventBus struct {
// contains filtered or unexported fields
}
JetStreamEventBus implements modulex.EventBus's Publish and Close using NATS JetStream for at-least-once, acknowledged publishing, and additionally implements modulex.DurableConsumer via SubscribeDurable for durable, explicitly-acknowledged consumption.
Use JetStreamEventBus when a module needs to publish (fire-and-confirm) to a JetStream stream, and optionally also consume durably from it with explicit ack/nack/dead-letter control — e.g. sourcing and/or consuming domain events shared with other services.
func NewJetStreamEventBus ¶ added in v0.5.1
func NewJetStreamEventBus(js nats.JetStreamContext, opts ...JetStreamOption) *JetStreamEventBus
NewJetStreamEventBus instantiates an EventBus backed by JetStream that publishes via Publish/Close and, when SubscribeDurable is used, also provides durable consumption (modulex.DurableConsumer). js is typically obtained via (*nats.Conn).JetStream(); the EventBus does not take ownership of the underlying connection, matching the other EventBus adapters in this module.
func (*JetStreamEventBus) Close ¶ added in v0.5.1
func (j *JetStreamEventBus) Close(ctx context.Context) error
Close implements modulex.EventBus. It cancels all active SubscribeDurable pull loops, unsubscribes their JetStream consumers, and waits for the loop goroutines to exit (bounded by ctx). JetStreamContext has no separate connection to close; the underlying *nats.Conn is caller-owned, matching the other EventBus adapters in this module.
func (*JetStreamEventBus) Publish ¶ added in v0.5.1
Publish implements modulex.EventBus. It publishes to the JetStream stream whose subject matches topic and waits for the broker's acknowledgement.
func (*JetStreamEventBus) Subscribe ¶ added in v0.5.1
func (j *JetStreamEventBus) Subscribe(context.Context, string, modulex.EventHandler) error
Subscribe implements modulex.EventBus. It always returns ErrJetStreamSubscribeUnsupported; see the JetStreamEventBus doc comment. Use SubscribeDurable for durable JetStream consumption.
func (*JetStreamEventBus) SubscribeDurable ¶ added in v0.6.0
func (j *JetStreamEventBus) SubscribeDurable(ctx context.Context, topic string, handler modulex.DurableHandler, opts ...modulex.DurableSubscribeOption) error
SubscribeDurable implements modulex.DurableConsumer using a JetStream pull-based durable consumer.
Acknowledgement: handler's returned modulex.AckDecision is translated to JetStream's native ack API — Ack calls msg.Ack(), Nack calls msg.Nak(), and DeadLetter republishes the message to the configured dead-letter subject (see WithDurableDeadLetterSuffix) and then calls msg.Term() so JetStream never redelivers it. An unrecognized AckDecision value (e.g. the zero value of a differently-typed constant) is treated as Nack, erring toward retry rather than silently acking or dropping.
Retry: on Nack, JetStream redelivers the message after WithDurableAckWait, up to WithDurableMaxDeliver total attempts (including the first); beyond that JetStream itself stops redelivering even without an explicit DeadLetter decision (this is JetStream's own max-deliver behavior, not something this adapter enforces separately).
Replay: modulex.WithReplayPolicy(modulex.ReplayAll) (the default) creates the durable consumer with JetStream's DeliverAll policy; modulex.ReplayNew creates it with DeliverNew. This only affects a brand-new ConsumerName — JetStream resumes a pre-existing durable consumer from its last acknowledged position regardless of ReplayPolicy.
Ordering: this implementation runs one sequential pull loop per SubscribeDurable call — it fetches a batch, resolves (ack/nack/dead- letter) every message in the batch in order, and only then fetches the next batch — so relative delivery order is preserved for that subscription. If multiple SubscribeDurable calls share one ConsumerName, JetStream load-balances deliveries across them and order across those concurrent pull loops is not guaranteed.
Consumer identity: modulex.WithConsumerName (required) becomes the JetStream durable consumer name. Reusing it resumes the same consumer's ack progress, including across process restarts; multiple concurrent SubscribeDurable calls with the same name form a JetStream competing-consumers group.
Dead-letter: see Acknowledgement above.
type JetStreamOption ¶ added in v0.5.1
type JetStreamOption func(*JetStreamEventBus)
JetStreamOption configures a JetStreamEventBus during construction.
func WithDurableAckWait ¶ added in v0.6.0
func WithDurableAckWait(d time.Duration) JetStreamOption
WithDurableAckWait sets how long JetStream waits for an ack/nack/term before considering a delivery unacknowledged and eligible for redelivery, for all SubscribeDurable subscriptions on this JetStreamEventBus. d must be positive; non-positive values are ignored and the default (30s) is used.
func WithDurableBatchSize ¶ added in v0.6.0
func WithDurableBatchSize(n int) JetStreamOption
WithDurableBatchSize sets how many messages a durable consumer's pull loop requests per fetch, for all SubscribeDurable subscriptions on this JetStreamEventBus. n must be positive; non-positive values are ignored and the default (10) is used.
func WithDurableDeadLetterSuffix ¶ added in v0.6.0
func WithDurableDeadLetterSuffix(suffix string) JetStreamOption
WithDurableDeadLetterSuffix sets the suffix appended to a topic to form the subject a dead-lettered message (AckDecision DeadLetter) is republished to before the original delivery is terminated, for all SubscribeDurable subscriptions on this JetStreamEventBus. The resulting subject must be covered by a JetStream stream (the same stream, if its subject list includes a matching wildcard, or a separate one) or the republish fails; the original message is still terminated in that case so it is not redelivered forever, and the republish failure is logged.
Pass an empty suffix to disable republishing: DeadLetter then only terminates the original delivery. The default suffix is ".DEAD".
func WithDurableFetchWait ¶ added in v0.6.0
func WithDurableFetchWait(d time.Duration) JetStreamOption
WithDurableFetchWait sets the maximum time a single pull request waits for at least one message before returning empty (and the loop checks for context cancellation again), for all SubscribeDurable subscriptions on this JetStreamEventBus. d must be positive; non-positive values are ignored and the default (5s) is used.
func WithDurableMaxDeliver ¶ added in v0.6.0
func WithDurableMaxDeliver(n int) JetStreamOption
WithDurableMaxDeliver sets the maximum number of delivery attempts (including the first) a durable consumer makes for a message before JetStream stops redelivering it, for all SubscribeDurable subscriptions on this JetStreamEventBus. This is a construction-time option (rather than a per-call modulex.DurableSubscribeOption) because it is a NATS/JetStream-specific knob that the adapter-agnostic core interface does not carry. n must be positive; non-positive values are ignored and the default (5) is used.
func WithJetStreamLogger ¶ added in v0.5.1
func WithJetStreamLogger(logger *slog.Logger) JetStreamOption
WithJetStreamLogger sets the logger used to report errors. If not provided, or if nil, slog.Default() is used.
type Option ¶ added in v0.5.1
type Option func(*EventBus)
Option configures an EventBus during construction.
func WithLogger ¶ added in v0.5.1
WithLogger sets the logger used to report handler errors encountered while consuming messages. If not provided, or if nil, slog.Default() is used.
Source Files