amqpadapter

README

go-amqp-adapter

Go adapter over amqp091-go: publish and consume RabbitMQ messages with publisher confirms, auto-reconnect, and optional retry via a delayed queue / DLX.

go get github.com/ivan-makarenkov/go-amqp-adapter

Package: amqpadapter.

Together with go-failedjobs

This library and go-failedjobs were designed to be used together as a replacement for Laravel's retry-through-queue mechanism (tries / delayed retries → failed_jobs → php artisan queue:retry).

• go-amqp-adapter handles in-broker retries (delay queue / DLX) — the analogue of Laravel's automatic job retries.
• When retries are exhausted (or the handler returns a non-retryable error), WithFailHandler persists the payload instead of Laravel's failed_jobs table. Wire it to go-failedjobs via GetFailedJobHandler.
• go-failedjobs stores those rows in MySQL/Postgres and republishes selected IDs to RabbitMQ (POST /retry-task) — the analogue of php artisan queue:retry.

Features

• Lazy publisher connect on first Publish
• Publisher confirms (one in-flight publish per connection)
• Automatic reconnect on connection/channel loss
• Consumers with parallelism (AddConsumerN)
• Retry via delay queue and expired header (requires WithFailHandler)
• Header/context plumbing (WithPublishHeadersBuilder / WithConsumeHeadersExtractor)
• otel subpackage for correlation ID and OpenTelemetry propagation

Examples

Runnable programs live in examples/:

docker compose -f examples/docker-compose.yml up -d
go run -C examples ./basic-consumer

Directory	Topic
basic-consumer	Publish and consume one queue
publisher-confirms	Publish waits for broker ack
retry	Delay-queue retries and WithFailHandler
multiple-consumers	AddConsumerN parallelism
failed-jobs	Persist permanent failures and republish
graceful-shutdown	Shutdown waits for in-flight handlers

See examples/README.md for details.

Quick start

package main

import (
	"context"
	"log"
	"time"

	mq "github.com/ivan-makarenkov/go-amqp-adapter"
)

