Overview

This page covers how to use KubeMQ Commands and Queries — the two request-response (RPC) patterns — from a service wired by the Aspire integration. For the RPC model itself (correlation, timeouts, response caching), see the core RPC docs. Below is the Aspire-specific part: how the injected IKubeMQClient exposes the Commands and Queries APIs.

KubeMQ RPC gives you synchronous, request-response messaging over the same Aspire-managed IKubeMQClient you already use for events and queues. It comes in two flavors:

Commands — a request that asks a handler to do something and report back. The response carries an execution confirmation (Executed) and an optional Error, but no data payload. Use commands when you only care whether the action succeeded.
Queries — a request that asks a handler for data. The response carries Executed, an optional Error, and a Body payload you decode on the requester side. Use queries when you expect a result back.

Both patterns are fully synchronous: the caller awaits the response and blocks (asynchronously) until the handler replies or the deadline elapses. Because the round trip is request-response rather than fanout, a handler must be subscribed before the request is sent — there is no buffering and no fanout. This is the key difference from fire-and-forget events.

Both commands and queries flow through the gRPC transport configured by AddKubeMQClient in your service project. No extra registration is required beyond the standard Aspire wiring — see Getting Started.

Send a command

A command is dispatched with SendCommandAsync, passing a CommandMessage with the target Channel, the Body payload, and a TimeoutInSeconds deadline. The returned CommandResponse reports whether the handler executed the action and surfaces any error text. The sample Web API exposes this as a controller action that resolves IKubeMQClient from DI:

Controllers/CommandsController.cs

using System.Text;
using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Commands;
using Microsoft.AspNetCore.Mvc;

namespace KubeMQ.Aspire.Sample.WebApi.Controllers;

[ApiController]
[Route("api/[controller]")]
public sealed class CommandsController : ControllerBase
{
    private readonly IKubeMQClient _client;

    public CommandsController(IKubeMQClient client) => _client = client;

    /// <summary>
    /// Sends a command to the "commands.example" channel.
    /// A responder must be subscribed for this to succeed.
    /// </summary>
    [HttpPost]
    public async Task<IActionResult> SendCommand([FromBody] string body)
    {
        var command = new CommandMessage
        {
            Channel = "commands.example",
            Body = Encoding.UTF8.GetBytes(body),
            TimeoutInSeconds = 10,
        };

        var response = await _client.SendCommandAsync(command);
        return Ok(new { response.Executed, response.Error });
    }
}

Executed is true when a handler accepted and processed the command; Error holds the failure reason when it is false.

Handle a command

A handler subscribes to a channel with SubscribeToCommandsAsync, passing a CommandsSubscription. The call returns an IAsyncEnumerable you iterate with await foreach — each item is an incoming command carrying a RequestId and a ReplyChannel. To complete the round trip, the handler echoes both values back in a CommandResponse via SendCommandResponseAsync, setting Executed = true (or false to signal failure):

Commands.HandleCommand/Program.cs

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

var builder = Host.CreateApplicationBuilder(args);
builder.AddServiceDefaults();
builder.AddKubeMQClient("messaging");

var host = builder.Build();
var client = host.Services.GetRequiredService<IKubeMQClient>();
var logger = host.Services.GetRequiredService<ILogger<Program>>();
var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();

await client.ConnectAsync();
logger.LogInformation("Connected to KubeMQ — waiting for commands on commands.send");

var stoppingToken = lifetime.ApplicationStopping;

_ = Task.Run(async () =>
{
    try
    {
        var subscription = new CommandsSubscription { Channel = "commands.send" };
        await foreach (var cmd in client.SubscribeToCommandsAsync(subscription, stoppingToken))
        {
            var body = Encoding.UTF8.GetString(cmd.Body.Span);
            logger.LogInformation("Received command: {Body}", body);

            // Process and respond
            await client.SendCommandResponseAsync(new CommandResponse
            {
                RequestId = cmd.RequestId,
                ReplyChannel = cmd.ReplyChannel!,
                Executed = true,
            }, stoppingToken);

            logger.LogInformation("Responded: executed=true");
        }
    }
    catch (OperationCanceledException) { /* shutting down */ }
}, stoppingToken);

await host.RunAsync();

Subscriptions run for the lifetime of the host. Pass IHostApplicationLifetime.ApplicationStopping as the cancellation token so the await foreach loop unwinds cleanly on shutdown, and swallow the resulting OperationCanceledException.

