Docs / Examples

Examples

Comprehensive examples for all supported message queue systems

Every example on this page targets the v2 API:

Import paths carry a /v2 suffix: go.digitalxero.dev/mqutils/v2, go.digitalxero.dev/mqutils/v2/types, and one go.digitalxero.dev/mq-<backend>/v2 module per broker. Blank-import a backend module to register its URL schemes.
Handlers are func(ctx context.Context, msg types.Message) error. Return nil to acknowledge the message, return an error to reject it. The shared consumer runtime settles the message for you, so there is no Ack()/Nack() bookkeeping in handlers.
Consumers are built with mqutils.NewConsumerBuilder() (typed, no viper) or mqutils.NewConsumer(ctx, viperConfig). Producers use mqutils.NewProducerBuilder() or mqutils.NewProducer(ctx, viperConfig) and must be started with Start(ctx) before publishing. The Kafka-specific transactional builder is an exception: its Build connects immediately.
The canonical keys destination, max_retries, and consumer_group are accepted by every backend alongside its native keys. Native keys win when both are set.

Working samples also live in the repository’s _examples/ directory.

Quick Start

Basic Consumer (typed builder)

package main

import (
    "context"
    "log"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-amqp/v2" // registers amqp:// and amqps://
)

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

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        WithDestination("my-queue"). // canonical key -> "queue" on AMQP
        WithHandler(func(ctx context.Context, msg types.Message) error {
            log.Printf("Received: %s", string(msg.Body()))
            return nil // nil acknowledges; an error rejects
        }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    // Run blocks until ctx is cancelled or an unrecoverable error occurs.
    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Basic Consumer (viper configuration)

Use viper when configuration comes from a file or environment, or when you need backend-native keys. Handlers referenced by name must be registered before NewConsumer is called; the backend validates the name at construction time.

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"
)

func main() {
    // Register the handler first so the consumer can resolve it by name.
    mqutils.RegisterHandler("process", func(ctx context.Context, msg types.Message) error {
        log.Printf("Received: %s", string(msg.Body()))
        return nil
    })

    config := viper.New()
    config.Set("url", "amqp://guest:guest@localhost:5672/")
    config.Set("queue", "my-queue")
    config.Set("handler", "process")

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

mqutils.RegisterHandler is an alias for types.RegisterHandler; either works. The typed builder’s WithHandler registers the function for you, and WithHandlerName references a handler registered elsewhere.

Basic Producer

package main

import (
    "context"
    "log"

    "go.digitalxero.dev/mq-amqp/v2"
    "go.digitalxero.dev/mqutils/v2"
)

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

    producer, err := mqutils.NewProducerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    // Start establishes the connection; it must be called before publishing.
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // Publish(ctx, correlationID, exchange, topic, contentType, body)
    err = producer.Publish(ctx,
        "req-123",               // correlation ID
        "events",                // exchange
        "greeting.created",      // routing key
        "text/plain",            // content type
        []byte("Hello, World!"), // body
    )
    if err != nil {
        log.Fatal(err)
    }

    // PublishMsg takes a pre-built Message for full control over properties.
    msg := amqp.NewMessageBuilder().
        WithCorrelationId("req-124").
        WithContentType("application/json").
        WithHeaders(map[string]any{"type": "greeting"}).
        WithBody([]byte(`{"text": "Hello from mqutils!"}`)).
        Build()

    if err := producer.PublishMsg(ctx, "events", "greeting.created", msg); err != nil {
        log.Fatal(err)
    }
}

What `exchange` and `topic` mean per backend

Publish(ctx, correlationID, exchange, topic, contentType, body) and PublishMsg(ctx, exchange, topic, msg) share one signature across backends, but the two positional parameters map onto each broker’s own concepts:

Backend	`exchange` parameter	`topic` parameter
AMQP	Exchange (empty = configured `exchange`)	Routing key
Kafka	Topic (empty = configured `topic`)	Message key (partitioning; empty = default partitioning)
NATS	Subject	Reply-to subject
AWS SQS	Queue URL (empty = queue from the connection URL)	FIFO message group ID (ignored on standard queues)
GCP Pub/Sub	Topic ID (empty = configured `topic_id`)	Ordering key (requires `enable_message_ordering`)
Redis Pub/Sub	Channel (empty = configured `channel`)	Ignored
Redis Streams	Ignored	Stream key (empty = configured `stream_key`)

Canonical configuration keys