func main() {
	ctx := context.Background()

	q, err := mq.New(mq.Config{
		URL:            "amqp://guest:guest@localhost:5672/",
		ReconnectDelay: time.Second,
		ReInitDelay:    time.Second,
		ResendDelay:    time.Second,
		QueueParams: map[mq.QueueName]mq.QueueItem{
			"jobs": {},
		},
	})
	if err != nil {
		log.Fatal(err)
	}
	defer func() {
		shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
		defer cancel()
		_ = q.Shutdown(shutdownCtx)
	}()

	err = q.AddConsumer(ctx, "jobs", func(ctx context.Context, body []byte) error {
		log.Printf("got: %s", body)
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	if err = q.InitConsumer(ctx); err != nil {
		log.Fatal(err)
	}

	err = q.Publish(ctx, "jobs", mq.PublishMessage{Body: []byte("hello")})
	if err != nil {
		log.Fatal(err)
	}

	select {}
}

Configuration

Field	Meaning
URL	AMQP URL
ReconnectDelay	Pause between dial attempts
ReInitDelay	Pause between channel/queue re-init attempts
ResendDelay	Pause before republish after error/nack
QueueParams	Per-queue parameters by name

QueueItem:

• Retry — nil = no retry; otherwise delay/DLX topology
• ConsumerOnly — consume only, no publisher connection

Retry

When Retry != nil, the library declares exchanges and a delay queue. For reprocessing, the handler must return amqpadapter.Retry(err).

After MaxDuration is exhausted (expired header) or a non-retryable error, WithFailHandler is called and the message is acked (removed from the queue).

q, err := mq.New(conf,
	mq.WithFailHandler(func(job mq.FailedJob) error {
		log.Printf("failed %s: %v", job.Queue, job.Err)
		return nil
	}),
)

// in handler:
return mq.Retry(err) // delay and retry
return err           // final fail (in retry mode)

WithFailHandler is required if at least one queue has Retry set.

Consumers

_ = q.AddConsumer(ctx, "jobs", handler)      // 1 worker
_ = q.AddConsumerN(ctx, "jobs", 4, handler) // 4 connections/workers
_ = q.InitConsumer(ctx)                     // start read loops

AddConsumer* is only allowed before InitConsumer. Prefetch: QoS = 1 per channel.

AddConsumerN(N) creates N separate AMQP connections and channels (one client per worker). That gives good parallelism and failure isolation, at the cost of more TCP connections to RabbitMQ compared to a single connection with N goroutines reading one channel.

Shutdown

Shutdown closes connections first (stops intake), then waits for in-flight handlers (inflight) within the context deadline.

Options

• WithLogger — custom logger (otherwise noop); *slog.Logger satisfies the interface:

import "log/slog"

q, err := mq.New(conf, mq.WithLogger(slog.Default()))

• WithFailHandler — permanent failures in retry mode
• WithPublishHeadersBuilder — headers from context on publish
• WithConsumeHeadersExtractor — restore context from headers

Example with otel:

import "github.com/ivan-makarenkov/go-amqp-adapter/otel"

cfg := otel.PropagationConfig{
	CorrelationIDKey: "x-correlation-id",
	TraceKeys:        []string{"traceparent", "tracestate"},
	// GetCorrelationID / SetCorrelationID / Propagator — as needed
}

q, err := mq.New(conf,
	mq.WithPublishHeadersBuilder(otel.NewPublishHeadersBuilder(cfg)),
	mq.WithConsumeHeadersExtractor(otel.NewConsumeHeadersExtractor(cfg)),
)

Tests

go test ./...                  # unit
go test -short ./...           # skip integration
# functional (needs RabbitMQ), see tests/ and Makefile

Documentation

Overview

Package amqpadapter provides a production-oriented RabbitMQ adapter for Go.

It provides publishing and consuming of RabbitMQ messages with automatic connection recovery, publisher confirms, concurrent consumers, delayed retries using RabbitMQ delay queues or dead-letter exchanges, graceful shutdown, context propagation, correlation IDs and OpenTelemetry support.

The package is intended for Go services that need reliable RabbitMQ message processing without implementing connection recovery and retry infrastructure in application code.

Index

• Constants
• Variables
• func IsRetryable(err error) bool
• func Retry(err error) error
• type Config
• type ConsumeHeadersExtractor
• type ConsumerHandler
• type FailJobHandler
• type FailedJob
• type Logger
• type Option
  • func WithConsumeHeadersExtractor(e ConsumeHeadersExtractor) Option
  • func WithFailHandler(h FailJobHandler) Option
  • func WithLogger(lgr Logger) Option
  • func WithPublishHeadersBuilder(b PublishHeadersBuilder) Option
• type PublishHeadersBuilder
• type PublishMessage
• type PublishMessagePriority
• type Queue
  • func New(conf Config, opts ...Option) (Queue, error)
• type QueueItem
• type QueueName
• type RetryConfig
• type RetryableError
  • func (e *RetryableError) Error() string
  • func (e *RetryableError) Unwrap() error

Constants

const DefaultContentType = "text/plain"

DefaultContentType is the default content type for published messages.

Variables

var ErrBindQueueToExchange = errors.New("error binding queue to exchange")

ErrBindQueueToExchange is returned when binding a queue to an exchange fails.

var ErrClientClosedBeforeReady = errors.New("client closed before ready")

ErrClientClosedBeforeReady is returned when the client is closed before becoming ready.

var ErrClosingQueueChannel = errors.New("error closing queue channel")

ErrClosingQueueChannel is returned when closing the queue channel fails.

var ErrClosingQueueConnection = errors.New("error closing queue connection")

ErrClosingQueueConnection is returned when closing the queue connection fails.

var ErrConnectToQueue = errors.New("failed to connect to queue at address")

ErrConnectToQueue is returned when dialing the broker fails.

var ErrConsumeQueueConnectionClosed = errors.New("queue connection closed during consume")

ErrConsumeQueueConnectionClosed is returned when consuming on a closed connection.

var ErrConsumersAlreadyStarted = errors.New("consumers already started")

ErrConsumersAlreadyStarted is returned when InitConsumer was already called.

var ErrContextCanceledBeforeAck = errors.New("context canceled before publish confirm")

ErrContextCanceledBeforeAck is returned when the context is canceled before publish confirm.

var ErrContextCanceledOnConsumerStart = errors.New("context canceled while starting consumers")

ErrContextCanceledOnConsumerStart is returned when the context is canceled during InitConsumer.

var ErrDoneSignalBeforeAck = errors.New("done signal received before publish confirm")

ErrDoneSignalBeforeAck is returned when the client is closed before publish confirm.

var ErrEmptyURL = errors.New("RabbitMQ URL is not set")

ErrEmptyURL is returned when the RabbitMQ URL is empty.

var ErrEnableConfirms = errors.New("failed to enable publisher confirms")

ErrEnableConfirms is returned when enabling publisher confirms fails.

var ErrFailHandlerRequired = errors.New("retry enabled but WithFailHandler was not provided")

ErrFailHandlerRequired is returned when retry is enabled but WithFailHandler is missing.

var ErrInvalidDelay = errors.New("config delay must be greater than zero")

ErrInvalidDelay is returned when a configured delay is not positive.

var ErrInvalidPriority = errors.New("message priority must be in range 0–9")

ErrInvalidPriority is returned when message priority is outside 0–9.

var ErrInvalidRetryConfig = errors.New("invalid retry configuration")

ErrInvalidRetryConfig is returned when retry parameters are invalid.

var ErrNoConsumersRegistered = errors.New("no consumers registered")

ErrNoConsumersRegistered is returned when InitConsumer is called with no consumers.

var ErrOpenChannel = errors.New("failed to open queue channel")

ErrOpenChannel is returned when opening an AMQP channel fails.

var ErrPublishFailed = errors.New("error publishing message to queue")

ErrPublishFailed is returned when publishing a message fails.

var ErrPublishNack = errors.New("broker nacked the published message")

ErrPublishNack is returned when the broker nacks a publish (publisher confirm).

var ErrPushQueueConnectionClosed = errors.New("queue connection closed during publish")

ErrPushQueueConnectionClosed is returned when publishing on a closed connection.

var ErrQueueConnectionClosed = errors.New("queue connection closed")

ErrQueueConnectionClosed is returned when the queue connection is closed.

var ErrQueueConsumerOnly = errors.New("queue is configured for consume only (ConsumerOnly)")

ErrQueueConsumerOnly is returned when publishing to a ConsumerOnly queue.

var ErrQueueNotFound = errors.New("queue not found in configuration")

ErrQueueNotFound is returned when the queue is missing from configuration.

var ErrStartingQueueConsumption = errors.New("error starting queue consumption")

ErrStartingQueueConsumption is returned when starting consume fails.

Functions

func IsRetryable

func IsRetryable(err error) bool

IsRetryable reports whether err is marked as retryable.

func Retry

func Retry(err error) error

Retry wraps err as retryable for ConsumerHandler.

Types

type Config

type Config struct {
	URL            string
	ReconnectDelay time.Duration
	ReInitDelay    time.Duration
	ResendDelay    time.Duration
	QueueParams    map[QueueName]QueueItem
}

Config holds RabbitMQ connection settings and per-queue parameters.

type ConsumeHeadersExtractor

type ConsumeHeadersExtractor func(parent context.Context, headers map[string]any) context.Context

ConsumeHeadersExtractor restores context from AMQP headers on consume. parent is the consumer-loop context (from InitConsumer); the returned context must be derived from it.

type ConsumerHandler

type ConsumerHandler func(ctx context.Context, data []byte) error

ConsumerHandler processes a message body. Return mq.Retry(err) to request delayed reprocessing.

type FailJobHandler

type FailJobHandler func(job FailedJob) error

FailJobHandler is called on permanent failure in retry mode.

type FailedJob

type FailedJob struct {
	Queue QueueName
	Body  []byte
	Err   error
}

FailedJob describes a job that failed after retries were exhausted.

type Logger

type Logger interface {
	Debug(msg string, args ...any)
	Info(msg string, args ...any)
	Warn(msg string, args ...any)
	Error(msg string, args ...any)
	DebugContext(ctx context.Context, msg string, args ...any)
	InfoContext(ctx context.Context, msg string, args ...any)
	WarnContext(ctx context.Context, msg string, args ...any)
	ErrorContext(ctx context.Context, msg string, args ...any)
}

Logger is a minimal logging interface with slog-style method names.

type Option

type Option func(*options)

Option configures Queue at New time.

func WithConsumeHeadersExtractor

func WithConsumeHeadersExtractor(e ConsumeHeadersExtractor) Option

WithConsumeHeadersExtractor sets the function that restores context from AMQP headers on consume.

func WithFailHandler

func WithFailHandler(h FailJobHandler) Option

WithFailHandler sets the permanent-failure handler (required when QueueItem.Retry is set).

func WithLogger

func WithLogger(lgr Logger) Option

WithLogger sets the logger; the default is a noop logger.

func WithPublishHeadersBuilder

func WithPublishHeadersBuilder(b PublishHeadersBuilder) Option

WithPublishHeadersBuilder sets the function that builds AMQP headers on publish.

type PublishHeadersBuilder

type PublishHeadersBuilder func(ctx context.Context) map[string]any

PublishHeadersBuilder builds AMQP headers from context at publish time.

type PublishMessage

type PublishMessage struct {
	Body             []byte
	ContentType      string
	MaxRetryDuration *time.Duration
	Priority         *PublishMessagePriority
}

PublishMessage describes the body and options of a published message.

type PublishMessagePriority

type PublishMessagePriority uint8

PublishMessagePriority is the RabbitMQ message priority (0–9).

type Queue

type Queue interface {
	Publish(ctx context.Context, queue QueueName, msg PublishMessage) error
	AddConsumer(ctx context.Context, queue QueueName, handler ConsumerHandler) error
	AddConsumerN(ctx context.Context, queue QueueName, parallelism int, handler ConsumerHandler) error
	InitConsumer(ctx context.Context) error
	Shutdown(ctx context.Context) error
}

Queue is the public contract for publish, consume registration, and shutdown.

func New

func New(conf Config, opts ...Option) (Queue, error)

New creates and initializes a Queue from config and options.

type QueueItem

type QueueItem struct {
	// Retry enables retry mode; nil means no retry.
	Retry *RetryConfig
	// ConsumerOnly means consume-only (no publisher connection).
	ConsumerOnly bool
}

QueueItem describes processing options for a single queue.

type QueueName

type QueueName string

QueueName is a typed RabbitMQ queue name.

type RetryConfig

type RetryConfig struct {
	Delay       time.Duration
	MaxDuration time.Duration
}

RetryConfig describes retry parameters for a queue.

type RetryableError

type RetryableError struct {
	Err error
}

RetryableError wraps an error to signal that the consumer loop should retry.

func (*RetryableError) Error

func (e *RetryableError) Error() string

func (*RetryableError) Unwrap

func (e *RetryableError) Unwrap() error