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 ¶
Constants ¶
This section is empty.
Variables ¶
var ErrJetStreamSubscribeUnsupported = errors.New("nats: JetStreamEventBus does not support Subscribe; use nats.EventBus or a direct JetStream consumer")
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. Use EventBus.Subscribe (core NATS, no durability) or a direct JetStream consumer for message 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.
Use JetStreamEventBus when a module only needs to publish (fire-and- confirm) to a JetStream stream, e.g. sourcing domain events for other services to consume via their own JetStream consumers.
func NewJetStreamEventBus ¶ added in v0.5.1
func NewJetStreamEventBus(js nats.JetStreamContext, opts ...JetStreamOption) *JetStreamEventBus
NewJetStreamEventBus instantiates a publish-only EventBus backed by JetStream. 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(context.Context) error
Close implements modulex.EventBus. 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.
type JetStreamOption ¶ added in v0.5.1
type JetStreamOption func(*JetStreamEventBus)
JetStreamOption configures a JetStreamEventBus during construction.
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