Backend	`destination` (consumer)	`destination` (producer)	`max_retries`	`consumer_group`
AMQP	`queue`	`exchange`	`retry_queue_max_retries`	—
Kafka	`topic`	`topic`	`retry_max_retries`	`consumer_group` (native)
NATS	`subject`	`stream_name`	`max_deliver`	`queue_group`
AWS SQS	`queue_name`	`queue_url`	`max_retries` (native)	—
GCP Pub/Sub	`topic_id`	`topic_id`	`max_delivery_attempts`	`subscription_id`
Redis	`channel_name` / `stream_name`	`channel` / `stream_key`	`max_retries` (native)	`consumer_group` (native, Streams only)

Backends without a competing-consumer concept (AMQP, SQS) accept consumer_group but do not use it.

AMQP / RabbitMQ

Consumer bound to an exchange

package main

import (
    "context"
    "encoding/json"
    "log"
    "os/signal"
    "syscall"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-amqp/v2"
)

type order struct {
    ID     string  `json:"id"`
    Amount float64 `json:"amount"`
}

func main() {
    mqutils.RegisterHandler("order_processor", func(ctx context.Context, msg types.Message) error {
        var o order
        if err := json.Unmarshal(msg.Body(), &o); err != nil {
            // A malformed payload can never succeed; acknowledge it away
            // instead of spending the retry budget on it.
            msg.Logger(ctx).Warn("dropping malformed order")
            return nil
        }

        log.Printf("Processing order %s for %.2f", o.ID, o.Amount)
        return nil
    })

    config := viper.New()
    config.Set("url", "amqp://guest:guest@localhost:5672/")
    config.Set("queue", "orders")
    config.Set("exchange", "events")
    config.Set("exchange_type", "topic")
    config.Set("routing_key", "order.*")
    config.Set("auto_declare", true) // declare the queue/exchange/binding if missing
    config.Set("auto_reconnect", true)
    config.Set("handler", "order_processor")

    // Cancel the context on SIGINT/SIGTERM; Run drains in-flight messages.
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Publishing through an exchange

package main

import (
    "context"
    "log"

    "go.digitalxero.dev/mqutils/v2"

    _ "go.digitalxero.dev/mq-amqp/v2"
)

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

    producer, err := mqutils.NewProducerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        WithDestination("events"). // canonical key -> default "exchange"
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    events := []struct {
        routingKey string
        body       string
    }{
        {"user.created", `{"event": "user_created", "id": 123}`},
        {"user.updated", `{"event": "user_updated", "id": 123}`},
        {"order.placed", `{"event": "order_placed", "order_id": 456}`},
    }

    for _, ev := range events {
        // An empty exchange argument falls back to the configured destination.
        err := producer.Publish(ctx, "", "", ev.routingKey, "application/json", []byte(ev.body))
        if err != nil {
            log.Printf("Failed to publish %s: %v", ev.routingKey, err)
        }
    }
}

Publisher confirms

By default AMQP publishes are fire-and-forget. With publisher_confirms enabled every publish waits for the broker’s confirmation and returns types.ErrPublishNacked if the broker rejects it. While RabbitMQ has the connection blocked (memory/disk alarm), confirmed publishes fail immediately with types.ErrConnectionBlocked; fire-and-forget publishes are queued internally (publish_queue_size) and flushed in order once the block lifts.

package main

import (
    "context"
    "errors"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-amqp/v2"
)

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

    native := viper.New()
    native.Set("publisher_confirms", true)
    native.Set("confirm_timeout", "5s")

    producer, err := mqutils.NewProducerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        WithDestination("events").
        WithConfig(native). // backend-specific keys merge underneath typed values
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    err = producer.Publish(ctx, "req-1", "", "order.placed", "application/json", []byte(`{"order_id": 1}`))
    switch {
    case errors.Is(err, types.ErrConnectionBlocked):
        log.Println("broker is blocked; back off and retry later")
    case errors.Is(err, types.ErrPublishNacked):
        log.Println("broker rejected the publish")
    case err != nil:
        log.Printf("publish failed: %v", err)
    }
}

Apache Kafka

Kafka has at-least-once semantics in v2: a nil handler return commits the offset. An error republishes the message to retry_topic with a retry_count header, and once retry_max_retries is exhausted it lands on dead_letter_topic. Without a retry_topic, the offset is simply left unmarked so the message is redelivered. Handlers must be idempotent: a rebalance can redeliver in-flight messages.

