mq

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 17, 2026 License: MIT Imports: 8 Imported by: 0

README

mq

Broker-agnostic message queue abstraction with pluggable backends.

Packages

Package Description
mq/ Core interfaces: Message, Delivery, Producer, Consumer, Broker, Handler, Middleware
mq/factory/ Unified factory — NewBroker(BrokerConfig) dispatches to all backends
mq/rabbitmq/ RabbitMQ backend (amqp091-go)
mq/kafka/ Kafka backend (segmentio/kafka-go)
mq/rocketmq/ RocketMQ backend (apache/rocketmq-client-go/v2)
mq/activemq/ ActiveMQ backend via STOMP (go-stomp/stomp)
mq/redisstream/ Redis Streams backend (go-redis)

Quick start via factory

import (
    "github.com/LingByte/ling-base/mq"
    "github.com/LingByte/ling-base/mq/factory"
    "github.com/LingByte/ling-base/mq/rabbitmq"
)

broker, err := factory.NewBroker(mq.BrokerConfig{
    Type:           mq.BrokerRabbitMQ,
    RabbitMQConfig: &rabbitmq.Config{URL: "amqp://guest:guest@localhost:5672/"},
})
if err != nil { panic(err) }
defer broker.Close()

broker.Connect()

// Produce
producer, _ := broker.Producer("events", mq.PublishOptions{Persistent: true})
producer.Publish(ctx, &mq.Message{
    RoutingKey:  "events.user.created",
    ContentType: "application/json",
    Body:        []byte(`{"user_id":123}`),
})

// Consume with middleware
consumer, _ := broker.Consumer("events.queue", mq.ConsumeOptions{
    Handler: func(ctx context.Context, d mq.Delivery) error {
        log.Println(string(d.Body()))
        return d.Ack()
    },
    Middleware: []mq.Middleware{
        mq.RecoverMiddleware(logger),
        mq.MetricsMiddleware(metrics),
        mq.RetryMiddleware(mq.RetryConfig{MaxAttempts: 3, Backoff: time.Second}),
    },
    QosPrefetchCount: 10,
    Concurrency:      4,
})
consumer.Start(ctx)

Switching backends

// Kafka
broker, _ := factory.NewBroker(mq.BrokerConfig{
    Type:       mq.BrokerKafka,
    KafkaConfig: &kafka.Config{Brokers: []string{"localhost:9092"}, GroupID: "my-group"},
})

// RocketMQ
broker, _ := factory.NewBroker(mq.BrokerConfig{
    Type:          mq.BrokerRocketMQ,
    RocketMQConfig: &rocketmq.Config{NameServer: []string{"127.0.0.1:9876"}},
})

// ActiveMQ (STOMP)
broker, _ := factory.NewBroker(mq.BrokerConfig{
    Type:          mq.BrokerActiveMQ,
    ActiveMQConfig: &activemq.Config{Addr: "localhost:61613"},
})

// Redis Streams
broker, _ := factory.NewBroker(mq.BrokerConfig{
    Type:             mq.BrokerRedisStream,
    RedisStreamConfig: &redisstream.Config{Addr: "localhost:6379", Group: "my-group"},
})

Backend capabilities

Backend Persistent Ordered Transaction ConsumerGroup PubSub
RabbitMQ ✅ ✅ ✅ ❌ ✅
Kafka ✅ ✅ ✅ ✅ ✅
RocketMQ ✅ ✅ ✅ ✅ ✅
ActiveMQ ✅ ✅ ❌ ❌ ✅
Redis Streams ✅ ✅ ❌ ✅ ❌

Middleware

Middleware Description
RecoverMiddleware Recovers from handler panics
LoggingMiddleware Logs each delivery before/after
MetricsMiddleware Records consume/error/redelivered metrics
RetryMiddleware Retries failed deliveries with backoff
DeadLetterMiddleware Routes failures to a dead-letter handler
Chain Composes multiple middleware

Documentation

Overview

Package mq provides a broker-agnostic message queue abstraction with pluggable backends (RabbitMQ, Redis Streams, Kafka, NATS, …).

