# MassTransit Concepts (/integrations/masstransit/concepts) ## Transport-Agnostic by Design [#transport-agnostic-by-design] `MassTransit.KubeMQ` is a MassTransit transport: it adapts MassTransit's messaging abstractions to KubeMQ-native patterns. The application-level code you write against MassTransit does not change when KubeMQ becomes the transport. The following work identically regardless of transport: * Message contracts and `IConsumer` implementations * Consumer and saga logic * The middleware pipeline (filters, observers) * Serialization, retry and redelivery policies, and the outbox pattern * OpenTelemetry integration and dependency-injection registration (`AddConsumer`, `AddSaga`) What the transport *does* change is where messages physically travel: instead of RabbitMQ exchanges or an Azure Service Bus namespace, the same contracts are routed to KubeMQ's native Queues, Events, EventsStore, and CQ subsystems over gRPC. The sections below explain that routing. Because the contracts and consumers are transport-agnostic, switching to KubeMQ is mainly a `Program.cs` change — replace `UsingRabbitMq`/`UsingAzureServiceBus`/`UsingAmazonSqs` with `UsingKubeMQ`. ## Pattern Mapping [#pattern-mapping] Each MassTransit messaging verb maps to a specific KubeMQ pattern. The mapping determines delivery semantics (point-to-point vs. fan-out, durable vs. volatile) and which KubeMQ channel type backs the message. | MassTransit Pattern | KubeMQ Pattern | Delivery | Channel Type | | --------------------- | --------------------- | ------------------------------------- | ------------ | | **Send** | Queues | Point-to-point (exactly one consumer) | Queue | | **Publish** | Events | Fan-out (all active subscribers) | Events | | **Publish** (durable) | EventsStore | Fan-out with persistence | EventsStore | | **Request/Response** | Commands/Queries (CQ) | Native request-reply | CQ | * **Send → [Queues](/learn/queues).** `ISendEndpointProvider.Send(uri)` routes to a KubeMQ Queue. Exactly one consumer receives each message, messages persist until consumed, and the transport acks on success / nacks on failure. Delayed delivery (`ctx.Delay`) maps to `QueueMessage.DelaySeconds`, and TTL maps to `QueueMessage.ExpirationSeconds`. * **Publish → [Events](/learn/events).** `IPublishEndpoint.Publish()` routes to KubeMQ Events: fire-and-forget fan-out to all active subscribers, with no persistence or acknowledgment. If no subscriber is active, the message is silently dropped. Delayed delivery is **not** supported on Events and throws `KubeMQTransportConfigurationException`. * **Publish (durable) → [EventsStore](/learn/events-store).** When EventsStore is enabled (`cfg.UseEventsStore()` globally, or `e.UseEventsStore()` per endpoint), `Publish()` uses KubeMQ EventsStore instead. Messages are persisted and can be replayed from configurable start positions, so subscribers can join later and catch up. * **Request/Response → [CQ](/learn/rpc).** `IRequestClient.GetResponse()` maps to KubeMQ's native CQ pattern, with built-in `ReplyChannel` and timeout. No temporary reply queues are created (unlike RabbitMQ or SQS, which simulate request/response). Two modes are available, covered in [CQ modes](#cq-modes-commands-vs-queries) below. ## Rider-Based Architecture [#rider-based-architecture] MassTransit transports are normally created with a `Using...` bus factory. `MassTransit.KubeMQ` instead layers KubeMQ on top of an **InMemory base bus** plus a **KubeMQ rider** that owns the actual KubeMQ connections and receive transports. There are two entry points, both defined in `KubeMQBusRegistrationExtensions`: `UsingKubeMQ()` is the common case — KubeMQ is the transport for the whole application. Internally it calls `UsingInMemory()` to provide the `IBusControl` lifecycle, then attaches a KubeMQ rider that manages connections and receive endpoints. ```csharp title="Program.cs" services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((ctx, cfg) => { cfg.Host("localhost", 50000, h => { h.AuthToken = "my-token"; }); cfg.ReceiveEndpoint("order-consumer", e => { e.ConfigureKubeMQ(t => { }); }); }); }); ``` `AddKubeMQRider()` adds KubeMQ to a bus that already uses **another** transport (RabbitMQ, InMemory, etc.) as a supplementary transport. The rider is configured with `IKubeMQRiderConfigurator`. ```csharp title="Program.cs" services.AddMassTransit(x => { x.UsingInMemory(); x.AddKubeMQRider((ctx, k) => { k.Host("localhost", 50000); k.ReceiveEndpoint("orders", e => { }); }); }); ``` ### The `IKubeMQRider` Surface [#the-ikubemqrider-surface] The rider exposes outbound operations alongside its receive transports. After the bus is started, you can resolve `IKubeMQRider` and use: | Method | Purpose | | --------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------ | | `GetSendEndpoint(Uri, ct)` | Resolves a send endpoint for `kubemq://host:port/channel` queue addresses (Send → Queues). | | `PublishEventAsync(message, channelName, ct)` | Publishes a volatile event to a KubeMQ Events channel (fan-out). | | `SendRequestAsync(request, channelName, timeoutSeconds, ct)` | Sends a request via KubeMQ Queries (or Commands) and returns the deserialized response. | ```csharp title="Using the rider directly" var rider = KubeMQRiderAccessor.Current ?? throw new InvalidOperationException("KubeMQ rider not started."); var endpoint = await rider.GetSendEndpoint( new Uri("kubemq://localhost:50000/competing-consumers-queue")); await endpoint.Send(new WorkItem("WORK-001", "Process work item #1", 1)); ``` `SendRequestAsync` defaults to a 30-second timeout (`timeoutSeconds = 30`). Whether it sends a Query or a Command is controlled by the configured [CQ mode](#cq-modes-commands-vs-queries). ## CQ Modes (Commands vs Queries) [#cq-modes-commands-vs-queries] Request/response can run in either of two CQ modes, selected by the `CqMode` enum: | Mode | Value | Semantics | | ---------- | ------------- | ---------------------------------------------------------------------------------------- | | `Queries` | `0` (default) | Request carries data, response carries data. | | `Commands` | `1` | Request carries data, response is an execution acknowledgment only (fire-and-await-ack). | The default is `Queries`. Override it globally with `cfg.UseCommandsForRequestResponse()`, or per endpoint with `e.UseQueries()` / `e.UseCommands()`. On a consumer error, the response includes `Executed = false` and an `Error` message. ## Connection Lifecycle and Health [#connection-lifecycle-and-health] The KubeMQ SDK manages its own connection through a `ConnectionState` state machine. The transport monitors `IKubeMQClient.StateChanged` events and surfaces each state to MassTransit's health pipeline. *The KubeMQ SDK connection state machine; the transport maps each state onto a MassTransit health status.* Each KubeMQ state maps to a MassTransit health status: | KubeMQ State | MassTransit Health | Meaning | | -------------- | ------------------ | ---------------------------------------- | | `Ready` | **Healthy** | Connection is active and operational | | `Connecting` | **Degraded** | Initial connection in progress | | `Reconnecting` | **Degraded** | Lost connection, attempting to reconnect | | `Closed` | **Unhealthy** | Connection permanently closed | | `Idle` | **Unhealthy** | Not connected | MassTransit registers these health checks automatically; expose them with `app.MapHealthChecks("/health")`. For queue receive transports, when a `QueueDownstreamReceiver` stream breaks the transport catches the connection exception, disposes the broken receiver, waits one second, and creates a new receiver on the next poll iteration. Unsent messages during a disconnection are handled by MassTransit's retry pipeline, not by the transport. Tune the timeouts on the host: ```csharp title="Reconnection configuration" cfg.Host("kubemq-server", 50000, h => { h.ConnectionTimeout = TimeSpan.FromSeconds(30); // Initial connection h.ReconnectTimeout = TimeSpan.FromSeconds(120); // Max wait for reconnection }); ``` ## Channel Naming Conventions [#channel-naming-conventions] KubeMQ channel names are derived from MassTransit endpoint and message-type names. There is no exchange or topology layer — channels are created automatically on first use. | Context | Pattern | Example | | ----------------- | ------------------------------ | --------------------------------------------- | | Send endpoint | Queue name from URI path | `queue:order-processing` → `order-processing` | | Publish endpoint | Message type full name | `MyApp.Events.OrderSubmitted` | | Consumer endpoint | Endpoint name (auto or manual) | `order-consumer` | | Error channel | `{channel}_error` | `order-processing_error` | | Skipped channel | `{channel}_skipped` | `order-processing_skipped` | | Priority high | `{channel}_high` | `order-processing_high` | | Priority normal | `{channel}_normal` | `order-processing_normal` | | Priority low | `{channel}_low` | `order-processing_low` | | Consumer group | Same as endpoint name | `order-consumer` | The `:` character in type names is replaced with `.` — for example, `Namespace:Type` becomes `Namespace.Type`. `IEndpointNameFormatter` conventions (PascalCase, kebab-case, snake\_case) are respected. Faulted messages route to the `_error` channel after all configured retries are exhausted; skipped messages route to the `_skipped` channel. ## Header and Envelope Mapping [#header-and-envelope-mapping] The MassTransit message envelope is preserved across KubeMQ by mapping headers to KubeMQ **Tags** with the `MT-` prefix. Custom user headers use the `MT-Header-{name}` form. | MassTransit Header | KubeMQ Tag | Description | | -------------------- | ----------------------- | --------------------------------------- | | `MessageId` | `MT-MessageId` | Unique message identifier | | `CorrelationId` | `MT-CorrelationId` | Correlation for related messages | | `ConversationId` | `MT-ConversationId` | Conversation tracking | | `RequestId` | `MT-RequestId` | Request/response correlation | | `InitiatorId` | `MT-InitiatorId` | Message initiator | | `SourceAddress` | `MT-SourceAddress` | Sending endpoint address | | `DestinationAddress` | `MT-DestinationAddress` | Target endpoint address | | `ResponseAddress` | `MT-ResponseAddress` | Response endpoint address | | `FaultAddress` | `MT-FaultAddress` | Fault handling address | | `ContentType` | `MT-ContentType` | Serialization content type | | `SentTime` | `MT-SentTime` | ISO 8601 send timestamp | | `ExpirationTime` | `MT-ExpirationTime` | ISO 8601 expiration (from TTL) | | `MessageType` | `MT-MessageType` | Supported message types (`;`-separated) | | Custom headers | `MT-Header-{name}` | User-defined headers | W3C distributed-trace context is propagated alongside these headers so OpenTelemetry traces span service boundaries: * `MT-TraceParent` — W3C `traceparent` header * `MT-TraceState` — W3C `tracestate` header When a message is faulted, additional tags carry the failure metadata: `MT-Fault-ExceptionType`, `MT-Fault-Message`, `MT-Fault-StackTrace`, `MT-Fault-Timestamp`, and `MT-Fault-RetryCount`. ## Consumer Groups (Competing Consumers) [#consumer-groups-competing-consumers] Competing consumers in MassTransit map directly onto KubeMQ's consumer-group model: **the consumer group is the endpoint name**. When multiple consumer instances share the same endpoint name, KubeMQ load-balances messages across the group so each message is delivered to exactly one member. The simplest demonstration declares two receive endpoints on the same channel — KubeMQ treats them as one group and distributes work between them: ```csharp title="Competing consumers on a shared channel" x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // Two endpoints on the same channel form one consumer group. // KubeMQ load-balances messages across members, // so each message is consumed exactly once. cfg.ReceiveEndpoint("competing-consumers-queue", e => { e.ConfigureKubeMQ(t => { }); }); cfg.ReceiveEndpoint("competing-consumers-queue", e => { e.ConfigureKubeMQ(t => { }); }); }); ``` This applies to both Send → Queues and Request/Response → CQ: the same endpoint name lets multiple instances share queue load or act as competing responders. To scale horizontally, run multiple application instances that all bind the same endpoint name. Competing consumers must use the **same** endpoint name to form one group. A name mismatch creates separate groups, and each group receives its own copy of every message. ## Endpoint Address Format [#endpoint-address-format] Send endpoints use the `kubemq://` URI scheme. The path segment is the target channel name: ```text kubemq://host:port/channel-name ``` The `queue:` shorthand resolves to the same target on the configured host and is the recommended form inside the bus: ```csharp title="Resolving a send endpoint" // Shorthand form — resolved against the configured KubeMQ host/port var endpoint = await bus.GetSendEndpoint(new Uri("queue:order-processing")); // Fully-qualified form — required when using IKubeMQRider.GetSendEndpoint directly var direct = await rider.GetSendEndpoint( new Uri("kubemq://localhost:50000/order-processing")); ``` ## Bus → Rider → Broker Flow [#bus--rider--broker-flow] The application talks to MassTransit abstractions as usual. The KubeMQ rider bridges those abstractions to the KubeMQ broker over gRPC on port `50000`, where the broker dispatches to the four native subsystems that back the [pattern mapping](#pattern-mapping). *The application talks to MassTransit abstractions; the rider bridges them to the four native KubeMQ subsystems over gRPC.* ## Running a Broker Locally [#running-a-broker-locally] `MassTransit.KubeMQ` connects to the KubeMQ broker over gRPC on port `50000`. Port `9090` is the shared HTTP server that hosts the REST, CloudEvents, MCP, and A2A connectors — it is not required by the MassTransit transport, but the standard image exposes both. Point the transport at it with `cfg.Host("localhost", 50000)`. ## Related topics [#related-topics]