Docs / Getting started

Getting started

Install the library and one backend module, then run a consumer and a producer

mqutils puts one consumer and producer interface across seven backend modules. This guide installs the core module plus one backend, then walks through a consumer, a producer, batching, health checks and shutdown.

Upgrading from v1? See Upgrading from v1: v2 changed the import paths and the handler contract.

Installation

Add the core library and the backend module(s) you need to your Go project:

bash

go get go.digitalxero.dev/mqutils/v2

# One or more backends
go get go.digitalxero.dev/mq-amqp/v2
go get go.digitalxero.dev/mq-kafka/v2
go get go.digitalxero.dev/mq-nats/v2
go get go.digitalxero.dev/mq-aws/v2
go get go.digitalxero.dev/mq-gcp/v2
go get go.digitalxero.dev/mq-redis/v2

The RabbitMQ Stream module requires Go 1.25 and has separate availability and local-checkout instructions while its new vanity mappings and release are prepared. Existing module imports above are unaffected.

Basic Usage

Creating a Consumer

The simplest way to start using mqutils is with the factory function and a viper configuration:

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
    _ "go.digitalxero.dev/mq-amqp/v2" // register the amqp:// scheme
)

func main() {
    // Register a message handler: return nil to acknowledge,
    // an error to reject (triggering retry/dead-letter handling)
    types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
        log.Printf("Received: %s", string(msg.Body()))
        return nil
    })

    // Configure the consumer; the URL scheme selects the backend
    config := viper.New()
    config.Set("url", "amqp://localhost:5672/")
    config.Set("queue", "myqueue")
    config.Set("handler", "process")

    ctx := context.Background()
    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    // Start consuming messages (blocks until the context is canceled)
    if err := consumer.Run(ctx); err != nil {
        log.Fatal(err)
    }
}

The Typed Builder

If you prefer not to use viper, the typed builder covers the configuration shared by every backend:

consumer, err := mqutils.NewConsumerBuilder().
    WithURL("amqp://localhost:5672/").
    WithDestination("myqueue").
    WithMaxRetries(3).
    WithHandler(func(ctx context.Context, msg types.Message) error {
        log.Printf("Received: %s", string(msg.Body()))
        return nil
    }).
    Build(ctx)

URL Schemes

mqutils automatically detects the message queue system from the URL scheme. Consumers and producers register identical scheme sets:

System	URL Schemes	Example
AMQP/RabbitMQ	`amqp://`, `amqps://`	`amqp://localhost:5672/`
RabbitMQ Streams	`rabbitmq-stream://`, `rabbitmq-stream+tls://`	`rabbitmq-stream://localhost:5552`
Kafka	`kafka://`, `kafkas://`	`kafka://localhost:9092`
NATS	`nats://`, `natss://`, `tls://`, `jetstream://`	`nats://localhost:4222`
AWS SQS	`sqs://`, `sqss://`	`sqs://region/queue-name`
GCP Pub/Sub	`pubsub://`, `pubsubs://`, `gcp://`	`pubsub://project-id/topic?subscription=sub`
Redis	`redis://`, `rediss://`, `redisstream://`, `redisstreams://`	`redis://localhost:6379/mychannel`

Every backend also accepts the canonical config keys destination, max_retries, and consumer_group alongside its native keys.

Core Concepts

Messages

All messages in mqutils implement the types.Message interface. The most commonly used methods:

type Message interface {
    MessageId() string               // Unique message identifier
    CorrelationId() string           // For request-response patterns
    ReplyTo() string                 // Reply destination
    Exchange() string                // AMQP exchange (empty for other systems)
    RoutingKey() string              // Message routing key
    Headers() map[string]interface{} // Message headers
    Body() []byte                    // Message payload
    Publisher() Publisher            // Access to publisher for replies
    Ack() error                      // Optional manual acknowledgment
    Nack() error                     // Optional manual rejection
    // ... additional metadata methods
}

Message Handlers

Handlers return an error: a nil return acknowledges the message, a non-nil return rejects it. The consumer runtime settles the message for you.

// Simple handler
types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
    // Process the message
    log.Printf("Processing: %s", string(msg.Body()))
    return nil
})

// Handler with reply
types.RegisterHandler("request", func(ctx context.Context, msg types.Message) error {
    // Process request and send reply
    response := []byte("processed")

    if msg.ReplyTo() != "" {
        return msg.Publisher().Publish(ctx,
            msg.CorrelationId(), // correlation ID
            "",                  // exchange (empty for default)
            msg.ReplyTo(),       // routing key (reply destination)
            "text/plain",        // content type
            response)            // body
    }

    return nil
})

Batch Processing

Batch processing works on every backend. A nil return settles the whole batch; an error rejects every message the handler did not settle itself.