Request-reply round trip

Putting both halves together gives you a complete request-reply exchange. The Patterns.RequestReply example runs a responder and a requester in the same host over the commands.reqreply channel. The responder subscribes first; the requester waits briefly for the subscription to establish, then fires three commands with a 10-second TimeoutInSeconds and logs each Executed reply:

Patterns.RequestReply/Program.cs

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

var builder = Host.CreateApplicationBuilder(args);
builder.AddServiceDefaults();
builder.AddKubeMQClient("messaging");

var host = builder.Build();
var client = host.Services.GetRequiredService<IKubeMQClient>();
var logger = host.Services.GetRequiredService<ILogger<Program>>();
var lifetime = host.Services.GetRequiredService<IHostApplicationLifetime>();

await client.ConnectAsync();
var stoppingToken = lifetime.ApplicationStopping;

// Responder
_ = Task.Run(async () =>
{
    try
    {
        var subscription = new CommandsSubscription { Channel = "commands.reqreply" };
        await foreach (var cmd in client.SubscribeToCommandsAsync(subscription, stoppingToken))
        {
            logger.LogInformation("[Responder] Received: {Body}", Encoding.UTF8.GetString(cmd.Body.Span));
            await client.SendCommandResponseAsync(new CommandResponse
            {
                RequestId = cmd.RequestId,
                ReplyChannel = cmd.ReplyChannel!,
                Executed = true,
            }, stoppingToken);
        }
    }
    catch (OperationCanceledException) { }
}, stoppingToken);

// Requester
_ = Task.Run(async () =>
{
    try
    {
        await Task.Delay(2000, stoppingToken);
        for (var i = 1; i <= 3; i++)
        {
            var response = await client.SendCommandAsync(new CommandMessage
            {
                Channel = "commands.reqreply",
                Body = Encoding.UTF8.GetBytes($"Request #{i}"),
                TimeoutInSeconds = 10
            }, stoppingToken);
            logger.LogInformation("[Requester] Reply #{I}: Executed={Executed}", i, response.Executed);
        }
    }
    catch (OperationCanceledException) { }
}, stoppingToken);

await host.RunAsync();

In a real deployment the responder and requester live in separate services. The Commands.SendCommand and Commands.HandleCommand examples split exactly this exchange across two projects on the commands.send channel — start the handler first, then the sender.

Timeout handling

If no handler is subscribed (or none replies within TimeoutInSeconds), the send call throws KubeMQTimeoutException. Always guard a request with a try/catch so an offline handler degrades gracefully instead of bubbling up as an unhandled fault:

Commands.SendCommand/Program.cs

try
{
    await Task.Delay(3000, stoppingToken);
    var response = await client.SendCommandAsync(new CommandMessage
    {
        Channel = "commands.send",
        Body = Encoding.UTF8.GetBytes("restart-service"),
        TimeoutInSeconds = 10
    }, stoppingToken);

    logger.LogInformation("Command result: Executed={Executed}, Error={Error}",
        response.Executed, response.Error ?? "none");
}
catch (KubeMQTimeoutException)
{
    logger.LogWarning("Command timed out — no handler responded");
}
catch (OperationCanceledException) { }

The dedicated Commands.CommandTimeout example forces this path with a 2-second timeout and no handler, also catching KubeMQOperationException for broker-side failures. Both exception types live in KubeMQ.Sdk.Exceptions.

Choose TimeoutInSeconds deliberately. Too short and a slow-but-healthy handler looks like a failure; too long and a stuck caller holds resources. Match the value to the handler's realistic worst-case processing time.

Send a query

A query works like a command but returns a payload. Call SendQueryAsync with a QueryMessage (Channel, Body, TimeoutInSeconds); the QueryResponse carries Executed, Error, and a Body you decode. The sample controller decodes the UTF-8 response body and returns it to the HTTP caller:

Controllers/QueriesController.cs

using System.Text;
using KubeMQ.Sdk.Client;
using KubeMQ.Sdk.Queries;
using Microsoft.AspNetCore.Mvc;

namespace KubeMQ.Aspire.Sample.WebApi.Controllers;

[ApiController]
[Route("api/[controller]")]
public sealed class QueriesController : ControllerBase
{
    private readonly IKubeMQClient _client;

    public QueriesController(IKubeMQClient client) => _client = client;

