Purge Queue
Purge all pending messages from a KubeMQ Queue channel using the C++ SDK administration API to reset its state.
Overview
Purging a queue is a management-plane operation for wiping a channel's backlog without receiving and discarding messages one at a time. Reach for it when a bad producer floods a channel, when you need a clean slate between test runs, or when you're resetting a queue during a maintenance window — all without deleting and recreating the channel itself.
AckAllQueueMessages tells the broker directly to acknowledge and drop every message still pending on the channel, entirely server-side. You give it a channel and a wait_time_seconds drain window so the broker can settle in-flight deliveries before finalizing, and it hands back an affected_messages count so you can confirm exactly how much was cleared.
Gotchas: the purge is irreversible — there's no undo once messages are acknowledged away. It only reaches messages still waiting in the queue; anything already delivered to and held by an active consumer is untouched, so a purge run right after a receive can still leave stragglers. And purging empties the channel, it doesn't delete it — new messages can be sent immediately afterward.
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: management/purge_queue
//
// Demonstrates purging all messages from a queue using AckAllQueueMessages.
// This effectively removes all pending messages from the queue.
//
// Channel: cpp-management.purge-queue
// Client ID: cpp-management-purge-queue-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
#include <kubemq/kubemq.h>
#include <iostream>
#include <string>
int main() {
std::cout << "[1] Connecting to localhost:50000" << std::endl;
kubemq::ClientOptions options;
options.set_address("localhost", 50000);
options.set_client_id("cpp-management-purge-queue-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-management.purge-queue";
// Send some messages to the queue.
for (int i = 1; i <= 5; i++) {
auto msg_result = kubemq::QueueMessage::Builder()
.SetChannel(channel)
.SetBody("to-purge-" + std::to_string(i))
.Build();
if (!msg_result.ok()) {
std::cerr << "[ERROR] Build message: " << msg_result.status().message() << std::endl;
return 1;
}
auto send_result = client->SendQueueMessage(*msg_result);
if (!send_result.ok()) {
std::cerr << "[ERROR] SendQueueMessage: " << send_result.status().message()
<< std::endl;
return 1;
}
}
std::cout << "[2] Sent 5 messages to queue" << std::endl;
// Purge the queue by acknowledging all messages.
kubemq::AckAllQueueMessagesRequest req;
req.channel = channel;
req.wait_time_seconds = 5;
auto purge_result = client->AckAllQueueMessages(req);
if (!purge_result.ok()) {
std::cerr << "[ERROR] AckAllQueueMessages: " << purge_result.status().message()
<< std::endl;
return 1;
}
if (purge_result->is_error) {
std::cerr << "[ERROR] Purge warning: " << purge_result->error << std::endl;
} else {
std::cout << "[3] Purged " << purge_result->affected_messages << " messages from queue"
<< std::endl;
}
// Close the client explicitly.
// Note: Client destructor would also handle cleanup (RAII)
auto close_status = client->Close();
if (!close_status.ok()) {
std::cerr << "[ERROR] Close failed: " << close_status.message() << std::endl;
return 1;
}
std::cout << "[4] Client closed" << std::endl;
return 0;
}How It Works
- Sends 5 messages to a queue, then purges them all with
AckAllQueueMessages(). - The request specifies the channel and a wait timeout for the operation.
- The response includes the count of affected (purged) messages.
- Effectively removes all pending messages from the queue.
Related
Was this page helpful?