Request-Reply

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 SubscribeToQueriesAsync, which returns an IAsyncEnumerable<QueryReceived>, and calls SendQueryResponseAsync using the incoming query.RequestId and query.ReplyChannel to route the response back to the exact caller — no manual correlation ID management needed. The caller's SendQueryAsync blocks until that reply lands or TimeoutInSeconds elapses, then returns a response with Executed and Body.

Gotchas: if no subscriber is listening — or the handler crashes before replying — SendQueryAsync simply times out, throwing KubeMQTimeoutException; there's no way to distinguish "no handler" from "handler is slow" from the exception alone. The response must echo back the same RequestId and ReplyChannel from the request it's answering, or 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
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Patterns: Request/Reply
//
// This example demonstrates the request/reply pattern using queries.
// A handler subscribes and responds with data, then a sender issues a query
// and waits for the response payload. Use queries when the caller needs a
// return value — use commands when only an ack is required.
//
// Prerequisites:
//   - KubeMQ server running on localhost:50000
//   - dotnet run

using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Queries;
using System.Text;

await using var client = new KubeMQClient(new KubeMQClientOptions
{
    ClientId = "csharp-patterns-request-reply-client",
});
await client.ConnectAsync();

Console.WriteLine("Connected to KubeMQ server");

// Start the handler in the background
var cts = new CancellationTokenSource();
var handlerTask = Task.Run(async () =>
{
    await foreach (var query in client.SubscribeToQueriesAsync(
        new QueriesSubscription { Channel = "csharp-patterns.request-reply" }, cts.Token))
    {
        var body = Encoding.UTF8.GetString(query.Body.Span);
        Console.WriteLine($"[Handler] Received query: {body}");

        var responseBody = Encoding.UTF8.GetBytes("{\"status\":\"ok\",\"order\":\"processed\"}");

        await client.SendQueryResponseAsync(new QueryResponse
        {
            RequestId = query.RequestId,
            ReplyChannel = query.ReplyChannel!,
            Body = responseBody,
            Executed = true,
        });

        Console.WriteLine("[Handler] Responded with data");
    }
});

// Allow time for subscription to establish
await Task.Delay(1000);

// Send a query (request) and wait for the reply
Console.WriteLine("[Sender] Sending query...");
var response = await client.SendQueryAsync(new QueryMessage
{
    Channel = "csharp-patterns.request-reply",
    Body = Encoding.UTF8.GetBytes("get-order-status"),
    TimeoutInSeconds = 10,
});

Console.WriteLine($"[Sender] Response received: Executed={response.Executed}");
Console.WriteLine($"[Sender] Payload: {Encoding.UTF8.GetString(response.Body.Span)}");

cts.Cancel();
Console.WriteLine("Done.");

How It Works

• SubscribeToQueriesAsync returns an IAsyncEnumerable<QueryReceived>; the handler iterates it in a background Task.Run while the main thread acts as the sender.
• SendQueryResponseAsync uses the incoming query.RequestId and query.ReplyChannel to route the response back to the exact caller — no manual correlation ID management is needed.
• TimeoutInSeconds = 10 on the QueryMessage tells the server how long to hold the reply channel open; if no handler responds within the window the SDK throws KubeMQTimeoutException.
• Queries carry a return payload (Body) — use commands instead when only an acknowledgement (no data) is required.

Related

• Pattern overview
• C# SDK Reference
• Fan-Out
• Work Queue