Send Command
Send a KubeMQ command and wait for an execution response from a handler using the Python SDK.
Overview
A command is KubeMQ's fire-and-confirm RPC pattern: reach for it when you need to know an action actually ran on the other end — "restart the service" — but don't need data back, just a yes/no on execution. It sits between one-way pub/sub, which gives no confirmation, and a query, which returns a result payload. Commands turn "I hope that worked" into a definite outcome your caller can branch on.
This sample builds that lesson: a handler subscribes via subscribe_to_commands as an async generator, and the sender calls send_command, which blocks the coroutine until a CommandResponse arrives or timeout_in_seconds elapses. The reply is built with CommandResponse(command_received=cmd, is_executed=True) — KubeMQ routes it back by correlation ID, so the sender never needs to know the responder's identity.
Gotchas: if no handler is subscribed (or it's still starting up), send_command blocks for the full timeout_in_seconds before raising KubeMQTimeoutError — there's no fast "nobody's listening" error. A handler that never calls send_response leaves the caller hanging until timeout. And a command's response carries no business data — if you need the handler to return a value, use a query instead.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Send command — send a command and receive a response."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
CommandMessage,
CommandReceived,
CommandResponse,
CommandsSubscription,
KubeMQConnectionError,
KubeMQError,
KubeMQTimeoutError,
)
async def main() -> None:
try:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-commands-send-command-client",
) as client:
# Set up a command responder
token = AsyncCancellationToken()
async def command_handler() -> None:
async for cmd in client.subscribe_to_commands(
subscription=CommandsSubscription(
channel="python-commands.send-command",
on_receive_command_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
print(f"Responder received: {cmd.body.decode('utf-8')}")
await client.send_response(
CommandResponse(
command_received=cmd,
is_executed=True,
)
)
handler_task = asyncio.create_task(command_handler())
await asyncio.sleep(1)
# Send a command and get the response
response = await client.send_command(
CommandMessage(
channel="python-commands.send-command",
body=b"hello kubemq, please reply!",
timeout_in_seconds=10,
)
)
print(
f"Response: executed={response.is_executed}, "
f"timestamp={response.timestamp}, error={response.error}"
)
token.cancel()
handler_task.cancel()
try:
await handler_task
except asyncio.CancelledError:
pass
except KubeMQConnectionError as e:
print(f"Connection error: {e}")
except KubeMQTimeoutError as e:
print(f"Timeout error: {e}")
except KubeMQError as e:
print(f"KubeMQ error: {e}")
if __name__ == "__main__":
asyncio.run(main())
# Expected output:
# Responder received: hello kubemq, please reply!
# Response: executed=True, timestamp=<timestamp>, error=
How It Works
send_command blocks the current coroutine until a CommandResponse arrives or timeout_in_seconds elapses. The responder subscribes via subscribe_to_commands as an async generator running in a background task; for each CommandReceived it calls send_response(CommandResponse(command_received=cmd, is_executed=True)). KubeMQ routes the response back by correlation ID — the sender never needs to know the responder's identity. If no responder is available within the timeout, send_command raises KubeMQTimeoutError.
Related
Was this page helpful?