Delayed Messages
Send a KubeMQ queue message with a delivery delay via QueueMessagePolicy using the Kotlin SDK.
Overview
A delivery delay holds a queue message out of consumers' reach for a fixed window after it's sent — the message is accepted and persisted immediately, but invisible to pollers until the delay expires. It's the building block for scheduled work — a reminder to fire in an hour, a retry with back-off, a task queued for off-peak processing — without standing up a separate scheduler or cron service.
Set it with QueueMessagePolicy(delaySeconds = ...) on the message before sending; the broker does the waiting. A poll via receiveQueuesMessages against the channel before the delay elapses simply returns zero messages — it isn't hidden in a separate place, it's the same queue, just not yet eligible for delivery. Once the delay window passes, the next poll retrieves it normally.
Gotchas: the delay is set once at send time, per message, and can't be extended or shortened afterward — if you need a different wait, send a new message. A long delay still counts as an in-flight, persisted message, so it survives a broker restart, but it also occupies queue storage for the whole waiting period. Don't confuse this with a visibility timeout after delivery — that's a separate mechanism for redelivery on failed acknowledgment, not initial availability.
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.queues
import io.kubemq.sdk.client.KubeMQClient
import io.kubemq.sdk.queues.QueueMessagePolicy
import io.kubemq.sdk.queues.queueMessage
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
private const val ADDRESS = "localhost:50000"
private const val CLIENT_ID = "kotlin-queues-delayed-messages"
private const val CHANNEL = "kotlin-queues.delayed-messages"
fun main() = runBlocking {
val client = KubeMQClient.queues {
address = ADDRESS
clientId = CLIENT_ID
}
client.use {
// Send a message with 3-second delay
println("Sending message with 3s delay...")
client.sendQueuesMessage(queueMessage {
channel = CHANNEL
body = "Delayed 3s".toByteArray()
policy = QueueMessagePolicy(delaySeconds = 3)
})
// Poll immediately -- expect no message (delay not expired)
println("Polling immediately (before delay expires)...")
val earlyResp = client.receiveQueuesMessages {
channel = CHANNEL
maxItems = 1
waitTimeoutMs = 1000
autoAck = true
}
println("Messages before delay: ${earlyResp.messages.size} (expected 0)")
// Wait for delay to expire
println("Waiting for delay to expire...")
delay(3500)
// Poll after delay -- message should be available
val lateResp = client.receiveQueuesMessages {
channel = CHANNEL
maxItems = 1
waitTimeoutMs = 5000
autoAck = true
}
println("Messages after delay: ${lateResp.messages.size} (expected 1)")
lateResp.messages.forEach { println(" Received: ${String(it.body)}") }
println("Done.")
}
}How It Works
QueueMessagePolicy(delaySeconds = 3)makes the message invisible for 3 seconds after sending.- Polling before the delay expires returns no messages.
- After the delay, the message becomes available for normal consumption.
- Useful for scheduled tasks, retry delays, or deferred processing.
Related
Was this page helpful?