Requeue All
Move all received KubeMQ queue messages to a different queue with reQueueAllMessages in Kotlin.
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 response returned by a manual poll: after receiving messages with autoAck = false, call client.reQueueAllMessages(response, targetChannel) to move every message in that response to the target channel in one call, 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 from that poll — anything already acked or expired beforehand is gone before reQueueAllMessages runs.
Prerequisites
- KubeMQ server running on
localhost:50000 - Kotlin SDK installed (
implementation("io.kubemq.sdk:kubemq-sdk-kotlin:1.0.1"))
Code
package io.kubemq.sdk.examples.queuesstream
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.runBlocking
private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-queues-requeue-all"
private const val CHANNEL = "kotlin-queues.requeue-all"
private const val REQUEUE_CHANNEL = "kotlin-queues.requeue-all-dest"
fun main() = runBlocking {
val client = KubeMQClient.queues {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
try {
client.createQueuesChannel(CHANNEL)
client.createQueuesChannel(REQUEUE_CHANNEL)
// Send messages to source queue
repeat(3) { i ->
client.sendQueuesMessage(queueMessage {
channel = CHANNEL
body = "Requeue msg ${i + 1}".toByteArray()
})
}
// Poll from source queue
val response = client.receiveQueuesMessages {
channel = CHANNEL
maxItems = 10
waitTimeoutMs = 5000
autoAck = false
}
println("Received ${response.messages.size} messages from source.")
// Requeue all to destination channel
client.reQueueAllMessages(response, REQUEUE_CHANNEL)
println("All messages requeued to: $REQUEUE_CHANNEL")
// Verify messages arrived in destination
val destResp = client.receiveQueuesMessages {
channel = REQUEUE_CHANNEL
maxItems = 10
waitTimeoutMs = 2000
autoAck = true
}
println("Destination queue received: ${destResp.messages.size} messages.")
} finally {
try { client.deleteQueuesChannel(CHANNEL) } catch (_: Exception) {}
try { client.deleteQueuesChannel(REQUEUE_CHANNEL) } catch (_: Exception) {}
}
println("Done.")
}
}How It Works
reQueueAllMessages(response, targetChannel)moves all messages from the poll response to a different queue.- Messages are removed from the source queue and placed in the target queue atomically.
- Useful for routing messages to different processing pipelines or priority queues.
Related
Was this page helpful?