Request-Reply
Implement synchronous request-reply messaging with the KubeMQ Python SDK using commands and queries.
Overview
Request-reply gives you synchronous RPC on top of KubeMQ's messaging fabric: a caller sends a query and blocks until the handler actually processing the request sends back a real answer — not just an acknowledgment. Reach for it whenever the caller needs a return value to proceed — a lookup, a computed result, a status check — the same shape as an HTTP call, but routed by KubeMQ instead of a service mesh or DNS.
A handler iterates subscribe_to_queries with async for and, for each incoming QueryReceived, builds a QueryResponse and calls send_response — KubeMQ uses the query's correlation ID under the hood to route that response to the one caller waiting, not broadcast it, so there's no reply channel to wire up manually. The caller's send_query blocks until that response lands or its timeout_in_seconds elapses, then returns a QueryResponse with the reply body.
Gotchas: if no subscriber is listening — or the handler crashes before replying — send_query simply times out; there's no way to distinguish "no handler" from "handler is slow" from the timeout alone. send_response must be called with the exact QueryReceived object it's replying to, since that's where the correlation data lives — build a fresh one and the reply is silently dropped or misrouted. If you don't actually need a return value, use commands instead — they only need an ack, so they don't tie up a caller waiting on a round trip.
Prerequisites
- KubeMQ server running on
localhost:50000 - Python SDK installed (
pip install kubemq)
Code
"""Example: Request-reply pattern — synchronous request with response using queries."""
from __future__ import annotations
import asyncio
from kubemq import (
AsyncCancellationToken,
AsyncCQClient,
QueriesSubscription,
QueryMessage,
QueryResponse,
)
async def main() -> None:
async with AsyncCQClient(
address="localhost:50000",
client_id="python-patterns-request-reply-client",
) as client:
token = AsyncCancellationToken()
async def server() -> None:
async for query in client.subscribe_to_queries(
subscription=QueriesSubscription(
channel="python-patterns.request-reply",
on_receive_query_callback=lambda q: None,
on_error_callback=lambda e: print(f"Error: {e}"),
),
cancellation_token=token,
):
query_body = query.body.decode("utf-8")
print(f"[Server] Received request: {query_body}")
reply_data = f"Processed: {query_body}"
await client.send_response(
QueryResponse(
query_received=query,
is_executed=True,
body=reply_data.encode(),
)
)
server_task = asyncio.create_task(server())
await asyncio.sleep(1)
# "Client" side — send requests and get replies
for i in range(3):
response = await client.send_query(
QueryMessage(
channel="python-patterns.request-reply",
body=f"Request #{i + 1}".encode(),
timeout_in_seconds=10,
)
)
print(f"[Client] Reply: {response.body}")
token.cancel()
server_task.cancel()
try:
await server_task
except asyncio.CancelledError:
pass
if __name__ == "__main__":
asyncio.run(main())
How It Works
A single AsyncCQClient acts as both server and client. The server() coroutine runs as a background task, iterating subscribe_to_queries with async for; for each incoming QueryReceived it builds a QueryResponse and calls send_response. The "client" side then calls send_query which blocks until the response arrives or timeout_in_seconds expires. KubeMQ routes the response back to the original sender by correlation ID — no manual reply-to channel setup is needed.
Related
Was this page helpful?