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
¶
Overview ¶
Package rabbitmq provides a Modulex EventBus adapter backed by RabbitMQ.
Index ¶
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).
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