Delay Policy
Configure a message delay policy on a KubeMQ Queue via the upstream stream API in the C++ SDK to defer delivery.
Overview
A delay policy defers when a queued message becomes visible to consumers — you send it now, but nothing can receive it until a countdown you set expires. That's the mechanism behind retry-after-backoff, rate-limited notifications, "remind me in an hour" workflows, and staggering a burst of work so it doesn't hit downstream consumers all at once, all without standing up a separate scheduler.
It works entirely at send time: setting the delay on the message before it's handed to the upstream stream's send call attaches the countdown to that message alone. The broker starts the countdown the moment it accepts the message and simply excludes it from delivery until the timer elapses — after that it behaves like any other queued message, available to whichever consumer polls next.
Gotchas: the delay is a floor, not a guarantee — the message becomes eligible when the timer expires, but actual delivery still waits for a consumer to poll, so don't rely on it for precise scheduling. It's one-shot: there's no recurrence or cron-like behavior, so long or repeating delays need application logic on top. And it's independent of redelivery — a delayed message that's later nacked or times out after delivery follows normal visibility-timeout/retry rules, not the original send-time delay.
Prerequisites
- KubeMQ server running on
localhost:50000 - C++ SDK installed (vcpkg or CMake FetchContent)
- C++17 compiler (GCC 9+, Clang 9+, MSVC 2019+)
Code
// Example: queues_stream/delay_policy
//
// Demonstrates sending queue messages with a delay policy via stream.
// Delayed messages are not available for consumption until the delay expires.
//
// Channel: cpp-queues.delay-policy
// Client ID: cpp-queues-delay-policy-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
#include <kubemq/kubemq.h>
#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>
#include <vector>
int main() {
std::cout << "[1] Connecting to localhost:50000" << std::endl;
kubemq::ClientOptions options;
options.set_address("localhost", 50000);
options.set_client_id("cpp-queues-delay-policy-client");
auto client_result = kubemq::Client::Create(options);
if (!client_result.ok()) {
std::cerr << "[ERROR] Failed to create client: " << client_result.status().message()
<< std::endl;
return 1;
}
auto& client = *client_result;
std::string channel = "cpp-queues.delay-policy";
std::atomic<bool> result_received{false};
// Open an upstream stream for sending delayed messages.
std::cout << "[2] Opening upstream stream" << std::endl;
auto upstream_result = client->QueueUpstream(
[&result_received](const kubemq::QueueUpstreamResult& res) {
if (res.is_error) {
std::cerr << "[ERROR] Upstream result: " << res.error << std::endl;
} else {
std::cout << "[4] Sent message with 10s delay policy" << std::endl;
}
result_received.store(true);
},
[](const kubemq::Status& err) {
std::cerr << "[ERROR] Upstream error: " << err.message() << std::endl;
});
if (!upstream_result.ok()) {
std::cerr << "[ERROR] QueueUpstream: " << upstream_result.status().message() << std::endl;
return 1;
}
auto& upstream = *upstream_result;
// Build a message with a 10-second delay.
auto msg_result = kubemq::QueueMessage::Builder()
.SetChannel(channel)
.SetBody("delayed by 10s")
.SetDelaySeconds(10)
.Build();
if (!msg_result.ok()) {
std::cerr << "[ERROR] Build message: " << msg_result.status().message() << std::endl;
return 1;
}
std::vector<kubemq::QueueMessage> messages;
messages.push_back(std::move(*msg_result));
std::cout << "[3] Sending message with delay_seconds=10" << std::endl;
auto send_status = upstream->Send("req-delay", messages);
if (!send_status.ok()) {
std::cerr << "[ERROR] Send: " << send_status.message() << std::endl;
return 1;
}
// Wait for the result callback.
std::this_thread::sleep_for(std::chrono::seconds(3));
if (!result_received.load()) {
std::cout << "[4] Sent message (no result confirmation within timeout)" << std::endl;
}
// Close the upstream stream and client explicitly.
// Note: Destructors also handle cleanup, but explicit calls are shown
// for clarity and to match Go's defer pattern.
upstream->Close();
std::cout << "[5] Upstream stream closed" << std::endl;
auto close_status = client->Close();
if (!close_status.ok()) {
std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
return 1;
}
std::cout << "[6] Client closed" << std::endl;
return 0;
}How It Works
- Sends messages via upstream stream with delay using
SetDelaySeconds(). - Delayed messages are not available for consumption until the delay expires.
- The delay is specified in seconds from the time of sending.
- Useful for scheduling tasks and implementing retry backoff.
Related
Was this page helpful?