    /// <summary>
    /// Sends a query to the "queries.example" channel.
    /// A responder must be subscribed for this to succeed.
    /// </summary>
    [HttpPost]
    public async Task<IActionResult> SendQuery([FromBody] string body)
    {
        var query = new QueryMessage
        {
            Channel = "queries.example",
            Body = Encoding.UTF8.GetBytes(body),
            TimeoutInSeconds = 10,
        };

        var response = await _client.SendQueryAsync(query);
        var responseBody = response.Body.Length > 0
            ? Encoding.UTF8.GetString(response.Body.Span)
            : "(empty)";

        return Ok(new { response.Executed, response.Error, Body = responseBody });
    }
}

The query handler side mirrors the command handler: subscribe with SubscribeToQueriesAsync(new QueriesSubscription { Channel }), then reply with SendQueryResponseAsync(new QueryResponse { ... }) — the difference being that a query response sets a Body:

Queries.HandleQuery/Program.cs

var subscription = new QueriesSubscription { Channel = "queries.send" };
await foreach (var query in client.SubscribeToQueriesAsync(subscription, stoppingToken))
{
    var body = Encoding.UTF8.GetString(query.Body.Span);
    logger.LogInformation("Received query: {Body}", body);

    // Respond with data
    await client.SendQueryResponseAsync(new QueryResponse
    {
        RequestId = query.RequestId,
        ReplyChannel = query.ReplyChannel!,
        Executed = true,
        Body = Encoding.UTF8.GetBytes("{\"name\":\"John\",\"role\":\"admin\"}")
    }, stoppingToken);
}

On the requester side, check Executed before reading the body, and inspect response.CacheHit to know whether the result was served from the broker's query cache:

Queries.SendQuery/Program.cs

var response = await client.SendQueryAsync(new QueryMessage
{
    Channel = "queries.send",
    Body = Encoding.UTF8.GetBytes("get-user-info"),
    TimeoutInSeconds = 10
}, stoppingToken);

if (response.Executed)
{
    var data = Encoding.UTF8.GetString(response.Body.Span);
    logger.LogInformation("Query response: {Data}, CacheHit={CacheHit}",
        data, response.CacheHit);
}
else
{
    logger.LogWarning("Query failed: {Error}", response.Error);
}

ReplyChannel and RequestId

The round trip is completed by two correlation fields that the handler must copy from the incoming message into its response:

Prop

Type

In every handler example (Commands.HandleCommand, Patterns.RequestReply, Queries.HandleQuery) the pattern is identical: read cmd.RequestId / cmd.ReplyChannel (or query.RequestId / query.ReplyChannel) off the received message and assign them to the CommandResponse / QueryResponse. The ! on ReplyChannel! asserts the value is non-null for an incoming request — every delivered command or query carries one.

If a handler does not echo the exact RequestId and ReplyChannel, the response never reaches the caller and the request fails with a timeout even though the handler "ran".

More RPC examples

The example suite ships several additional RPC scenarios, registered in the AppHost under the Commands and Queries regions. Each is a standalone runnable project:

Commands: SendCommand

Send a command and await execution confirmation from a handler on commands.send.

Commands: HandleCommand

Subscribe to a command channel and reply with Executed=true.

Commands: CommandTimeout

Force a KubeMQTimeoutException with a short deadline and no handler online.

Commands: ConsumerGroup

Load-balance command handling across a group of subscribers.

Queries: SendQuery

Send a query and decode the returned Body payload, inspecting CacheHit.

Queries: HandleQuery

Subscribe to a query channel and reply with a data Body.

Queries: ConsumerGroup

Distribute incoming queries across a group of responders.

Queries: CachedResponse

Serve repeated queries from the broker's response cache.

Run any of them through the shared AppHost — for example dotnet run --project examples/AppHost provisions the KubeMQ container and starts the registered projects.

Handler-before-request

The defining constraint of RPC is that the request and the handler must overlap in time. Unlike fire-and-forget events — which fan out to whoever happens to be subscribed and are simply lost otherwise — a command or query that finds no live handler does not queue or retry; it blocks until TimeoutInSeconds elapses and then throws KubeMQTimeoutException. Every send-side example deliberately delays the request (await Task.Delay(...)) to give the handler time to subscribe first. In production, deploy and start handler services before the services that call them, and use health checks so Aspire only routes traffic once a handler is ready.

Commands and Queries (RPC)