Architecture

  • Message: a unit of data with headers, payload, and metadata.
  • Delivery: a Message received by a consumer, with ACK/Nack/Reject semantics.
  • Producer: publishes messages to an exchange/topic.
  • Consumer: subscribes to a queue/topic and dispatches deliveries to a Handler.
  • Broker: manages connections, declares topology, and creates producers and consumers.
  • Handler: processes a Delivery.
  • Middleware:wraps Handlers for retry, logging, metrics, dead-letter, etc.

Backends

  • rabbitmq/ — RabbitMQ via amqp091-go (exchange/queue/binding, QoS, publisher confirms, auto-reconnect)

Basic usage 

broker, _ := rabbitmq.New(rabbitmq.Config{URL: "amqp://guest:guest@localhost:5672/"})
defer broker.Close()

producer, _ := broker.Producer("events", mq.PublishOptions{})
_ = producer.Publish(ctx, &mq.Message{
    Body: []byte(`{"event":"user.created"}`),
})

consumer, _ := broker.Consumer("events.queue", mq.ConsumeOptions{
    Handler: func(ctx context.Context, d mq.Delivery) error {
        fmt.Println(string(d.Body))
        return d.Ack()
    },
})
_ = consumer.Start(ctx)

Index

Constants

This section is empty.

Variables

View Source 
var ErrAlreadyRunning = errors.New("mq: consumer already running")

ErrAlreadyRunning is returned by Consumer.Start when the consumer is already running.

View Source 
var ErrClosed = errors.New("mq: broker closed")

ErrClosed is returned when an operation is attempted on a closed broker/producer/consumer.

View Source 
var ErrNoHandler = errors.New("mq: no handler configured")

ErrNoHandler is returned when a consumer is started without a handler.

View Source 
var ErrNotConnected = errors.New("mq: not connected")

ErrNotConnected is returned when the broker is not connected.

View Source 
var SupportedBrokers = []BrokerType{
	BrokerRabbitMQ,
	BrokerKafka,
	BrokerRocketMQ,
	BrokerActiveMQ,
	BrokerRedisStream,
}

SupportedBrokers lists all broker types the factory can create.

Functions

func DecodeJSON 

func DecodeJSON(d Delivery, target any) error

DecodeJSON decodes the delivery body into the target type.

func DecodeJSONBody 

func DecodeJSONBody(body []byte, target any) error

DecodeJSONBody decodes a raw message body into the target type.

func IsSupported 

func IsSupported(t BrokerType) bool

IsSupported reports whether the given broker type has a registered factory function.

func PublishBatch 

func PublishBatch(ctx context.Context, p Producer, msgs []*Message) error

PublishBatch is a fallback helper that publishes messages one by one using a standard Producer. Backends that support native batching should implement BatchPublisher directly.

Types

type BackendInfo 

type BackendInfo struct {
	Type          BrokerType
	Name          string
	Persistent    bool // survives broker restart
	Ordered       bool // preserves message ordering
	Transaction   bool // supports transactions
	ConsumerGroup bool // supports consumer groups
	PubSub        bool // supports pub/sub pattern
}

BackendInfo describes a backend implementation's capabilities.

type BatchPublisher 

type BatchPublisher interface {
	PublishBatch(ctx context.Context, msgs []*Message) error
}

BatchPublisher publishes multiple messages in a single broker transaction or network round-trip. Not all backends support batching; use PublishBatch as a fallback that calls Publish sequentially.

type Broker 

type Broker interface {
	// Connect establishes the connection to the broker.
	Connect() error

	// IsConnected reports whether the broker is currently connected.
	IsConnected() bool

	// Close shuts down the broker, closing all producers and consumers
	// and releasing the connection.
	Close() error

	// Producer creates or returns a cached producer for the given
	// exchange/topic.
	Producer(exchange string, opts PublishOptions) (Producer, error)

	// Consumer creates a consumer for the given queue with the given
	// options. The consumer is not started; call Start to begin.
	Consumer(queue string, opts ConsumeOptions) (Consumer, error)

	// DeclareExchange declares an exchange (RabbitMQ) or topic (Kafka).
	DeclareExchange(name string, opts ExchangeOptions) error

	// DeclareQueue declares a queue.
	DeclareQueue(name string, opts QueueOptions) error

	// Bind binds a queue to an exchange with a routing key.
	Bind(queue, exchange, routingKey string) error

	// Unbind removes a binding.
	Unbind(queue, exchange, routingKey string) error

	// DeleteQueue removes a queue.
	DeleteQueue(name string) error

	// DeleteExchange removes an exchange.
	DeleteExchange(name string) error
}

