Overview

The Queues pattern (PatternQueues) maps Watermill's message.Publisher and message.Subscriber onto KubeMQ's reliable, point-to-point queues. Unlike Events (fire-and-forget) and EventsStore (persistent fan-out), Queues give you at-least-once delivery with explicit acknowledgment: every message is persisted on the broker until a consumer acknowledges it, and a failed delivery is redelivered rather than lost. This is the pattern to reach for when you are building task queues and job processing, where each unit of work must be handled exactly once and reliably.

For what the Queues pattern guarantees at the broker level, see Queues. This page documents how the plugin exposes it through the Watermill interfaces — the queue-specific config fields, ack/nack bridging, and native DLQ routing.

Property	Queues
Delivery	At-least-once
Ack/Nack	Explicit (`msg.Ack()` / `msg.Nack()`)
Persistence	Yes (until acked)
Best for	Task queues, job processing, reliable delivery

A publisher writes messages to a queue channel; a subscriber polls the channel, processes each message, and settles it with an ack or a nack. Because the broker holds the message until it is acked, a consumer can crash mid-processing and the message will be redelivered.

The Watermill plugin is a native gRPC client — it dials the broker's gRPC port (50000) directly, with no HTTP connector or enable flag to configure. If you have not set up a broker yet, see Getting Started.

Publisher and Subscriber Configuration

Both the publisher and subscriber are created with Pattern: kubemq.PatternQueues. The subscriber has two queue-specific fields that control polling:

Prop

Type

MaxItems accepts a value in the range 1–1000, and WaitTimeoutSeconds is the server-side poll timeout. For the Queues pattern, both fields are default-corrected to 1 whenever they are set to a value less than or equal to zero, so an unset subscriber still polls one message at a time with a one-second wait:

config.go (Validate)

if c.Pattern == PatternQueues {
    if c.MaxItems <= 0 {
        c.MaxItems = 1
    }
    if c.WaitTimeoutSeconds <= 0 {
        c.WaitTimeoutSeconds = 1
    }
}

A minimal publish-and-receive setup looks like this — note that queues are persistent, so the publisher can write before any subscriber exists:

queues/send-receive/main.go

package main

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

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"

	kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

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

	// Create queue publisher
	pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
		Address: "localhost:50000",
		Pattern: kubemq.PatternQueues,
		Logger:  logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer pub.Close()

	// Create queue subscriber
	// MaxItems: how many messages to fetch per poll cycle
	// WaitTimeoutSeconds: how long to wait for messages before returning empty
	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	// Publish 3 queue messages (queues are persistent -- no need to subscribe first)
	for i := 1; i <= 3; i++ {
		msg := message.NewMessage(watermill.NewUUID(), []byte(fmt.Sprintf("Queue message %d", i)))
		if err := pub.Publish("watermill-queues.send-receive", msg); err != nil {
			log.Fatal(err)
		}
		fmt.Printf("Published: %s\n", string(msg.Payload))
	}

	// Subscribe and receive messages
	msgs, err := sub.Subscribe(ctx, "watermill-queues.send-receive")
	if err != nil {
		log.Fatal(err)
	}

	received := 0
	for received < 3 {
		select {
		case msg := <-msgs:
			fmt.Printf("Received: UUID=%s, Payload=%s\n", msg.UUID, string(msg.Payload))
			// Queue messages MUST be acknowledged to remove them from the queue.
			msg.Ack()
			received++
		case <-ctx.Done():
			log.Fatal("Timeout waiting for messages")
		}
	}

	fmt.Println("Done! All 3 queue messages received and acknowledged.")
}

A larger MaxItems reduces the number of round trips when the queue is busy: the subscriber fetches a batch per poll and delivers the messages one at a time on its output channel. See Batch send below for the matching publisher side.

Ack and Nack Semantics

Queue messages are not removed from the broker on delivery — they are removed only when acknowledged. Each delivered Watermill message carries the two standard settlement calls:

msg.Ack() — acknowledges the message, removing it from the queue.
msg.Nack() — negative-acknowledges, returning the message to the queue for redelivery.

Internally, the subscriber bridges Watermill's settlement signals to the KubeMQ queue. For each delivered message it starts a small goroutine that waits on the message's Acked() or Nacked() channel and forwards the result to the underlying KubeMQ queue message. Critically, if the subscription context is cancelled while a message is still in flight, the bridge nacks it so the broker can redeliver it to another consumer rather than dropping it:

subscriber.go (ack/nack bridge)

// Bridge goroutine: Watermill ack/nack -> KubeMQ ack/nack
s.wg.Add(1)
go func(qm *kubemqSDK.QueueDownstreamMessage, wm *message.Message) {
	defer s.wg.Done()
	select {
	case <-wm.Acked():
		if err := qm.Ack(); err != nil {
			s.config.Logger.Error("Queue ack error", err, watermill.LogFields{
				"topic": topic, "uuid": wm.UUID,
			})
		}
	case <-wm.Nacked():
		if err := qm.Nack(); err != nil {
			s.config.Logger.Error("Queue nack error", err, watermill.LogFields{
				"topic": topic, "uuid": wm.UUID,
			})
		}
	case <-subCtx.Done():
		_ = qm.Nack()
	}
}(qMsg, wmMsg)

The following example publishes one message, nacks it on first receive to simulate a processing failure (triggering redelivery), then acks it on the second receive:

queues/ack-nack/main.go

package main

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

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"

	kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

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

	pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
		Address: "localhost:50000",
		Pattern: kubemq.PatternQueues,
		Logger:  logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer pub.Close()

	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	// Publish 1 message
	msg := message.NewMessage(watermill.NewUUID(), []byte("Important task"))
	if err := pub.Publish("watermill-queues.ack-nack", msg); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Published: %s\n", string(msg.Payload))

	msgs, err := sub.Subscribe(ctx, "watermill-queues.ack-nack")
	if err != nil {
		log.Fatal(err)
	}

	// First receive: Nack the message (simulates processing failure)
	select {
	case received := <-msgs:
		fmt.Printf("First receive: %s -- Nacking (simulating failure)\n", string(received.Payload))
		received.Nack() // Message goes back to the queue for redelivery
	case <-ctx.Done():
		log.Fatal("Timeout waiting for first message")
	}

	// Brief pause for redelivery
	time.Sleep(2 * time.Second)

	// Second receive: Ack the message (successful processing)
	select {
	case received := <-msgs:
		fmt.Printf("Second receive: %s -- Acking (processing success)\n", string(received.Payload))
		received.Ack() // Message removed from queue
	case <-ctx.Done():
		log.Fatal("Timeout waiting for redelivered message")
	}

	fmt.Println("Done! Message was nacked, redelivered, and then acked.")
}

Queue Message Policy

A publisher can attach a QueueMessagePolicy that controls delivery behavior for every message it sends. The policy is set once on the PublisherConfig and applies to all messages from that publisher.

Prop

Type

These four fields cover the three reliability features most task queues need: time-to-live (ExpirationSeconds), delayed/scheduled visibility (DelaySeconds), and automatic dead-lettering (MaxReceiveCount + MaxReceiveQueue). The sections below show each in isolation.

Dead Letter Queue

When a message is repeatedly redelivered without ever being acked, you usually do not want it to cycle forever. Set MaxReceiveCount and MaxReceiveQueue on the publisher's policy and KubeMQ will automatically route a message that exceeds the receive count to the named DLQ channel — no extra code on the consumer side.

In the example below the publisher routes anything received more than three times to watermill-queues.dlq. The consumer nacks the message three times, after which the broker moves it to the DLQ, where a second subscriber picks it up:

queues/dead-letter-queue/main.go

package main

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

	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"

	kubemq "github.com/kubemq-io/watermill-kubemq/pkg/kubemq"
)

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

	// Create publisher with DLQ policy
	// MaxReceiveCount: message moves to DLQ after 3 failed receive attempts
	// MaxReceiveQueue: the DLQ channel name
	pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
		Address: "localhost:50000",
		Pattern: kubemq.PatternQueues,
		QueueMessagePolicy: &kubemq.QueueMessagePolicy{
			MaxReceiveCount: 3,
			MaxReceiveQueue: "watermill-queues.dlq",
		},
		Logger: logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer pub.Close()

	// Publish 1 message
	msg := message.NewMessage(watermill.NewUUID(), []byte("Problematic task"))
	if err := pub.Publish("watermill-queues.dlq-source", msg); err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Published: %s\n", string(msg.Payload))

	// Create subscriber for the source queue
	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	msgs, err := sub.Subscribe(ctx, "watermill-queues.dlq-source")
	if err != nil {
		log.Fatal(err)
	}

	// Nack the message 3 times to trigger DLQ
	for attempt := 1; attempt <= 3; attempt++ {
		select {
		case received := <-msgs:
			fmt.Printf("Attempt %d: Nacking message: %s\n", attempt, string(received.Payload))
			received.Nack()
		case <-ctx.Done():
			log.Fatal("Timeout waiting for message")
		}
		time.Sleep(time.Second) // Brief pause between attempts
	}

	time.Sleep(2 * time.Second) // Allow DLQ transfer

	// Subscribe to the DLQ to confirm the message was moved there
	dlqSub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer dlqSub.Close()

	dlqMsgs, err := dlqSub.Subscribe(ctx, "watermill-queues.dlq")
	if err != nil {
		log.Fatal(err)
	}

	select {
	case dlqMsg := <-dlqMsgs:
		fmt.Printf("DLQ received: %s\n", string(dlqMsg.Payload))
		dlqMsg.Ack()
	case <-ctx.Done():
		log.Fatal("Timeout waiting for DLQ message")
	}

	fmt.Println("Done! Message moved to dead letter queue after 3 failed attempts.")
}

Native DLQ vs. Poison Queue Middleware

KubeMQ's native DLQ is not the only way to dead-letter a message. Watermill ships a PoisonQueue middleware that does something similar at a different layer. The two solve different problems and can be combined.

Mechanism	Layer	Catches	Configured by
Native DLQ (`QueueMessagePolicy`)	Transport	Consumer crashes, repeatedly unacked messages	Publisher `MaxReceiveCount` + `MaxReceiveQueue`
Poison Queue middleware	Application	Handler panics, business-logic errors	`middleware.PoisonQueue(pub, topic)` on the Router

The native DLQ is driven by the broker counting receive attempts, so it captures transport-level failures even when the consumer never gets a chance to return an error (for example, it crashes mid-processing). The Poison Queue middleware wraps your handler and catches the error it returns, publishing the failed message to a dedicated topic — so it captures application-level failures that the broker would otherwise see as a normal nack.

middleware/poison-queue/main.go (excerpt)

// PoisonQueue middleware: messages that fail processing are sent to the poison topic.
// It wraps the handler and catches errors, publishing the failed message to the
// specified topic via the provided publisher.
poisonQueue, err := middleware.PoisonQueue(poisonPub, "watermill-mw.poison")
if err != nil {
	log.Fatal(err)
}
router.AddMiddleware(poisonQueue)

// Handler that always fails -- the message will end up in the poison queue
router.AddNoPublisherHandler(
	"always-failing",
	"watermill-mw.poison-input",
	sub,
	func(msg *message.Message) error {
		fmt.Printf("Handler received (will fail): %s\n", string(msg.Payload))
		return fmt.Errorf("processing failed: invalid data format")
	},
)

Use the native DLQ to guard against infrastructure failures and the Poison Queue middleware to isolate messages your handler cannot process. They operate independently — running both gives you coverage at the transport and application layers at once.

Delayed and Expiring Messages

The same QueueMessagePolicy controls message timing.

Delayed visibility (DelaySeconds): the message is published immediately but the broker withholds it from consumers until the delay elapses — useful for scheduled or deferred work.

queues/delayed-messages/main.go (excerpt)

// Create publisher with a 5-second delay policy
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
	Address: "localhost:50000",
	Pattern: kubemq.PatternQueues,
	QueueMessagePolicy: &kubemq.QueueMessagePolicy{
		DelaySeconds: 5,
	},
	Logger: logger,
})

Expiration / TTL (ExpirationSeconds): if no consumer reads the message within the window, it expires and is never delivered — useful for time-sensitive work that is worthless if stale.

queues/expiration/main.go (excerpt)

// Create publisher with a 3-second expiration policy
pub, err := kubemq.NewPublisher(kubemq.PublisherConfig{
	Address: "localhost:50000",
	Pattern: kubemq.PatternQueues,
	QueueMessagePolicy: &kubemq.QueueMessagePolicy{
		ExpirationSeconds: 3,
	},
	Logger: logger,
})

Run the full programs to see the timing in action:

go run ./examples/queues/delayed-messages/main.go
go run ./examples/queues/expiration/main.go

Competing Consumers