Consumer group with the retry pipeline

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-kafka/v2"
)

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

    // Kafka-only keys go through WithConfig; typed builder values win on overlap.
    native := viper.New()
    native.Set("initial_offset", "earliest")
    native.Set("retry_topic", "user-events-retry")
    native.Set("dead_letter_topic", "user-events-dlt") // required when retry_topic is set

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("kafka://broker1:9092,broker2:9092").
        WithDestination("user-events").  // -> topic
        WithConsumerGroup("user-service"). // -> consumer_group
        WithMaxRetries(3).                 // -> retry_max_retries
        WithAutoReconnect(true).
        WithConfig(native).
        WithHandler(func(ctx context.Context, msg types.Message) error {
            // Exchange() is the topic, RoutingKey() is the message key.
            log.Printf("%s/%s (attempt %d of %d): %s",
                msg.Exchange(), msg.RoutingKey(), msg.RetryCount()+1, msg.MaxRetries(), string(msg.Body()))
            return nil
        }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Producer

package main

import (
    "context"
    "fmt"
    "log"

    "go.digitalxero.dev/mq-kafka/v2"
    "go.digitalxero.dev/mqutils/v2"
)

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

    producer, err := mqutils.NewProducerBuilder().
        WithURL("kafka://localhost:9092").
        WithDestination("user-events"). // default topic for empty-topic publishes
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    for i := 0; i < 100; i++ {
        key := fmt.Sprintf("user-%d", i) // same key -> same partition
        body := []byte(fmt.Sprintf(`{"user_id": %d, "action": "login"}`, i))

        if err := producer.Publish(ctx, fmt.Sprintf("msg-%d", i), "", key, "application/json", body); err != nil {
            log.Printf("Failed to publish message %d: %v", i, err)
        }
    }

    // Headers and other properties go through the message builder.
    msg := kafka.NewMessageBuilder().
        WithBody([]byte(`{"event": "user_signup", "user_id": "12345"}`)).
        WithCorrelationId("corr-456").
        WithContentType("application/json").
        WithHeaders(map[string]any{"event_type": "user_event", "version": "v1"}).
        Build()

    if err := producer.PublishMsg(ctx, "user-events", "user-12345", msg); err != nil {
        log.Fatal(err)
    }
}

NATS Core and JetStream

NATS Core is at-most-once: there is no acknowledgment protocol, so returning an error from a handler only logs the failure. Nothing is redelivered or dead-lettered. Use JetStream when you need redelivery.

NATS Core consumer with a queue group

package main

import (
    "context"
    "log"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-nats/v2" // registers nats://, natss://, tls://, jetstream://
)

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

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("nats://localhost:4222").
        WithDestination("events.>").            // -> subject (wildcards allowed)
        WithConsumerGroup("event-processors"). // -> queue_group (load balancing)
        WithHandler(func(ctx context.Context, msg types.Message) error {
            // RoutingKey() is the subject the message arrived on.
            log.Printf("%s: %s", msg.RoutingKey(), string(msg.Body()))
            return nil
        }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

JetStream durable consumer

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-nats/v2"
)

func main() {
    mqutils.RegisterHandler("order_processor", func(ctx context.Context, msg types.Message) error {
        log.Printf("Processing order: %s", string(msg.Body()))
        // Returning an error NAKs the message; JetStream redelivers it up to
        // max_deliver times and then drops it.
        return nil
    })

    config := viper.New()
    config.Set("url", "jetstream://localhost:4222")
    config.Set("use_jetstream", true)
    config.Set("subject", "orders.created")
    config.Set("stream_name", "orders")
    config.Set("consumer_name", "order-processor")
    config.Set("durable", true)
    config.Set("max_deliver", 3) // "max_retries" also works (canonical key)
    config.Set("ack_wait", 30000) // milliseconds
    config.Set("fetch_batch_size", 10)
    config.Set("handler", "order_processor")

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Publishing to NATS

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"

    _ "go.digitalxero.dev/mq-nats/v2"
)

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

    // JetStream producers declare the stream on Start.
    native := viper.New()
    native.Set("use_jetstream", true)

    producer, err := mqutils.NewProducerBuilder().
        WithURL("jetstream://localhost:4222").
        WithDestination("orders"). // -> stream_name
        WithConfig(native).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // For NATS the "exchange" argument is the subject and "topic" is the
    // reply-to subject (empty for plain publishes).
    err = producer.Publish(ctx, "order-1", "orders.created", "", "application/json", []byte(`{"order_id": 1}`))
    if err != nil {
        log.Fatal(err)
    }
}

AWS SQS