Broker manages connections and creates producers and consumers.

type BrokerConfig 

type BrokerConfig struct {
	Type BrokerType

	// RabbitMQ configuration. Required when Type == BrokerRabbitMQ.
	RabbitMQConfig any // *rabbitmq.Config

	// Kafka configuration. Required when Type == BrokerKafka.
	KafkaConfig any // *kafka.Config

	// RocketMQ configuration. Required when Type == BrokerRocketMQ.
	RocketMQConfig any // *rocketmq.Config

	// ActiveMQ configuration. Required when Type == BrokerActiveMQ.
	ActiveMQConfig any // *activemq.Config

	// RedisStream configuration. Required when Type == BrokerRedisStream.
	RedisStreamConfig any // *redisstream.Config
}

BrokerConfig is a union config that carries backend-specific settings. The factory selects the appropriate field based on BrokerType.

Usage: 

cfg := mq.BrokerConfig{
    Type:       mq.BrokerRabbitMQ,
    RabbitMQ:   &rabbitmq.Config{URL: "amqp://..."},
}
broker, err := factory.NewBroker(cfg)

type BrokerType 

type BrokerType string

BrokerType identifies a message-queue backend implementation. 

const (
	// BrokerRabbitMQ selects a RabbitMQ backend.
	BrokerRabbitMQ BrokerType = "rabbitmq"
	// BrokerKafka selects a Kafka backend.
	BrokerKafka BrokerType = "kafka"
	// BrokerRocketMQ selects a RocketMQ backend.
	BrokerRocketMQ BrokerType = "rocketmq"
	// BrokerActiveMQ selects an ActiveMQ backend (STOMP protocol).
	BrokerActiveMQ BrokerType = "activemq"
	// BrokerRedisStream selects a Redis Streams backend.
	BrokerRedisStream BrokerType = "redisstream"
)

func (BrokerType) String 

func (t BrokerType) String() string

String returns the broker type as a string.

type ConsumeOptions 

type ConsumeOptions struct {
	// Handler processes each delivery. Required.
	Handler Handler

	// Middleware wraps the handler. Applied in order: the first
	// middleware is the outermost.
	Middleware []Middleware

	// AutoAck: if true, messages are auto-acknowledged on delivery
	// (fire-and-forget). Default: false (manual ACK).
	AutoAck bool

	// QosPrefetchCount is the maximum number of unacknowledged
	// deliveries. Default: 10.
	QosPrefetchCount int

	// QosPrefetchSize is the maximum total bytes of unacknowledged
	// deliveries. Default: 0 (unlimited).
	QosPrefetchSize int

	// QosGlobal: if true, QoS applies to the entire channel, not just
	// the consumer. Default: false.
	QosGlobal bool

	// Concurrency is the number of goroutines processing deliveries.
	// Default: 1. Set > 1 for parallel processing.
	Concurrency int

	// ConsumerTag identifies this consumer in broker logs.
	ConsumerTag string

	// Exclusive: if true, only this consumer can access the queue.
	Exclusive bool

	// Args are additional broker-specific arguments (e.g. x-headers).
	Args map[string]any
}

ConsumeOptions configures a consumer.

type Consumer 

type Consumer interface {
	// Start begins consuming. The consumer runs until ctx is cancelled
	// or Stop is called. Start is idempotent if already running.
	Start(ctx context.Context) error

	// Stop gracefully stops consuming, waiting for in-flight handlers
	// to complete up to the given timeout. A zero timeout returns
	// immediately after signalling.
	Stop(timeout time.Duration) error

	// IsRunning reports whether the consumer is actively consuming.
	IsRunning() bool
}

Consumer subscribes to a queue and dispatches deliveries to a Handler.

type Delivery 

