Send Command
Send a KubeMQ Command with the C++ SDK and receive the handler's execution confirmation response.
Overview
A command is KubeMQ's fire-and-confirm RPC pattern: you reach for it when you need to know that an action actually ran on the other end — "do-something" — but you don't need any data back, just a yes/no on execution. It's the middle ground between one-way pub/sub, where you get no confirmation at all, and a query, where the handler returns a result payload. Commands turn "I hope that worked" into a definite outcome your caller can branch on.
This sample builds that lesson: client->SubscribeToCommands() registers the handler, and client->SendCommand() blocks until the handler replies or the timeout set with Command::Builder().SetTimeout(...) expires. The handler builds its reply with CommandReply::Builder().SetRequestId(cmd.id).SetResponseTo(cmd.response_to).SetExecuted(true) — that correlation is what lets the broker route the response back to the exact caller waiting on it.
Gotchas: if no handler is subscribed (or it's still starting up), SendCommand() blocks for the full timeout before returning a failed status — there's no fast "nobody's listening" error. A handler that omits SetRequestId/SetResponseTo on the reply leaves the caller hanging until timeout. And a command's reply carries no business data — if you need the handler to return a value, use a query instead.
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: commands/send_command
//
// Demonstrates sending a command (RPC-style request) and receiving a response.
// A handler subscribes to the command channel, processes the command,
// and sends back an execution response.
//
// Channel: cpp-commands.send-command
// Client ID: cpp-commands-send-command-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>
int main() {
std::cout << "[1] Connecting to localhost:50000" << std::endl;
kubemq::ClientOptions options;
options.set_address("localhost", 50000);
options.set_client_id("cpp-commands-send-command-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-commands.send-command";
std::atomic<bool> command_handled{false};
// Subscribe to handle incoming commands on the channel.
std::cout << "[2] Subscribing to commands on channel " << channel << std::endl;
auto sub_result = client->SubscribeToCommands(
channel, "",
[&client, &command_handled](const kubemq::CommandReceive& cmd) {
std::cout << "[4] Command received: channel=" << cmd.channel << " body=" << cmd.body
<< std::endl;
// Build and send a response indicating successful execution.
auto now_epoch = std::chrono::duration_cast<std::chrono::seconds>(
std::chrono::system_clock::now().time_since_epoch())
.count();
auto reply_result = kubemq::CommandReply::Builder()
.SetRequestId(cmd.id)
.SetResponseTo(cmd.response_to)
.SetBody("executed")
.SetExecuted(true)
.SetExecutedAt(now_epoch)
.Build();
if (!reply_result.ok()) {
std::cerr << "[ERROR] Build reply: " << reply_result.status().message()
<< std::endl;
return;
}
auto send_status = client->SendCommandResponse(*reply_result);
if (!send_status.ok()) {
std::cerr << "[ERROR] SendCommandResponse: " << send_status.message() << std::endl;
return;
}
command_handled.store(true);
},
[](const kubemq::Status& err) {
std::cerr << "[ERROR] Command subscription error: " << err.message() << std::endl;
});
if (!sub_result.ok()) {
std::cerr << "[ERROR] SubscribeToCommands: " << sub_result.status().message() << std::endl;
return 1;
}
auto& sub = *sub_result;
// Allow subscription to fully establish before sending.
std::this_thread::sleep_for(std::chrono::milliseconds(300));
// Build and send a command with a 10-second timeout.
std::cout << "[3] Sending command to channel " << channel << std::endl;
auto cmd_result = kubemq::Command::Builder()
.SetChannel(channel)
.SetBody("do-something")
.SetTimeout(std::chrono::seconds(10))
.Build();
if (!cmd_result.ok()) {
std::cerr << "[ERROR] Build command: " << cmd_result.status().message() << std::endl;
return 1;
}
auto resp_result = client->SendCommand(*cmd_result);
if (!resp_result.ok()) {
std::cerr << "[ERROR] SendCommand: " << resp_result.status().message() << std::endl;
return 1;
}
std::cout << "[5] Command response: executed=" << std::boolalpha << resp_result->executed
<< std::endl;
// Wait for the handler to finish processing.
std::this_thread::sleep_for(std::chrono::seconds(1));
// Cancel the subscription explicitly.
// Note: The Subscription destructor also calls Cancel(), but explicit
// cleanup is shown here for clarity and to match Go's defer pattern.
sub->Cancel();
std::cout << "[6] Subscription cancelled" << 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 << "[7] Client closed" << std::endl;
return 0;
}How It Works
- Subscribes to handle commands with
SubscribeToCommands(). - The handler builds a
CommandReplywithSetExecuted(true)and sends it back. - Sends a command using
Command::Builder()with channel, body, and timeout. - The
SendCommand()call blocks until the handler responds or the timeout expires.
Related
Was this page helpful?