rabbitmq

package
v0.4.2 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 22, 2026 License: MIT Imports: 6 Imported by: 0

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

  • Publish publishes a message to the default exchange using the topic as the routing key, which routes to the queue of the same name.
  • Subscribe starts a consumer on the named queue and invokes the handler for each delivered message.
  • Close cancels 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

func NewEventBus(ch *amqp.Channel) *EventBus

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

func (r *EventBus) Close(ctx context.Context) error

Close implements modulex.EventBus. It cancels all active queue consumers but does not close the underlying *amqp.Channel or its connection, which the caller owns.

func (*EventBus) Publish

func (r *EventBus) Publish(ctx context.Context, topic string, payload []byte) error

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

func (r *EventBus) Subscribe(ctx context.Context, topic string, handler modulex.EventHandler) error

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.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL