Requeue All
Return all received KubeMQ queue messages to the queue for redelivery using the Go SDK.
Overview
Requeue all moves an entire batch of polled messages to a different channel in one server-side operation, without republishing them from the client. Reach for it when you need to make a routing decision after looking at a batch — shovel a stuck batch into a review queue, redirect it to a priority pipeline, or migrate messages off a channel that's being retired, all while the source queue is cleared atomically.
It works against the batch returned by a manual poll: after receiving messages with AutoAck: false, call resp.ReQueueAll(dstChannel) to move every message held in that transaction to the destination channel, removing them from the source at the same instant. The messages keep their original body, tags, and policies — the broker relocates them, it doesn't recreate them.
Gotchas: requeuing is all-or-nothing for the batch — there's no per-message filter, so split the batch yourself first if only some messages should move. The destination channel is an ordinary queue with no special semantics; nothing consumes it automatically. And the operation only affects messages still held in the open transaction — anything already acked or expired out of the transaction is gone before ReQueueAll runs.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: queues-stream/requeue-all
//
// Demonstrates moving all messages from one queue to another using ReQueueAll.
// This is useful for routing messages to different processing pipelines.
//
// Channel: go-queues-stream.requeue-all
// Client ID: go-queues-stream-requeue-all-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/kubemq-io/kubemq-go/v2"
)
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
client, err := kubemq.NewClient(ctx,
kubemq.WithAddress("localhost", 50000),
kubemq.WithClientId("go-queues-stream-requeue-all-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
srcChannel := "go-queues-stream.requeue-all"
dstChannel := "go-queues-stream.requeue-all.dest"
// Send a message to the source queue with a receive policy.
_, err = client.SendQueueMessage(ctx, kubemq.NewQueueMessage().
SetChannel(srcChannel).
SetBody([]byte("will be requeued")).
SetMaxReceiveCount(3))
if err != nil {
log.Fatal(err)
}
fmt.Println("Message sent to source queue")
// Receive from source queue via downstream receiver.
receiver, err := client.NewQueueDownstreamReceiver(ctx)
if err != nil {
log.Fatal(err)
}
defer receiver.Close()
resp, err := receiver.Poll(ctx, &kubemq.PollRequest{
Channel: srcChannel,
MaxItems: 10,
WaitTimeoutSeconds: 5,
AutoAck: false,
})
if err != nil {
log.Fatal(err)
}
for _, dm := range resp.Messages {
if dm.Message != nil {
fmt.Printf("Received: body=%s tx=%s\n", dm.Message.Body, dm.TransactionID)
}
}
// ReQueueAll: move all messages to the destination queue.
if len(resp.Messages) > 0 {
fmt.Printf("ReQueueAll: moving messages to %s\n", dstChannel)
if err := resp.ReQueueAll(dstChannel); err != nil {
log.Printf("ReQueueAll: %v", err)
}
fmt.Println("Messages requeued to destination")
}
}
How It Works
receiver.PollonsrcChannelwithAutoAck: falseholds messages in an open transaction without removing them from the source queue.resp.ReQueueAll(dstChannel)atomically moves all messages held in the transaction todstChannelon the broker, removing them fromsrcChannelat the same time.- The destination channel (
go-queues-stream.requeue-all.dest) is a regular queue; consumers polling it will find the requeued messages as if they had been sent there directly. - This pattern is useful for routing messages to a different processing pipeline (e.g. priority queue, dead-letter review queue) without republishing.
Related
Was this page helpful?