Overview

MassTransit's request/response abstraction — IRequestClient<T>.GetResponse<TResponse>() 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). 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

Member	Where	Purpose
`IRequestClient<T>.GetResponse<TResponse>(request)`	caller	Send a request and await the typed `Response<TResponse>`
`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<TReq, TResp>(request, channelName, timeoutSeconds = 30, ct)`	rider	Send a request over CQ directly, with an explicit timeout

Two CQ modes: Queries vs Commands

KubeMQ CQ has two modes, modeled by the CqMode enum in the transport:

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

Request and response

Inject IRequestClient<TRequest> and call GetResponse<TResponse>(). The responder is an ordinary IConsumer<TRequest> that calls context.RespondAsync(result) — the transport routes the reply over the CQ reply channel.

Program.cs

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<ProductLookupConsumer>();
    x.AddRequestClient<LookupProduct>();

    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<ProductLookupConsumer> logger) : IConsumer<LookupProduct>
{
    public async Task Consume(ConsumeContext<LookupProduct> 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<LookupProduct> client)
{
    public async Task<ProductResult> Lookup(string sku)
    {
        var response = await client.GetResponse<ProductResult>(
            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

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.

appsettings.json

{
  "KubeMQ": {
    "Host": "localhost",
    "Port": 50000,
    "DefaultCqMode": "Queries"
  }
}

cfg.UseCommandsForRequestResponse() flips the default for all request/response endpoints to Commands.

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.

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)

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.

Program.cs

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<ShipOrderConsumer>();

    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<ShipOrderConsumer> logger) : IConsumer<ShipOrder>
{
    public Task Consume(ConsumeContext<ShipOrder> 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

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<TRequest, TResponse> sends a request over CQ directly; its timeoutSeconds parameter defaults to 30:

IKubeMQRider.cs

Task<TResponse> SendRequestAsync<TRequest, TResponse>(
    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:

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<SlowQuery, SlowQueryResult>(
        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

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.

Program.cs

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<WorkerConsumer>();

    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

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<T>() / SendRequestAsync<T>() 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

Diagnosing timeouts and Query/Command request failed errors → Error Handling & DLQ.
The CqMode enum, configurator API, and exception hierarchy → reference.

Commands & Queries (Request/Response)

Overview

API surface

Two CQ modes: Queries vs Commands

Usage

Request and response

Selecting the CQ mode

Command (fire-and-ack)

Timeouts

Competing responders

Error semantics

Errors and next steps

Queues (Send)

Concepts

Overview

On this page