// Register a batch handler globally
types.RegisterBatchHandler("bulk-process", func(ctx context.Context, msgs []types.Message) error {
    log.Printf("Processing batch of %d messages", len(msgs))

    for _, msg := range msgs {
        // Process each message in the batch
        log.Printf("Message: %s", string(msg.Body()))
    }

    return nil
})

// Configure consumer with batch processing
config := viper.New()
config.Set("url", "amqp://localhost:5672/")
config.Set("queue", "myqueue")
config.Set("handler", "bulk-process")
config.Set("enable_batch_processing", true)
config.Set("batch_size", 50)
config.Set("batch_timeout", "1s") // duration string

consumer, err := mqutils.NewConsumer(ctx, config)

Health Monitoring

All consumers implement types.HealthChecker:

// Check health status
health, err := consumer.HealthCheck(context.Background())
if err != nil {
    log.Printf("Health check failed: %v", err)
    return
}

if health.Status() == types.HealthStatusHealthy {
    log.Println("Consumer is healthy")
} else {
    log.Printf("Consumer health: %s - %s", health.Status(), health.Message())
}

Publishing Messages

Create a producer with the factory function (or mqutils.NewProducerBuilder()):

config := viper.New()
config.Set("url", "amqp://localhost:5672/")

producer, err := mqutils.NewProducer(context.Background(), config)
if err != nil {
    log.Fatal(err)
}
if err := producer.Start(context.Background()); err != nil {
    log.Fatal(err)
}

// Simple publish
err = producer.Publish(
    context.Background(),
    "req-123",               // correlation ID
    "",                      // exchange (empty for default)
    "destination",           // routing key/destination
    "text/plain",            // content type
    []byte("Hello, World!"), // body
)

// Publish with message builder
import "go.digitalxero.dev/mq-amqp/v2"

message := amqp.NewMessageBuilder().
    WithCorrelationId("req-123").
    WithContentType("application/json").
    WithHeaders(map[string]interface{}{
        "priority": 1,
    }).
    WithBody([]byte(`{"event": "user.created", "id": 123}`)).
    Build()

err = producer.PublishMsg(context.Background(), "events", "user.created", message)

Configuration Examples

AMQP/RabbitMQ

config := viper.New()
config.Set("url", "amqp://user:pass@localhost:5672/")
config.Set("queue", "queue")
config.Set("exchange", "events")
config.Set("routing_key", "user.created")
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)

Kafka

config := viper.New()
config.Set("url", "kafka://localhost:9092")
config.Set("topic", "mytopic")
config.Set("consumer_group", "my-service")
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)

For authentication and atomic producer writes, see Kafka security and transactions. For native RabbitMQ Stream publishing, durable offsets, routing, and deduplication, see the RabbitMQ Streams guide.

AWS SQS

config := viper.New()
config.Set("url", "sqs://us-east-1/my-queue")       // or my-queue.fifo
config.Set("handler", "process")
consumer, err := mqutils.NewConsumer(ctx, config)

Error Handling

mqutils provides structured error handling driven by the handler’s return value:

types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
    // Your processing logic here
    if err := processMessage(ctx, msg); err != nil {
        // Return error to trigger message retry/dead letter
        return fmt.Errorf("processing failed: %w", err)
    }

    return nil // Message will be acknowledged automatically
})

// Manual settlement still works when you need to settle mid-handler.
// The runtime treats the duplicate settlement as success.
types.RegisterHandler("manual", func(ctx context.Context, msg types.Message) error {
    if err := processMessage(ctx, msg); err != nil {
        // Explicitly nack the message
        return msg.Nack()
    }

    // Explicitly ack the message
    return msg.Ack()
})

Graceful Shutdown

Always implement graceful shutdown:

func main() {
    // Register handler first
    types.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
        log.Printf("Processing: %s", string(msg.Body()))
        return nil
    })

    // Enable graceful shutdown in configuration
    config := viper.New()
    config.Set("url", "amqp://localhost:5672/")
    config.Set("queue", "myqueue")
    config.Set("handler", "process")
    config.Set("enable_graceful_shutdown", true)
    config.Set("graceful_shutdown_timeout", 30) // seconds

    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    consumer, err := mqutils.NewConsumer(ctx, config)
    if err != nil {
        log.Fatal(err)
    }

    go func() {
        sigChan := make(chan os.Signal, 1)
        signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
        <-sigChan
        log.Println("Shutdown signal received, stopping consumer...")
        cancel()
    }()

    // Run will drain in-flight messages and return when the context is canceled
    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer error: %v", err)
    }
}

Note: if the broker connection is lost and auto_reconnect is disabled, Run returns an error. Set auto_reconnect to true to reconnect with backoff instead.

Next steps

Core types for the interface reference
Examples for a working program per backend
The backend pages under Docs for every configuration key with its default
Source and issues on GitLab