README
¶
Modulex RabbitMQ EventBus Adapter
This package provides a modulex.EventBus implementation backed by
RabbitMQ via github.com/rabbitmq/amqp091-go.
Usage
import (
amqp "github.com/rabbitmq/amqp091-go"
rabbitadapter "github.com/mediusfy/modulex/rabbitmq"
)
conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
return err
}
defer conn.Close()
ch, err := conn.Channel()
if err != nil {
return err
}
defer ch.Close()
eb := rabbitadapter.NewEventBus(ch)
manager, err := modulex.NewManager(modulex.WithEventBus(eb), modulex.WithLogger(logger))
if err != nil {
return err
}
Behavior
Publishpublishes a message to the default exchange using the topic as the routing key, which routes to the queue of the same name.Subscribestarts a consumer on the named queue and invokes the handler for each delivered message.Closecancels the internal consumer goroutines.
Testing
The adapter tests connect to a live RabbitMQ broker. By default they target
amqp://guest:guest@localhost:5672/. Set RABBITMQ_URL to use a different
broker. Tests skip gracefully when no broker is available.
RABBITMQ_URL=amqp://guest:guest@localhost:5672/ go test ./rabbitmq/...
Documentation
¶
Index ¶
- type EventBus
- func (r *EventBus) Close(ctx context.Context) error
- func (r *EventBus) Publish(ctx context.Context, topic string, payload []byte) error
- func (r *EventBus) Subscribe(ctx context.Context, topic string, handler modulex.EventHandler) error
- func (r *EventBus) SubscribeWithOptions(ctx context.Context, topic string, handler modulex.EventHandler, ...) error
- type Option
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type EventBus ¶
type EventBus struct {
// contains filtered or unexported fields
}
func NewEventBus ¶
NewEventBus instantiates the RabbitMQ event bus driver.
The EventBus does not take ownership of ch: the caller creates and closes the underlying *amqp.Channel (and its connection), typically after modulex.Manager.StopModules has closed the EventBus. This lets a single channel or connection be shared across multiple concerns outside the module lifecycle if desired.
func (*EventBus) Close ¶
Close implements modulex.EventBus. It cancels all active queue consumers, waits for their goroutines to exit, and does not close the underlying *amqp.Channel or its connection, which the caller owns.
func (*EventBus) Publish ¶
Publish implements modulex.EventBus.
Publishing uses the RabbitMQ default exchange (""), where the routing key is interpreted as the target queue name. Therefore the topic parameter is the queue to which the message is delivered. For routed exchanges, use a broker-specific publisher instead of this adapter.
func (*EventBus) Subscribe ¶
Subscribe implements modulex.EventBus. It declares the target queue and then consumes messages from it in a background routine. The queue is declared as durable and non-exclusive so the adapter works out of the box. If the caller cancels the supplied context, the consumer goroutine exits.
Messages are acknowledged manually: a successful handler call acks the message, and a failing handler call nacks it without requeue and logs the error. Requeuing is deliberately not attempted here since a persistently failing handler would otherwise redeliver the same message forever; this matches the acknowledge-and-log policy used by the other EventBus adapters in this module (see watermill.EventBus.Subscribe).
Subscribe fails if SubscribeWithOptions has already been called on this EventBus: RabbitMQ's per-consumer Qos default is scoped to the whole channel and stays changed for every consumer created afterward, so a plain consumer created after a pooled one would silently inherit the pool's prefetch instead of the channel's original value. Use a dedicated *amqp.Channel (and EventBus) for plain subscriptions made after any SubscribeWithOptions call.
func (*EventBus) SubscribeWithOptions ¶ added in v0.8.0
func (r *EventBus) SubscribeWithOptions(ctx context.Context, topic string, handler modulex.EventHandler, options workerpool.Options) error
SubscribeWithOptions enables opt-in bounded concurrent processing. RabbitMQ prefetch is set to Workers plus QueueCapacity, messages are acknowledged or nacked only after their handler completes, and processing order is not guaranteed when Workers is greater than one.
Calling SubscribeWithOptions changes this channel's default prefetch for every consumer created afterward, for the rest of this EventBus's lifetime; see Subscribe's doc comment for why a plain Subscribe call made after this one is rejected.
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