Send Command

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.subscribe_to_commands(...) registers an async callback, and client.send_command(command) blocks until the handler replies or the timeout set on CommandBuilder expires. The handler builds its reply with CommandReplyBuilder::new().request_id(&cmd.id).response_to(&cmd.response_to) — 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), send_command waits out the full timeout before returning an error — there's no fast "nobody's listening" error. A handler that omits request_id/response_to 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
• Rust SDK installed (cargo add kubemq)

Code

use kubemq::prelude::*;
use kubemq::{CommandBuilder, 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-send";

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

    tokio::time::sleep(Duration::from_millis(500)).await;

    let command = CommandBuilder::new()
        .channel(channel)
        .body(b"do-something".to_vec())
        .metadata("command-metadata")
        .timeout(Duration::from_secs(10))
        .build();

    let response = client.send_command(command).await?;
    println!(
        "Command response: executed={}, error='{}'",
        response.executed, response.error
    );

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

How It Works

• The handler clones the client (client.clone() is cheap, Arc-based) to send the reply from within the callback.
• CommandReplyBuilder must set request_id and response_to from the received CommandReceive.
• The reply is sent in a tokio::spawn to avoid blocking the subscription task.
• 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 Handle
• Command Timeout