Handle Command

Overview

A command handler is the receiving side of KubeMQ's Commands pattern — the code that actually does the work a caller is blocked waiting on. Instead of building your own request-routing layer on top of a queue, you register a handler once with subscribe_to_commands, and KubeMQ delivers every matching command on that channel to it as a long-lived, server-streamed subscription, turning the channel into a synchronous RPC endpoint.

Handling happens inside the closure passed to subscribe_to_commands: you read the command's id and body, run your business logic, then build a reply with CommandReplyBuilder::new().request_id(&cmd.id).response_to(&cmd.response_to) and send it with send_command_response. Copying request_id and response_to from the received command is what lets the broker correlate the reply back to the exact caller blocked on the send call — nothing else identifies which request the response belongs to.

Gotchas: the reply must reach the broker before the caller's timeout elapses or the caller sees a timeout even if you eventually respond; call .error("message") on the builder to signal a failed execution instead of a successful one; and the closure runs on the subscription's async task, so slow or blocking business logic head-of-line blocks the next command.

Prerequisites

• KubeMQ server running on localhost:50000
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::CommandReplyBuilder;
use std::time::Duration;

#[tokio::main]
async fn main() -> kubemq::Result<()> {
    let client = KubemqClient::builder()
        .host("localhost")
        .port(50000)
        .build()
        .await?;

    let channel = "rust-rpc.command-handle";
    let rc = client.clone();

    let sub = client.subscribe_to_commands(
        channel, "",
        move |cmd| {
            let c = rc.clone();
            Box::pin(async move {
                println!("Processing command: id={}, body={}", cmd.id, String::from_utf8_lossy(&cmd.body));
                let reply = CommandReplyBuilder::new()
                    .request_id(&cmd.id)
                    .response_to(&cmd.response_to)
                    .build();
                let _ = c.send_command_response(reply).await;
            })
        },
        None,
    ).await?;

    println!("Listening for commands on '{}'...", channel);
    tokio::time::sleep(Duration::from_secs(30)).await;

    sub.unsubscribe().await;
    client.close().await?;
    Ok(())
}

How It Works

• The handler receives CommandReceive with id, body, response_to, and other fields.
• CommandReply with no .error() marks the command as executed successfully.
• Call .error("message") on the builder to report a failure.
• Review timeouts, channel names, and client IDs before running against shared environments.
• Run the program while the server from the prerequisites is available.

Related

• RPC overview
• Rust SDK Reference
• Command Send