Handle Query

Overview

A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.

Registering a handler by iterating SubscribeToQueriesAsync opens a subscription; the broker delivers every matching QueryReceived to your await foreach loop as it arrives. The handler builds a QueryResponse carrying the original RequestId and ReplyChannel back to the broker, so the answer routes to the specific caller blocked waiting, and sets Body with the real result and Executed = true before calling SendQueryResponseAsync.

Gotchas: if the handler never sends a response, the caller blocks until its own timeout elapses and fails with a timeout, not a fast error. An exception inside the loop body doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query arrives on the same stream reader, slow handler code delays every other in-flight caller.

Prerequisites

• KubeMQ server running on localhost:50000
• C# SDK installed (dotnet add package KubeMQ.SDK.CSharp)

Code

// KubeMQ .NET SDK — Queries: Handle Query
//
// This example demonstrates subscribing to incoming queries and responding with data.
// Run this before Queries.SendQuery to handle the request.
//
// 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-queries-handle-query-client",
});
await client.ConnectAsync();

Console.WriteLine("Waiting for queries on 'csharp-demo.queries'...");

var cts = new CancellationTokenSource();
Console.CancelKeyPress += (_, e) =>
{
    e.Cancel = true;
    cts.Cancel();
};

await foreach (var query in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "csharp-queries.handle-query" }, cts.Token))
{
    var body = Encoding.UTF8.GetString(query.Body.Span);
    Console.WriteLine($"Received query: {body}");

    // Process query and respond with data
    var responseBody = Encoding.UTF8.GetBytes("{\"id\":42,\"name\":\"Alice\",\"email\":\"alice@example.com\"}");

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

    Console.WriteLine("  -> Responded with user data");
}

Console.WriteLine("Done.");

How It Works

• SubscribeToQueriesAsync returns an IAsyncEnumerable<QueryReceived>. The await foreach yields one incoming query per iteration, running on the gRPC stream reader.
• The handler produces a response body (here a JSON string) and calls SendQueryResponseAsync with RequestId and ReplyChannel from the query — the broker correlates and routes the reply to the original sender.
• Executed = true tells the sender that the handler successfully processed the query. Set it to false and populate the Error field to signal a failure.
• The CancellationTokenSource wired to Ctrl+C gives the long-running handler a clean shutdown path without dropping in-flight queries.

Related

• Pattern overview
• C# SDK Reference
• Send Query
• Cached Query