type Delivery interface {
	// Message returns the underlying message.
	Message() *Message

	// Body returns the raw payload (convenience shortcut).
	Body() []byte

	// Headers returns the message headers.
	Headers() map[string]any

	// RoutingKey returns the routing key.
	RoutingKey() string

	// Exchange returns the source exchange.
	Exchange() string

	// Redelivered reports whether this message has been delivered before.
	Redelivered() bool

	// DeliveryTag is the broker-specific delivery tag.
	DeliveryTag() uint64

	// Ack acknowledges successful processing. The message is removed
	// from the queue.
	Ack() error

	// Nack negatively acknowledges the message. If requeue is true, the
	// message is requeued for redelivery; otherwise it is discarded or
	// routed to a dead-letter exchange (if configured).
	Nack(requeue bool) error

	// Reject rejects the message. If requeue is true, the message is
	// requeued. Reject is similar to Nack but typically does not support
	// multiple flag.
	Reject(requeue bool) error
}

Delivery is a Message received by a consumer. It provides ACK/Nack/Reject semantics for at-least-once or exactly-once processing.

type DeliveryMode 

type DeliveryMode int

DeliveryMode selects message persistence. 

const (
	// Transient means the message is not persisted to disk.
	Transient DeliveryMode = 1
	// Persistent means the message is persisted to disk (survives broker restart).
	Persistent DeliveryMode = 2
)

type ExchangeOptions 

type ExchangeOptions struct {
	// Kind is the exchange type: "direct", "topic", "fanout", "headers".
	// Default: "topic".
	Kind string

	// Durable: if true, the exchange survives broker restarts.
	// Default: true.
	Durable bool

	// AutoDelete: if true, the exchange is deleted when no queues are
	// bound. Default: false.
	AutoDelete bool

	// Internal: if true, the exchange cannot be published to directly
	// (RabbitMQ). Default: false.
	Internal bool

	// NoWait: if true, do not wait for a broker confirmation.
	// Default: false.
	NoWait bool

	// Args are additional broker-specific arguments.
	Args map[string]any
}

ExchangeOptions configures an exchange declaration.

func DefaultExchangeOptions 

func DefaultExchangeOptions() ExchangeOptions

DefaultExchangeOptions returns sensible defaults for an exchange.

type Handler 

type Handler func(ctx context.Context, d Delivery) error

Handler processes a Delivery. Returning a nil error implies the message was processed successfully and the broker will ACK. Returning a non-nil error triggers Nack with requeue (or dead-letter, depending on config).

If the handler calls Ack/Nack/Reject explicitly, the framework will NOT auto-ACK. Use HandlerFunc to convert a function to Handler.

type HealthChecker 

type HealthChecker interface {
	Check(ctx context.Context) error
}

HealthChecker is implemented by brokers that support health checks. The Check method returns nil if the broker is healthy, or an error describing the problem.

type HealthStatus 

type HealthStatus struct {
	Healthy   bool          `json:"healthy"`
	Backend   string        `json:"backend"`
	Error     string        `json:"error,omitempty"`
	Latency   time.Duration `json:"latency,omitempty"`
	Connected bool          `json:"connected"`
}

HealthStatus describes the health of a broker.

func CheckHealth 

func CheckHealth(ctx context.Context, b Broker, backend string) HealthStatus

CheckHealth performs a health check with a timeout. If the broker implements HealthChecker, its Check method is called. Otherwise, IsConnected is used as a fallback.

func (HealthStatus) String 

func (h HealthStatus) String() string

String returns a human-readable health status.

type Message 

type Message struct {
	// ID is a unique identifier. If empty, the broker may generate one.
	ID string

	// Exchange is the target exchange (RabbitMQ) or topic (Kafka/NATS).
	// If empty, the producer's default exchange is used.
	Exchange string

	// RoutingKey is used by RabbitMQ to route the message.
	RoutingKey string

	// Headers are arbitrary key-value metadata.
	Headers map[string]any

	// Body is the raw message payload.
	Body []byte

	// ContentType is the MIME type of Body (e.g. "application/json").
	ContentType string

	// ContentEncoding is the encoding of Body (e.g. "gzip").
	ContentEncoding string

	// Priority is the message priority (0-9 for RabbitMQ).
	Priority uint8

	// CorrelationID is used for request/reply patterns.
	CorrelationID string

	// ReplyTo is the reply-to queue for RPC patterns.
	ReplyTo string

	// Expiration is the message TTL. Zero means no expiration.
	Expiration time.Duration

	// Timestamp is the message creation time. If zero, time.Now() is used.
	Timestamp time.Time

	// Type is a message type hint (optional).
	Type string

	// UserID is the user ID for RabbitMQ authenticated publishing.
	UserID string

	// AppID is the publishing application ID.
	AppID string

	// DeliveryMode selects persistent vs transient. Default: Persistent.
	DeliveryMode DeliveryMode
}

