# Commands & Queries (Request/Response) (/integrations/masstransit/how-to/commands-queries) ## Overview [#overview] MassTransit's request/response abstraction — `IRequestClient.GetResponse()` on the caller and `context.RespondAsync(result)` on the responder — maps to KubeMQ's native **Commands/Queries (CQ)** pattern. CQ provides built-in request-reply: each request carries a `ReplyChannel` and a timeout, and KubeMQ correlates the response back to the caller for you. Unlike RabbitMQ or Amazon SQS, which *simulate* request/response with a temporary reply queue per request, KubeMQ has request-reply as a first-class primitive — no temporary queues to create or tear down. → For what Commands and Queries are and KubeMQ's CQ semantics, see [Commands & Queries (RPC)](/learn/rpc). This page documents the MassTransit transport's API surface only. *A request flows over the CQ transport; the broker delivers it to the responder and routes the reply back to the original requester — no temporary reply queue.* ## API surface [#api-surface] | Member | Where | Purpose | | ------------------------------------------------------------------------------------------- | ---------------- | ------------------------------------------------------------------------------- | | `IRequestClient.GetResponse(request)` | caller | Send a request and await the typed `Response` | | `ConsumeContext.RespondAsync(result)` | responder | Reply over the CQ reply channel | | `KubeMQTransportOptions.DefaultCqMode` | options | Global default CQ mode (`Queries` or `Commands`); binds from `appsettings.json` | | `IKubeMQBusFactoryConfigurator.UseCommandsForRequestResponse()` | bus factory | Set the global CQ mode to Commands | | `IKubeMQReceiveEndpointConfigurator.UseQueries()` / `.UseCommands()` | receive endpoint | Override CQ mode for one endpoint | | `IKubeMQRider.SendRequestAsync(request, channelName, timeoutSeconds = 30, ct)` | rider | Send a request over CQ directly, with an explicit timeout | ### Two CQ modes: Queries vs Commands [#two-cq-modes-queries-vs-commands] KubeMQ CQ has two modes, modeled by the `CqMode` enum in the transport: ```csharp title="CqMode.cs" public enum CqMode { // Use KubeMQ Queries for request/response (default). Returns data. Queries = 0, // Use KubeMQ Commands for request/response. Fire-and-await-ack only. Commands = 1, } ``` | Mode | Request | Response | Use for | | --------------------- | ------------ | ------------------------------------------------------------- | ----------------------------------------------------- | | **Queries** (default) | Carries data | Carries data | Reads/lookups that return a typed result | | **Commands** | Carries data | Execution acknowledgment only (`Executed` + optional `Error`) | Write actions where you only need confirmation it ran | In **Queries** mode the response body is deserialized into your `TResponse` type. In **Commands** mode the response is fire-and-await-ack: the broker returns `Executed = true` (or `false` with an `Error`) and **no response body**. Choose Queries when the caller needs data back, Commands when it only needs to know the action succeeded. ## Usage [#usage] ### Request and response [#request-and-response] Inject `IRequestClient` and call `GetResponse()`. The responder is an ordinary `IConsumer` that calls `context.RespondAsync(result)` — the transport routes the reply over the CQ reply channel. ```csharp title="Program.cs" builder.Services.AddMassTransit(x => { x.AddConsumer(); x.AddRequestClient(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // Default CqMode is Queries — request/response via KubeMQ query channel cfg.ReceiveEndpoint("reqresp-basic-query", e => { e.UseQueries(); }); }); }); // --- Message contracts --- public record LookupProduct(Guid RequestId, string Sku); public record ProductResult(Guid RequestId, string Sku, string Name, decimal Price, bool Found); // --- Responder --- public class ProductLookupConsumer(ILogger logger) : IConsumer { public async Task Consume(ConsumeContext context) { var msg = context.Message; var result = new ProductResult(msg.RequestId, msg.Sku, $"Widget-{msg.Sku}", 29.99m, Found: true); await context.RespondAsync(result); } } // --- Caller --- public class ProductCaller(IRequestClient client) { public async Task Lookup(string sku) { var response = await client.GetResponse( new LookupProduct(Guid.NewGuid(), sku)); return response.Message; } } ``` The CQ mode of the responder must match the requester. If the requester sends a Query, the receive endpoint must subscribe with `UseQueries()`; if it sends a Command, the endpoint must use `UseCommands()`. A mismatch surfaces as a request timeout — no responder is listening on the channel the sender used. ### Selecting the CQ mode [#selecting-the-cq-mode] There are three places to choose the mode, in increasing order of precedence. `KubeMQTransportOptions.DefaultCqMode` sets the global default and binds from `appsettings.json`. The shipped default is `Queries`. ```json title="appsettings.json" { "KubeMQ": { "Host": "localhost", "Port": 50000, "DefaultCqMode": "Queries" } } ``` `cfg.UseCommandsForRequestResponse()` flips the default for **all** request/response endpoints to Commands. ```csharp title="Program.cs" x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // All request/response endpoints use Commands instead of Queries cfg.UseCommandsForRequestResponse(); // Inherits the global Commands mode set above cfg.ReceiveEndpoint("reqresp-reserve-stock", e => { e.ConfigureKubeMQ(t => { }); }); }); ``` `e.UseQueries()` and `e.UseCommands()` override the mode for a single receive endpoint, regardless of the global default — so read endpoints return data while write endpoints stay fire-and-ack in the same bus. ```csharp title="Program.cs" x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // Read-only lookup: returns typed data cfg.ReceiveEndpoint("reqresp-price-query", e => e.UseQueries()); // Write action: fire-and-ack only cfg.ReceiveEndpoint("reqresp-reserve-stock", e => e.UseCommands()); }); ``` ### Command (fire-and-ack) [#command-fire-and-ack] A Commands-mode endpoint is fire-and-ack: the consumer processes the command and KubeMQ returns `Executed = true` on success. Commands do **not** carry a response body — for typed data back, use Queries. ```csharp title="Program.cs" builder.Services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); cfg.UseCommandsForRequestResponse(); cfg.ReceiveEndpoint("reqresp-basic-command", e => { e.UseCommands(); }); }); }); // --- Message contract --- public record ShipOrder(Guid OrderId, string Address, int ItemCount); // --- Command handler --- public class ShipOrderConsumer(ILogger logger) : IConsumer { public Task Consume(ConsumeContext context) { var msg = context.Message; // Process the command — KubeMQ acks Executed=true on successful return var tracking = $"TRK-{msg.OrderId.ToString()[..8].ToUpperInvariant()}"; logger.LogInformation("Command processed: TrackingNumber={Tracking}", tracking); return Task.CompletedTask; } } ``` ### Timeouts [#timeouts] Every CQ request carries a timeout. If no response arrives within the deadline, the request fails — a Query throws, and a Command reports it was not executed in time. The rider's `SendRequestAsync` sends a request over CQ directly; its `timeoutSeconds` parameter **defaults to 30**: ```csharp title="IKubeMQRider.cs" Task SendRequestAsync( TRequest request, string channelName, int timeoutSeconds = 30, CancellationToken cancellationToken = default) where TRequest : class where TResponse : class; ``` A request that exceeds the timeout faults; catch it and handle the timed-out case gracefully: ```csharp title="QueryTimeout.cs" var rider = KubeMQRiderAccessor.Current ?? throw new InvalidOperationException("KubeMQ rider not started."); // Slow query — 30s of work against a 3s timeout, expected to fault try { await rider.SendRequestAsync( new SlowQuery(Guid.NewGuid(), DelaySeconds: 30), "reqresp-query-timeout", timeoutSeconds: 3, stoppingToken); } catch (Exception ex) { logger.LogWarning("Query timed out as expected: {Message}", ex.Message); } ``` Set the request timeout above the responder's realistic worst-case processing time. A timeout shorter than the work the consumer performs faults the request even though the consumer eventually succeeds — and for a Query the work is wasted because the reply has nowhere to go. ### Competing responders [#competing-responders] To scale request handling, run **multiple processes** that subscribe the same receive endpoint (channel). KubeMQ distributes incoming requests across the group automatically — each request is handled by exactly one responder, so adding instances increases throughput. ```csharp title="Program.cs" builder.Services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // Consumer-group load balancing happens at the KubeMQ broker level // when multiple processes subscribe to the same channel. cfg.ReceiveEndpoint("reqresp-consumer-group", e => { e.UseQueries(); }); }); }); ``` ## Error semantics [#error-semantics] When a responder's consumer throws, KubeMQ does not silently drop the reply: the CQ response is returned with `Executed = false` and an `Error`, which the transport surfaces as a **failed request**. * In **Queries** mode, `GetResponse()` / `SendRequestAsync()` faults rather than returning a `Message`. * In **Commands** mode, the response reports `Executed = false` with the consumer's `Error`. The transport raises a `KubeMQTransportException` whose message reads `"Query request failed: {error}"` or `"Command request failed: {error}"`. A request timeout and a consumer error are different failures. A **timeout** means no response arrived in the allotted seconds (often a missing responder or a CQ mode mismatch). A **consumer error** means the responder ran, threw, and returned `Executed = false`. Both surface as a failed request, so inspect the error text and the responder logs to tell them apart. ## Errors and next steps [#errors-and-next-steps] * Diagnosing timeouts and `Query/Command request failed` errors → [Error Handling & DLQ](/integrations/masstransit/how-to/error-handling-dlq). * The `CqMode` enum, configurator API, and exception hierarchy → [reference](/integrations/masstransit/reference/configuration).