Consumer with a visibility timeout and dead-letter queue

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-aws/v2" // registers sqs:// and sqss://
)

func main() {
    mqutils.RegisterHandler("sqs_processor", func(ctx context.Context, msg types.Message) error {
        // Long-running work must finish inside visibility_timeout, otherwise
        // SQS makes the message visible again and another consumer picks it up.
        return processTask(ctx, msg.Body())
    })

    config := viper.New()
    // sqs://<region>/<queue-name> or sqs://<region>/<account-id>/<queue-name>
    config.Set("url", "sqs://us-east-1/123456789012/my-queue")
    config.Set("visibility_timeout", 300) // seconds
    config.Set("wait_time_seconds", 20)   // long polling
    config.Set("max_messages", 10)
    // Points the queue's RedrivePolicy at the DLQ with maxReceiveCount = max_retries.
    config.Set("dead_letter_queue_url", "https://sqs.us-east-1.amazonaws.com/123456789012/my-dlq")
    config.Set("max_retries", 3)
    config.Set("handler", "sqs_processor")

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

func processTask(ctx context.Context, body []byte) error {
    log.Printf("Processing %d bytes", len(body))
    return nil
}

AWS credentials come from the default credential chain (environment, shared config, instance role). access_key_id, secret_access_key, and session_token override it when set.

FIFO queue producer

For .fifo queues the topic argument is the message group ID (ordering scope). The deduplication ID falls back to the correlation ID unless message_deduplication_id is configured.

package main

import (
    "context"
    "fmt"
    "log"

    "go.digitalxero.dev/mqutils/v2"

    _ "go.digitalxero.dev/mq-aws/v2"
)

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

    producer, err := mqutils.NewProducerBuilder().
        WithURL("sqs://us-east-1/123456789012/orders.fifo").
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    for i := 0; i < 5; i++ {
        err := producer.Publish(ctx,
            fmt.Sprintf("order-%d", i), // correlation ID (doubles as the deduplication ID)
            "",                          // queue URL: empty uses the queue from the connection URL
            "customer-42",               // message group ID (FIFO ordering scope)
            "application/json",
            []byte(fmt.Sprintf(`{"sequence": %d}`, i)),
        )
        if err != nil {
            log.Printf("Failed to publish FIFO message: %v", err)
        }
    }
}

GCP Pub/Sub

Pub/Sub is at-least-once. Returning an error NACKs the message and the subscription’s retry policy redelivers it. A dead_letter_topic with max_delivery_attempts configures a DeadLetterPolicy on the subscription; this requires granting the project’s Pub/Sub service account roles/pubsub.publisher on the dead-letter topic and roles/pubsub.subscriber on the subscription, which the library cannot do for you. Exactly-once delivery is off by default; exactly_once_delivery: true passes the flag through to the subscription.

Consumer with message ordering and a dead-letter topic

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-gcp/v2" // registers pubsub://, pubsubs://, gcp://
)

func main() {
    mqutils.RegisterHandler("pubsub_processor", func(ctx context.Context, msg types.Message) error {
        // Pub/Sub attributes arrive as headers; the ordering key is RoutingKey().
        log.Printf("ordering key %q, attributes %v: %s",
            msg.RoutingKey(), msg.Headers(), string(msg.Body()))
        return nil
    })

    config := viper.New()
    config.Set("url", "pubsub://my-project/my-topic?subscription=my-subscription")
    config.Set("create_subscription_if_not_exists", true)
    config.Set("enable_message_ordering", true)
    config.Set("max_concurrent_handlers", 10)
    config.Set("ack_deadline_seconds", 60)
    config.Set("dead_letter_topic", "my-topic-dlt")
    config.Set("max_delivery_attempts", 5) // "max_retries" also works (canonical key)
    config.Set("handler", "pubsub_processor")

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Ordered publishing

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"

    _ "go.digitalxero.dev/mq-gcp/v2"
)

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

    native := viper.New()
    native.Set("enable_message_ordering", true)

    producer, err := mqutils.NewProducerBuilder().
        WithURL("pubsub://my-project/my-topic").
        WithConfig(native).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // "exchange" is the topic (empty = topic from the URL); "topic" is the ordering key.
    for _, body := range []string{`{"step": 1}`, `{"step": 2}`, `{"step": 3}`} {
        if err := producer.Publish(ctx, "", "", "customer-42", "application/json", []byte(body)); err != nil {
            log.Fatal(err)
        }
    }
}

Redis Pub/Sub and Streams

