# Commands and Queries (RPC) (/integrations/aspire/how-to/commands-queries)



## Overview [#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](/learn/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 `await`s 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](/integrations/aspire/how-to/events).

<Mermaid
  chart="sequenceDiagram
    participant R as Requester
    participant K as KubeMQ
    participant H as Handler
    H->>K: SubscribeToCommandsAsync (channel)
    R->>K: SendCommandAsync (RequestId, ReplyChannel)
    K->>H: deliver command
    H->>K: SendCommandResponseAsync (RequestId, Executed)
    K-->>R: CommandResponse (Executed, Error)"
/>

<Callout type="info">
  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](/integrations/aspire/tutorials/getting-started).
</Callout>

## Send a command [#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:

```csharp title="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 [#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):

```csharp title="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();
```

<Callout type="info">
  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`.
</Callout>

## Request-reply round trip [#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:

```csharp title="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 [#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:

```csharp title="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`.

<Callout type="warn">
  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.
</Callout>

## Send a query [#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:

```csharp title="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`:

```csharp title="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:

```csharp title="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 [#replychannel-and-requestid]

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

<TypeTable
  type="{
  RequestId: {
    description: 'Unique id assigned by KubeMQ to each request. The handler echoes it back so the broker can correlate the response with the waiting caller.',
    type: 'string',
  },
  ReplyChannel: {
    description: 'Ephemeral channel the broker created for this request. The handler posts its response here. Echo the incoming value verbatim.',
    type: 'string',
  },
}"
/>

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.

<Callout type="warn">
  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".
</Callout>

## More RPC examples [#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:

<Cards>
  <Card title="Commands: SendCommand" description="Send a command and await execution confirmation from a handler on commands.send." />

  <Card title="Commands: HandleCommand" description="Subscribe to a command channel and reply with Executed=true." />

  <Card title="Commands: CommandTimeout" description="Force a KubeMQTimeoutException with a short deadline and no handler online." />

  <Card title="Commands: ConsumerGroup" description="Load-balance command handling across a group of subscribers." />

  <Card title="Queries: SendQuery" description="Send a query and decode the returned Body payload, inspecting CacheHit." />

  <Card title="Queries: HandleQuery" description="Subscribe to a query channel and reply with a data Body." />

  <Card title="Queries: ConsumerGroup" description="Distribute incoming queries across a group of responders." />

  <Card title="Queries: CachedResponse" description="Serve repeated queries from the broker's response cache." />
</Cards>

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 [#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](/integrations/aspire/tutorials/getting-started) so Aspire only routes traffic once a handler is ready.

## Related [#related]

<Cards>
  <Card href="/learn/rpc" title="RPC" description="The core KubeMQ request-response model — Commands, Queries, correlation, and response caching." />

  <Card href="/integrations/aspire/how-to/events" title="Pub/Sub & Events Store" description="Fire-and-forget events and persistent, replayable event streams." />

  <Card href="/integrations/aspire/how-to/queues" title="Queues" description="Durable, at-least-once message queues with acknowledgements." />

  <Card href="/integrations/aspire/reference/client-api" title="Client API" description="Full client registration API, health checks, and configuration options." />
</Cards>