Docs / Why mqutils

Why mqutils

What one interface across messaging backends buys you, what it costs, and what it deliberately does not do

The problem it solves

Every broker SDK has its own connection lifecycle, its own way to acknowledge or reject a message, its own retry story, and its own configuration shape. A service that runs against RabbitMQ in one deployment and SQS in another ends up with two consumer implementations, or with an in-house abstraction that someone has to maintain.

mqutils is that abstraction, maintained once. It puts a single consumer and producer contract in front of RabbitMQ AMQP and Streams, Kafka, NATS Core and JetStream, AWS SQS, GCP Pub/Sub, and Redis Pub/Sub and Streams. The URL scheme selects the backend; each backend is its own Go module, so you compile only the ones you use.

What you get

One handler contract. A handler is func(ctx context.Context, msg types.Message) error. Return nil to acknowledge, return an error to reject. There is no manual ack or nack to forget.

types.RegisterHandler("orders", func(ctx context.Context, msg types.Message) error {
    return orders.Handle(ctx, msg.Body())
})

One consumer runtime. The runner package owns everything that is identical across brokers: settling messages based on the handler’s return value (with retried acks), retry budgets with a dead-letter hook, batch collection with a real flush timer, reconnect with exponential backoff, and graceful draining of in-flight work on shutdown. Backends supply only the broker-specific pieces.

One configuration vocabulary. destination, consumer_group and max_retries are accepted by every backend alongside its native keys. Configure with viper, or with the typed builders if you would rather not depend on viper:

consumer, err := mqutils.NewConsumerBuilder().
    WithURL("kafka://localhost:9092").
    WithDestination("orders").
    WithConsumerGroup("billing").
    WithMaxRetries(3).
    WithHandler(handleOrder).
    Build(ctx)

Batching on every backend. Set enable_batch_processing, batch_size and batch_timeout (a duration string such as "250ms") and register a batch handler. A nil return settles the whole batch.

Health checks. Consumers and producers implement types.HealthChecker, with transport-specific diagnostics behind one interface.

TLS and broker authentication. Secure schemes include amqps://, rabbitmq-stream+tls://, kafkas://, natss://, and rediss://. Certificate configuration varies by backend. Kafka supports SASL PLAIN and SCRAM.

What it does not do

This matters more than the feature list.

It does not make delivery guarantees uniform. The guarantee is the broker’s. AMQP, RabbitMQ Streams, Kafka, JetStream, SQS, Pub/Sub and Redis Streams are at-least-once. NATS Core and Redis Pub/Sub are at-most-once: they cannot redeliver, so max_retries and dead-lettering do not apply to them. Where the broker has a dead-letter mechanism (a dead-letter exchange, a dead-letter topic, a RedrivePolicy, a DeadLetterPolicy) mqutils uses it. Where it does not (JetStream, Redis Streams), messages that exhaust the budget are dropped and logged. The backend table on the homepage spells this out per broker.

It does not fail over between brokers. A consumer talks to one broker. Reconnect with backoff is built in; switching to a different backend is a configuration change and a restart.

It does not hide every broker-specific key. The canonical keys cover destination, group and retry budget. Exchanges and routing keys are still an AMQP concept, subjects and streams are still a NATS concept, and you set them with the backend’s native keys. Your handler does not change; the config block does.

It does not make external effects exactly once. Kafka producer transactions and Stream deduplication have explicit scopes. A transactional inbox is needed when an application database effect must be atomic with its completion record.

It does not publish throughput numbers. The overhead is a thin layer over each broker’s client. The benchmarks/ module in the repository measures that overhead on your own hardware against your own broker, which is the only number that matters.

Migrating from a raw SDK

Before, with the RabbitMQ client directly:

conn, err := amqp.Dial(url)
ch, err := conn.Channel()
q, err := ch.QueueDeclare("orders", true, false, false, false, nil)
msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
for d := range msgs {
    if err := handle(d.Body); err != nil {
        _ = d.Nack(false, true)
        continue
    }
    _ = d.Ack(false)
}

After:

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

consumer, err := mqutils.NewConsumerBuilder().
    WithURL(url).
    WithDestination("orders").
    WithMaxRetries(5).
    WithHandler(func(ctx context.Context, msg types.Message) error {
        return handle(msg.Body())
    }).
    Build(ctx)
if err != nil {
    return err
}
return consumer.Run(ctx)

Point the same code at kafka://, jetstream:// or sqs://, import the matching module, and the handler is untouched.

When to use something else

Use the backend capability APIs for Kafka producer transactions and RabbitMQ Streams, including routing, filtering, and durable publication replay. Features outside those APIs, such as Pub/Sub schema administration or transactions that commit Kafka consumer offsets, still require the native SDK and application-level coordination. The shared Transport interface does not promise every feature a broker supports.

Getting started walks through the first consumer and producer.
Examples has a working program for each backend.
Core types is the interface reference.