Handle Command
Register a handler for incoming KubeMQ commands and send responses using the Python SDK.
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 by iterating 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 async for loop: each iteration yields a CommandReceived with id, channel, body, metadata, and tags; you run your business logic, then send a reply with send_response(CommandResponse(command_received=cmd, is_executed=True)). Passing the original cmd back is what lets the broker correlate the reply to the exact caller blocked on send_command — nothing else identifies which request the response belongs to.
Gotchas: the reply must be sent within timeout_in_seconds or the caller sees a timeout even if you eventually respond; wrap handler logic in try/except since an uncaught exception can silently kill the subscriber task and leave the sender waiting forever; and the handler runs sequentially per subscription, so slow business logic head-of-line blocks the next command.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Handle command — subscribe and respond to incoming commands."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
CommandMessage,
CommandReceived,
CommandResponse,
CommandsSubscription,
)
async def main() -> None:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-commands-handle-command-client",
) as client:
# Subscribe to commands
token = AsyncCancellationToken()
async def command_handler() -> None:
async for cmd in client.subscribe_to_commands(
subscription=CommandsSubscription(
channel="python-commands.handle-command",
on_receive_command_callback=lambda e: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
"""Handle incoming command and send response."""
try:
body = cmd.body.decode("utf-8")
print(f"Handling command: Id={cmd.id}, Body={body}")
# Process the command (simulate work)
success = True
# Send response back
await client.send_response(
CommandResponse(
command_received=cmd,
is_executed=success,
)
)
print(f" Response sent: executed={success}")
except Exception as e:
print(f" Error handling command: {e}")
handler_task = asyncio.create_task(command_handler())
await asyncio.sleep(1)
print("Listening for commands on 'python-commands.handle-command'...")
# Send a test command
response = await client.send_command(
CommandMessage(
channel="python-commands.handle-command",
body=b"process this task",
timeout_in_seconds=10,
)
)
print(f"Command result: executed={response.is_executed}")
await asyncio.sleep(1)
token.cancel()
handler_task.cancel()
try:
await handler_task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
subscribe_to_commands returns an async generator; each iteration yields a CommandReceived with id, channel, body, metadata, and tags. The handler must call send_response(CommandResponse(command_received=cmd, is_executed=True)) within timeout_in_seconds or the sender receives a timeout error. Wrapping the handler logic in try/except prevents an uncaught exception from silently killing the subscriber task and leaving the sender waiting.
Related
Was this page helpful?