Message is a unit of data published to the broker.

func NewJSONMessage 

func NewJSONMessage(payload any) (*Message, error)

NewJSONMessage creates a Message with a JSON-encoded body and ContentType set to "application/json".

type Metrics 

type Metrics struct {
	Published    int64
	Consumed     int64
	Acked        int64
	Nacked       int64
	Rejected     int64
	Redelivered  int64
	Errors       int64
	AvgPublishMs float64
	AvgConsumeMs float64
}

Metrics is a point-in-time snapshot of broker observability data.

type MetricsCollector 

type MetricsCollector struct {
	// contains filtered or unexported fields
}

MetricsCollector is a thread-safe metrics collector.

func NewMetricsCollector 

func NewMetricsCollector() *MetricsCollector

NewMetricsCollector returns a fresh collector.

func (*MetricsCollector) RecordAck 

func (m *MetricsCollector) RecordAck()

func (*MetricsCollector) RecordConsume 

func (m *MetricsCollector) RecordConsume()

func (*MetricsCollector) RecordError 

func (m *MetricsCollector) RecordError()

func (*MetricsCollector) RecordNack 

func (m *MetricsCollector) RecordNack()

func (*MetricsCollector) RecordPublish 

func (m *MetricsCollector) RecordPublish()

MetricsCollector methods — defined here to keep mq.go focused on the public API.

func (*MetricsCollector) RecordRedelivered 

func (m *MetricsCollector) RecordRedelivered()

func (*MetricsCollector) RecordReject 

func (m *MetricsCollector) RecordReject()

func (*MetricsCollector) Snapshot 

func (m *MetricsCollector) Snapshot() Metrics

Snapshot returns a point-in-time copy of the metrics.

type Middleware 

type Middleware func(next Handler) Handler

Middleware wraps a Handler, adding cross-cutting behavior.

func Chain 

func Chain(mw ...Middleware) Middleware

Chain composes multiple middleware into a single middleware. The first middleware in the slice is the outermost.

func DeadLetterMiddleware 

func DeadLetterMiddleware(dlqHandler Handler) Middleware

DeadLetterMiddleware routes failed deliveries to a dead-letter handler and ACKs the original message so it is not redelivered.

func LoggingMiddleware 

func LoggingMiddleware(log *slog.Logger) Middleware

LoggingMiddleware logs each delivery before and after processing.

func MetricsMiddleware 

func MetricsMiddleware(collector *MetricsCollector) Middleware

MetricsMiddleware records delivery metrics in the given collector.

func RecoverMiddleware 

func RecoverMiddleware(log *slog.Logger) Middleware

RecoverMiddleware recovers from panics in the handler, logs the panic, and returns an error so the message is Nacked (and potentially requeued or dead-lettered).

func RetryMiddleware 

func RetryMiddleware(cfg RetryConfig) Middleware

RetryMiddleware retries failed deliveries up to MaxAttempts. After exhausting retries, the original error is returned so the message can be Nacked/dead-lettered.

Note: this middleware tracks attempts in-memory. For broker-level redelivery, rely on Nack(requeue=true) instead.

type Producer 

type Producer interface {
	// Publish sends a message. If msg.ID is empty, a UUID is generated.
	// If msg.Timestamp is zero, time.Now() is used.
	Publish(ctx context.Context, msg *Message) error

	// Close releases producer resources. After Close, Publish returns
	// ErrClosed.
	Close() error
}

Producer publishes messages to the broker.

type PublishOptions 