Redis Pub/Sub is at-most-once with no persistence: a subscriber that is offline misses messages, and a handler error cannot trigger redelivery. Redis Streams with consumer groups are at-least-once and support pending-message recovery.

Redis Pub/Sub with channel patterns

package main

import (
    "context"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-redis/v2" // registers redis://, rediss://, redisstream://, redisstreams://
)

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

    native := viper.New()
    native.Set("use_patterns", true) // PSUBSCRIBE instead of SUBSCRIBE

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("redis://localhost:6379").
        WithDestination("events.*"). // -> channel_name
        WithConfig(native).
        WithHandler(func(ctx context.Context, msg types.Message) error {
            log.Printf("Received: %s", string(msg.Body()))
            return nil
        }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Redis Streams with a consumer group

package main

import (
    "context"
    "log"
    "os"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-redis/v2"
)

func main() {
    mqutils.RegisterHandler("stream_processor", func(ctx context.Context, msg types.Message) error {
        // MessageId() is the stream entry ID.
        log.Printf("Entry %s: %s", msg.MessageId(), string(msg.Body()))
        return nil
    })

    hostname, _ := os.Hostname()

    config := viper.New()
    // The redisstream:// scheme selects Streams mode; the path is the stream key.
    config.Set("url", "redisstream://localhost:6379/orders")
    config.Set("consumer_group", "processors")
    config.Set("consumer_name", hostname) // must be unique per consumer instance
    config.Set("max_retries", 5)
    // Messages left pending by a dead consumer are reclaimed with XAUTOCLAIM
    // (Redis >= 6.2) after claim_idle_time_seconds; entries pending longer
    // than pending_message_max_age_seconds are dropped as poison messages.
    config.Set("claim_idle_time_seconds", 30)
    config.Set("pending_message_max_age_seconds", 300)
    config.Set("handler", "stream_processor")

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Publishing to a stream

package main

import (
    "context"
    "log"

    "go.digitalxero.dev/mqutils/v2"

    _ "go.digitalxero.dev/mq-redis/v2"
)

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

    producer, err := mqutils.NewProducerBuilder().
        WithURL("redisstream://localhost:6379"). // stream scheme -> "stream" mode
        WithDestination("orders").               // -> stream_key
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }
    if err := producer.Start(ctx); err != nil {
        log.Fatal(err)
    }

    // In stream mode the "topic" argument selects the stream key; empty uses
    // the configured destination. Entry IDs are auto-generated (XADD ... *).
    if err := producer.Publish(ctx, "order-1", "", "", "application/json", []byte(`{"order_id": 1}`)); err != nil {
        log.Fatal(err)
    }
}

Batch Processing

Batch processing works on every backend. The runtime collects up to batch_size messages, or whatever has arrived when batch_timeout elapses, and hands the slice to a types.BatchHandlerFunc. A nil return acknowledges every message in the batch; an error rejects every message.

package main

import (
    "context"
    "encoding/json"
    "log"
    "time"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-kafka/v2"
)

type metric struct {
    Name  string  `json:"name"`
    Value float64 `json:"value"`
}

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

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("kafka://localhost:9092").
        WithDestination("metrics").
        WithConsumerGroup("metrics-writer").
        // handler, batch size, flush timeout
        WithBatchHandler(func(ctx context.Context, msgs []types.Message) error {
            metrics := make([]metric, 0, len(msgs))
            for _, msg := range msgs {
                var m metric
                if err := json.Unmarshal(msg.Body(), &m); err != nil {
                    msg.Logger(ctx).Warn("skipping malformed metric")
                    continue
                }
                metrics = append(metrics, m)
            }

            // A failed bulk write returns an error, which rejects the whole
            // batch so it is redelivered together.
            return bulkInsert(ctx, metrics)
        }, 100, 5*time.Second).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

func bulkInsert(ctx context.Context, metrics []metric) error {
    log.Printf("Writing %d metrics", len(metrics))
    return nil
}

The same configuration with viper uses enable_batch_processing, batch_size, batch_timeout (a Go duration string such as "5s"), and a handler registered with mqutils.RegisterBatchHandler:

mqutils.RegisterBatchHandler("metrics_batch", func(ctx context.Context, msgs []types.Message) error {
    log.Printf("Processing batch of %d messages", len(msgs))
    return nil
})

config := viper.New()
config.Set("url", "kafka://localhost:9092")
config.Set("topic", "metrics")
config.Set("consumer_group", "metrics-writer")
config.Set("enable_batch_processing", true)
config.Set("batch_size", 100)
config.Set("batch_timeout", "5s")
config.Set("handler", "metrics_batch")

