Expiration Policy
Configure message expiration TTL on a KubeMQ Queue via the upstream stream API in the C++ SDK to drop stale messages.
Overview
An expiration policy puts a hard time limit on how long a queue message may sit unconsumed. It solves a different problem than a dead-letter policy — this isn't about messages that fail processing, it's about messages that go stale: a price quote, a one-time code, a cache-invalidation signal, where late delivery is actively wrong, not just delayed. Instead of every consumer re-checking timestamps itself, the deadline lives on the message and the broker enforces it.
At the API level, .SetExpirationSeconds(60) attaches a per-message TTL when you build the QueueMessage, and the clock starts the moment the broker accepts it via upstream->Send, not when a consumer picks it up. Let the TTL elapse unconsumed and the broker silently removes it — a later poll just comes back empty, no error, no trace.
Gotchas: expiration is silent — no DLQ routing, no event, just a message that vanishes — so pair it with monitoring if you need visibility into how much work is being dropped. The timer starts at send time, not when a consumer picks up the work, so a message can expire mid-backlog even while a consumer is actively polling. And setting the TTL too short for your real consumer lag just turns ordinary slowness into silent data loss.
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/expiration_policy
//
// Demonstrates sending queue messages with an expiration (TTL) policy.
// Messages that are not consumed before the expiration time are automatically
// removed from the queue.
//
// Channel: cpp-queues.expiration-policy
// Client ID: cpp-queues-expiration-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-expiration-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.expiration-policy";
std::atomic<bool> result_received{false};
// Open an upstream stream for sending messages with expiration.
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 60s expiration 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 60-second expiration.
auto msg_result = kubemq::QueueMessage::Builder()
.SetChannel(channel)
.SetBody("expires in 60s")
.SetExpirationSeconds(60)
.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 expiration_seconds=60" << std::endl;
auto send_status = upstream->Send("req-expiration", 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 TTL using
SetExpirationSeconds(). - Messages that are not consumed before expiration are automatically removed.
- The expiration is specified in seconds from the time of sending.
- Useful for time-sensitive messages that lose relevance after a period.
Related
Was this page helpful?