type PublishOptions struct {
	// Mandatory: if true, the broker returns an unroutable message
	// instead of silently dropping it.
	Mandatory bool

	// Immediate is a RabbitMQ-specific flag (deprecated in AMQP 0-9-1).
	Immediate bool

	// Persistent sets the default delivery mode for messages without
	// an explicit DeliveryMode.
	Persistent bool

	// Confirm enables publisher confirms (RabbitMQ).
	Confirm bool
}

PublishOptions configures a producer.

type QueueOptions 

type QueueOptions struct {
	// Durable: if true, the queue survives broker restarts.
	// Default: true.
	Durable bool

	// AutoDelete: if true, the queue is deleted when the last consumer
	// disconnects. Default: false.
	AutoDelete bool

	// Exclusive: if true, the queue is only accessible by the declaring
	// connection and is deleted when the connection closes.
	// Default: false.
	Exclusive bool

	// NoWait: if true, do not wait for a broker confirmation.
	// Default: false.
	NoWait bool

	// Args are additional broker-specific arguments (e.g.
	// "x-message-ttl", "x-dead-letter-exchange").
	Args map[string]any
}

QueueOptions configures a queue declaration.

func DefaultQueueOptions 

func DefaultQueueOptions() QueueOptions

DefaultQueueOptions returns sensible defaults for a queue.

func WithDeadLetter 

func WithDeadLetter(opts QueueOptions, dlxExchange, dlxRoutingKey string) QueueOptions

WithDeadLetter configures a queue to route dead-lettered messages to the specified exchange and routing key. This is RabbitMQ-specific (x-dead-letter-exchange / x-dead-letter-routing-key arguments) but the pattern is common across brokers.

func WithMaxPriority 

func WithMaxPriority(opts QueueOptions, maxPriority int) QueueOptions

WithMaxPriority configures a queue to support priority messages. The maxPriority value should be between 1 and 255.

func WithMessageTTL 

func WithMessageTTL(opts QueueOptions, ttl time.Duration) QueueOptions

WithMessageTTL configures a queue with a message time-to-live. Messages that expire are dead-lettered (if DLX is configured) or discarded.

func WithQueueTTL 

func WithQueueTTL(opts QueueOptions, ttl time.Duration) QueueOptions

WithQueueTTL configures a queue that is automatically deleted after the specified period of inactivity (no consumers).

type RetryConfig 

type RetryConfig struct {
	// MaxAttempts is the maximum number of delivery attempts (including
	// the first). Default: 3.
	MaxAttempts int

	// Backoff is the delay between retries. If BackoffFn is set, it
	// takes precedence. Default: 1s.
	Backoff time.Duration

	// BackoffFn is an optional function that computes the backoff for
	// a given attempt (1-based). If set, Backoff is ignored.
	BackoffFn func(attempt int) time.Duration
}

RetryConfig configures RetryMiddleware.

func DefaultRetryConfig 

func DefaultRetryConfig() RetryConfig

DefaultRetryConfig returns sensible retry defaults.

type Topology 

type Topology struct {
	// contains filtered or unexported fields
}

Topology is a fluent builder for declaring exchanges, queues, and bindings on a broker. It collects declarations and applies them in order when Apply is called.

func NewTopology 

func NewTopology(b Broker) *Topology

NewTopology creates a new Topology builder for the given broker.

func (*Topology) Apply 

func (t *Topology) Apply() error

Apply executes all collected declarations in order: exchanges first, then queues, then bindings. Returns the first error encountered.

func (*Topology) Bind 

func (t *Topology) Bind(queue, exchange, routingKey string) *Topology

Bind binds a queue to an exchange.

func (*Topology) Exchange 

func (t *Topology) Exchange(name string, opts ...ExchangeOptions) *Topology

Exchange declares an exchange.

func (*Topology) MustApply 

func (t *Topology) MustApply()

MustApply is like Apply but panics on error.

func (*Topology) Queue 

func (t *Topology) Queue(name string, opts ...QueueOptions) *Topology

Queue declares a queue.

func (*Topology) Reset 

func (t *Topology) Reset() *Topology

Reset clears all collected declarations.

Source Files

View all Source files

Directories

Path Synopsis
activemq module
kafka module
rabbitmq module
redisstream module
rocketmq module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.