Partial success in a batch

A nil return acknowledges every message in the batch and an error rejects every message. To settle only some of them, reject those individually with msg.Nack() and return nil; the runtime skips messages the handler already settled.

mqutils.RegisterBatchHandler("bulk", func(ctx context.Context, msgs []types.Message) error {
    for _, msg := range msgs {
        if err := store.Write(ctx, msg.Body()); err != nil {
            _ = msg.Nack() // redelivered and counted against max_retries
        }
    }
    return nil // everything not nacked above is acknowledged
})

TLS

TLS is selected by the URL scheme: amqps://, kafkas:// (or tls_enabled: true), natss:// / tls://, sqss://, rediss://, and redisstreams://, and rabbitmq-stream+tls://. skip_verify only relaxes certificate verification; it never disables TLS.

AMQP with mutual TLS

The AMQP backend reads tls_cert/tls_key (client certificate, both required together), tls_ca (optional; system roots when unset), and sni_hostname (when the TLS server name differs from the URL host). The typed builders expose the same keys.

package main

import (
    "context"
    "log"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-amqp/v2"
)

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

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("amqps://app@rabbitmq.internal:5671/").
        WithDestination("secure-queue").
        WithTLSClientCert("/etc/ssl/client.pem", "/etc/ssl/client-key.pem").
        WithTLSCA("/etc/ssl/ca.pem").
        WithTLSServerName("rabbitmq.internal"). // -> sni_hostname
        WithHandler(func(ctx context.Context, msg types.Message) error {
            log.Printf("Received: %s", string(msg.Body()))
            return nil
        }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

Other backends

yaml

# Kafka: TLS via scheme or flag. SASL is optional; client certificates are not exposed.
url: "kafkas://broker.example.com:9093"
tls_enabled: true
skip_verify: false

# NATS: TLS via scheme; NKey/JWT credentials via a .creds file.
url: "natss://nats.example.com:4222"
credentials_file: "/etc/nats/app.creds"

# Redis: TLS via scheme.
url: "rediss://redis.example.com:6380"

Health Monitoring

Every consumer and producer implements types.HealthChecker. The returned types.HealthCheck exposes Status(), Message(), ConsumerType(), ConnectionURL(), LastChecked(), and a Details() map of backend-specific values.

package main

import (
    "context"
    "log"
    "net/http"
    "time"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-kafka/v2"
)

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

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("kafka://localhost:9092").
        WithDestination("events").
        WithConsumerGroup("health-demo").
        WithHandler(func(ctx context.Context, msg types.Message) error { return nil }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    // Expose the consumer's health as a readiness endpoint.
    http.HandleFunc("/ready", func(w http.ResponseWriter, r *http.Request) {
        checkCtx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
        defer cancel()

        health, err := consumer.HealthCheck(checkCtx)
        if err != nil || health.Status() != types.HealthStatusHealthy {
            http.Error(w, "not ready", http.StatusServiceUnavailable)
            return
        }
        w.WriteHeader(http.StatusOK)
    })
    go func() {
        if err := http.ListenAndServe(":8080", nil); err != nil {
            log.Printf("health server stopped: %v", err)
        }
    }()

    // Or poll it and log the details.
    go func() {
        ticker := time.NewTicker(30 * time.Second)
        defer ticker.Stop()

        for range ticker.C {
            health, err := consumer.HealthCheck(ctx)
            if err != nil {
                log.Printf("Health check failed: %v", err)
                continue
            }
            log.Printf("%s: %s - %s", health.ConsumerType(), health.Status(), health.Message())
            for k, v := range health.Details() {
                log.Printf("  %s=%s", k, v)
            }
        }
    }()

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

The possible statuses are types.HealthStatusHealthy, types.HealthStatusUnhealthy, types.HealthStatusConnecting, and types.HealthStatusClosed.

Graceful Shutdown

Cancelling the context passed to Run stops the consumer. With graceful shutdown enabled, the runtime stops receiving new messages and waits up to the configured timeout for in-flight handlers to finish before returning.

package main

import (
    "context"
    "log"
    "os/signal"
    "syscall"
    "time"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-amqp/v2"
)

func main() {
    // ctx is cancelled on SIGINT/SIGTERM.
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        WithDestination("jobs").
        WithGracefulShutdown(30 * time.Second). // enable_graceful_shutdown + graceful_shutdown_timeout
        WithAutoReconnect(true).
        WithPrefetch(50). // -> message_channel_buffer
        WithHandler(func(ctx context.Context, msg types.Message) error {
            // Long-running work should watch ctx so it can stop promptly
            // when the drain timeout is about to expire.
            select {
            case <-time.After(2 * time.Second):
                return nil
            case <-ctx.Done():
                return ctx.Err() // rejected; the broker redelivers it
            }
        }).
        Build(ctx)
    if err != nil {
        log.Fatal(err)
    }

    log.Println("Consumer running; press Ctrl+C to stop")
    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
    log.Println("Shutdown complete")
}

With viper the equivalent keys are enable_graceful_shutdown: true and graceful_shutdown_timeout: 30 (seconds).

Retry Budgets and Dead-Lettering

max_retries is the canonical retry budget. Once a message exhausts it, the runtime acknowledges it away, first invoking the backend’s dead-letter hook where the broker supports one. What happens on a rejected message depends on the backend:

Backend	On handler error	Dead-letter destination
AMQP	Requeued through a TTL retry queue (`retry_queue_ttl` ms); native delayed retry on RabbitMQ ≥ 4.3 quorum queues	`dead_letter_exchange`
Kafka	Republished to `retry_topic` with a `retry_count` header	`dead_letter_topic`
NATS JetStream	NAK; redelivered up to `max_deliver`	None (dropped)
NATS Core	Nothing (at-most-once)	None
AWS SQS	Message becomes visible again after `visibility_timeout`	`dead_letter_queue_url` via RedrivePolicy
GCP Pub/Sub	NACK; redelivered per the subscription RetryPolicy	`dead_letter_topic` via DeadLetterPolicy (IAM grants required)
Redis Streams	Left pending; reclaimed via XAUTOCLAIM	None (dropped after `max_retries` or `pending_message_max_age_seconds`)
Redis Pub/Sub	Nothing (at-most-once)	None

AMQP retry queue and dead-letter exchange

package main

import (
    "context"
    "errors"
    "log"

    "github.com/spf13/viper"

    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"

    _ "go.digitalxero.dev/mq-amqp/v2"
)

var errTemporary = errors.New("downstream unavailable")

func main() {
    mqutils.RegisterHandler("payment_processor", func(ctx context.Context, msg types.Message) error {
        if err := chargeCard(ctx, msg.Body()); err != nil {
            if errors.Is(err, errTemporary) {
                // Retryable: the error return sends the message through the
                // retry queue and back after retry_queue_ttl.
                return err
            }
            // Permanent failure: log and acknowledge so it is not retried.
            msg.Logger(ctx).Error("permanent payment failure; dropping")
            return nil
        }
        return nil
    })

    config := viper.New()
    config.Set("url", "amqp://guest:guest@localhost:5672/")
    config.Set("queue", "payments")
    config.Set("queue_type", "quorum")        // native delayed retry on RabbitMQ >= 4.3
    config.Set("retry_queue_name", "payments") // in-place retry
    config.Set("retry_queue_ttl", 60000)       // 60s between attempts
    config.Set("max_retries", 5)               // -> retry_queue_max_retries
    config.Set("dead_letter_exchange", "payments-dlx")
    config.Set("auto_declare", true)
    config.Set("handler", "payment_processor")

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

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Consumer stopped: %v", err)
    }
}

func chargeCard(ctx context.Context, body []byte) error {
    return nil
}

On brokers older than RabbitMQ 4.3, or with non-quorum queues, the consumer falls back to a classic <queue>-retry dead-letter/TTL queue. That fallback queue is only created automatically when auto_declare is true.

Request-Response (AMQP)

Consumed AMQP messages carry a Publisher() bound to the consumer’s connection, so a handler can answer on the message’s ReplyTo() queue with the same correlation ID.

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "go.digitalxero.dev/mq-amqp/v2"
    "go.digitalxero.dev/mqutils/v2"
    "go.digitalxero.dev/mqutils/v2/types"
)

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()

    go runResponder(ctx)
    time.Sleep(time.Second) // let the responder subscribe

    if err := sendRequest(ctx); err != nil {
        log.Fatal(err)
    }
}

