Peek Messages
Peek at messages in a KubeMQ Queue channel without consuming them using the C++ SDK to inspect pending work.
Overview
Peeking lets you look at what's sitting in a queue without touching it — the messages stay exactly where they are, still waiting for whichever consumer eventually receives them. It's the tool you reach for when you need visibility into queue state — checking backlog depth, inspecting payloads while debugging a stuck pipeline, or building an operational dashboard — without risking a collision with real consumers competing for the same work.
ReceiveQueueMessages() with req.is_peek = true is a variant of the same call your consumers use, just in read-only mode: it takes the same channel, max_number_of_messages, and wait_time_seconds, but the broker never marks the returned messages as delivered, locks them, or starts a visibility timeout — so no acknowledgment is needed or even possible.
Gotchas: peeked messages aren't reserved for you — a consumer calling ReceiveQueueMessages() with is_peek = false can remove them the instant after you peek, so treat the count as a point-in-time estimate, not a guarantee. Peek also won't surface messages already locked inside another consumer's in-flight receive, and it's not a substitute for receiving when you actually intend to process what you see.
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/peek_messages
//
// Demonstrates peeking at queue messages without consuming them.
// Uses ReceiveQueueMessages with is_peek=true so messages remain
// in the queue after the call — they are not removed or locked.
//
// Channel: cpp-queues.peek-messages
// Client ID: cpp-queues-peek-messages-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
#include <kubemq/kubemq.h>
#include <iostream>
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-peek-messages-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.peek-messages";
// Send a message to peek at.
std::cout << "[2] Sending message to queue " << channel << std::endl;
auto msg_result = kubemq::QueueMessage::Builder()
.SetChannel(channel)
.SetBody("peek message")
.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;
}
if (send_result->is_error) {
std::cerr << "[ERROR] Send failed: " << send_result->error << std::endl;
return 1;
}
std::cout << "[3] Message sent" << std::endl;
// Peek at messages without consuming them (is_peek=true).
std::cout << "[4] Peeking at queue " << channel << " (non-destructive)" << std::endl;
kubemq::ReceiveQueueMessagesRequest req;
req.channel = channel;
req.max_number_of_messages = 10;
req.wait_time_seconds = 5;
req.is_peek = true; // view without consuming
auto peek_result = client->ReceiveQueueMessages(req);
if (!peek_result.ok()) {
std::cerr << "[ERROR] ReceiveQueueMessages: " << peek_result.status().message()
<< std::endl;
return 1;
}
if (peek_result->is_error) {
std::cerr << "[ERROR] Peek failed: " << peek_result->error << std::endl;
return 1;
}
std::cout << "[5] Peeked: " << peek_result->messages_received
<< " messages (still in queue)" << std::endl;
for (const auto& m : peek_result->messages) {
std::cout << " body=" << m.body() << std::endl;
}
// Close the client explicitly.
// Note: The Client destructor also calls Close(), but explicit
// cleanup is shown here for clarity and to match Go's defer pattern.
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
- Peeks at messages in a queue without removing them.
- Uses
ReceiveQueueMessages()withis_peek=true— messages are never locked or consumed. - Peeked messages remain in the queue for future consumption.
- Useful for monitoring queue depth and inspecting message contents.
Related
Was this page helpful?