To process a queue in parallel, run multiple subscribers that share the same ConsumerGroup. KubeMQ distributes the queue's messages across the group members, so each message is handled by exactly one consumer — the standard competing-consumers pattern for scaling out workers.

queues/competing-consumers/main.go (excerpt)

// Create 3 subscribers with the same consumer group
channels := make([]<-chan *message.Message, 3)
for i := 0; i < 3; i++ {
	sub, err := kubemq.NewSubscriber(kubemq.SubscriberConfig{
		Address:            "localhost:50000",
		Pattern:            kubemq.PatternQueues,
		ConsumerGroup:      "queue-workers",
		MaxItems:           1,
		WaitTimeoutSeconds: 5,
		Logger:             logger,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Close()

	msgs, err := sub.Subscribe(ctx, "watermill-queues.competing-consumers")
	if err != nil {
		log.Fatal(err)
	}
	channels[i] = msgs
}

The queues/competing-consumers example publishes nine tasks and watches them spread across three subscribers in the queue-workers group. For a Router-based take on the same idea — where each worker is a Router handler rather than a raw channel read — see advanced/competing-consumers.

go run ./examples/queues/competing-consumers/main.go
go run ./examples/advanced/competing-consumers/main.go

Batch Send

Publish is variadic, so you can hand it several messages in a single call. On the Queues pattern these are sent upstream to the broker as one batch rather than one request per message, which cuts round trips when you are enqueuing many tasks at once:

queues/batch-send/main.go (excerpt)

// Build 10 messages
msgs := make([]*message.Message, 10)
for i := 0; i < 10; i++ {
	msgs[i] = message.NewMessage(watermill.NewUUID(), []byte(fmt.Sprintf("Batch message %d", i+1)))
}

// Publish all 10 messages in one call.
// Internally this sends them as a single batch via queueUpstream.Send(requestID, queueMsgs).
if err := pub.Publish("watermill-queues.batch-send", msgs...); err != nil {
	log.Fatal(err)
}
fmt.Printf("Published batch of %d messages\n", len(msgs))

Pair a batch publisher with a subscriber that sets a higher MaxItems (the batch-send example uses MaxItems: 5) to fetch several messages per poll on the consumer side as well.

Using the Router with Queues

The Watermill Router works the same with Queues as it does with Events, with one important behavioral difference: because Queues support ack/nack, the Router acks a message only after the handler returns successfully and nacks it on error. That turns the Router into a reliable processing loop — a handler that fails will have its message redelivered.

The queues/router-handler example wires a handler that reads from a pending queue, transforms the payload, and forwards the result to a completed queue:

queues/router-handler/main.go (excerpt)

router.AddMiddleware(middleware.Recoverer)

// Add handler: process pending tasks and publish to the completed queue
router.AddHandler(
	"task-processor",
	"watermill-queues.pending",
	sub,
	"watermill-queues.completed",
	pub,
	func(msg *message.Message) ([]*message.Message, error) {
		result := fmt.Sprintf("DONE: %s", strings.ToUpper(string(msg.Payload)))
		fmt.Printf("Handler processed: %s -> %s\n", string(msg.Payload), result)
		outMsg := message.NewMessage(watermill.NewUUID(), []byte(result))
		return []*message.Message{outMsg}, nil
	},
)

go run ./examples/queues/router-handler/main.go

A Note on the Retry Middleware

Watermill's Retry middleware re-invokes a handler when it returns an error, but it only does anything useful when a failed message can actually be redelivered. Because Queues are the only pattern where a nack triggers redelivery, Retry is meaningful for Queues and is not applicable to Events or EventsStore:

Middleware	Events	EventsStore	Queues	Notes
Retry	N/A	N/A	Yes	Nack triggers redelivery; only meaningful for Queues
Poison Queue	N/A	N/A	Yes	Application-level DLQ (alternative to KubeMQ native DLQ)

Pair Retry with the Poison Queue middleware: retry handles transient failures in place, and anything that exhausts its retries is routed to the poison topic instead of blocking the consumer.

Queues with Ack/Nack & DLQ

Overview

Publisher and Subscriber Configuration

Ack and Nack Semantics

Queue Message Policy

Dead Letter Queue

Native DLQ vs. Poison Queue Middleware

Delayed and Expiring Messages

Competing Consumers

Batch Send

Using the Router with Queues

A Note on the Retry Middleware

Next Steps

Concepts

Getting Started

Reference

On this page