func runResponder(ctx context.Context) {
    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        WithDestination("requests").
        WithHandler(func(ctx context.Context, msg types.Message) error {
            if msg.ReplyTo() == "" {
                return nil // nothing to answer
            }

            reply := fmt.Sprintf("Processed: %s", string(msg.Body()))

            // Publish to the default exchange with the reply queue as the
            // routing key. Returning the publish error rejects the request.
            return msg.Publisher().Publish(ctx,
                msg.CorrelationId(),
                "",            // default exchange
                msg.ReplyTo(), // reply queue
                "text/plain",
                []byte(reply))
        }).
        Build(ctx)
    if err != nil {
        log.Printf("Responder failed to start: %v", err)
        return
    }

    if err := consumer.Run(ctx); err != nil {
        log.Printf("Responder stopped: %v", err)
    }
}

func sendRequest(ctx context.Context) error {
    replies := make(chan string, 1)

    // Consumer for the reply queue.
    consumer, err := mqutils.NewConsumerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        WithDestination("responses").
        WithHandler(func(ctx context.Context, msg types.Message) error {
            if msg.CorrelationId() == "req-123" {
                replies <- string(msg.Body())
            }
            return nil
        }).
        Build(ctx)
    if err != nil {
        return err
    }
    go func() {
        if err := consumer.Run(ctx); err != nil {
            log.Printf("Requester consumer stopped: %v", err)
        }
    }()

    // Producer for the request itself.
    producer, err := mqutils.NewProducerBuilder().
        WithURL("amqp://guest:guest@localhost:5672/").
        Build(ctx)
    if err != nil {
        return err
    }
    if err := producer.Start(ctx); err != nil {
        return err
    }

    request := amqp.NewMessageBuilder().
        WithCorrelationId("req-123").
        WithReplyTo("responses").
        WithContentType("text/plain").
        WithBody([]byte("Hello, please process this")).
        Build()

    if err := producer.PublishMsg(ctx, "", "requests", request); err != nil {
        return err
    }

    select {
    case reply := <-replies:
        log.Printf("Received reply: %s", reply)
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

Testing Handlers

Handlers are plain functions, so unit tests can call them directly. Any backend’s NewMessageBuilder() produces a types.Message suitable for driving a handler; no broker is required for handlers that use the normal error-returning contract.

package orders_test

import (
    "context"
    "encoding/json"
    "errors"
    "testing"

    "github.com/stretchr/testify/assert"
    "github.com/stretchr/testify/require"

    "go.digitalxero.dev/mq-amqp/v2"
    "go.digitalxero.dev/mqutils/v2/types"
)

var errEmptyOrder = errors.New("order id is required")

// processOrder is the handler under test.
func processOrder(ctx context.Context, msg types.Message) error {
    var order struct {
        ID string `json:"id"`
    }
    if err := json.Unmarshal(msg.Body(), &order); err != nil {
        return err
    }
    if order.ID == "" {
        return errEmptyOrder
    }
    return nil
}

func TestProcessOrder(t *testing.T) {
    ctx := context.Background()

    valid := amqp.NewMessageBuilder().
        WithContentType("application/json").
        WithBody([]byte(`{"id": "order-1"}`)).
        Build()
    require.NoError(t, processOrder(ctx, valid), "valid orders are acknowledged")

    empty := amqp.NewMessageBuilder().
        WithContentType("application/json").
        WithBody([]byte(`{}`)).
        Build()
    err := processOrder(ctx, empty)
    require.Error(t, err, "an error return rejects the message")
    assert.ErrorIs(t, err, errEmptyOrder)

    malformed := amqp.NewMessageBuilder().
        WithBody([]byte(`not json`)).
        Build()
    assert.Error(t, processOrder(ctx, malformed))
}

func TestHandlerRegistry(t *testing.T) {
    types.RegisterHandler("orders_test", processOrder)

    registered := types.GetHandler("orders_test")
    require.NotNil(t, registered)
    assert.Nil(t, types.GetHandler("missing"))
}

For end-to-end tests against real brokers, the repository’s docker-compose.test.yml starts every supported broker, and the acceptance/ suite exercises the public API against them.

Broker capability examples

Kafka SASL example: authenticated publication using environment-provided credentials. The Kafka guide lists supported mechanisms and TLS requirements.
Kafka transaction example: commit or abort order/audit records together, including scoped Close and uncertain outcomes.
RabbitMQ Stream outbox/inbox example: persist immutable publication identities, replay across restart, and couple database effects with an inbox transaction. Its -super option uses a partitioned super stream.

These are standalone Go modules with executable programs and verification instructions. The Stream guide also shows typed single-stream, super-stream routing, filtering, and Redis application-deduplication composition. It records the pending new-module vanity mappings and local replacement workflow.