MassTransit Concepts

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<T> 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<T>, AddSaga<T>)

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.

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.

• Send → Queues. ISendEndpointProvider.Send<T>(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. IPublishEndpoint.Publish<T>() 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. When EventsStore is enabled (cfg.UseEventsStore() globally, or e.UseEventsStore() per endpoint), Publish<T>() 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. IRequestClient<T>.GetResponse<TResponse>() 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 below.

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:

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();

    x.UsingKubeMQ((ctx, cfg) =>
    {
        cfg.Host("localhost", 50000, h =>
        {
            h.AuthToken = "my-token";
        });

        cfg.ReceiveEndpoint("order-consumer", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    });
});

services.AddMassTransit(x =>
{
    x.UsingInMemory();

    x.AddKubeMQRider((ctx, k) =>
    {
        k.Host("localhost", 50000);
        k.ReceiveEndpoint("orders", e => { });
    });
});

The IKubeMQRider Surface

The rider exposes outbound operations alongside its receive transports. After the bus is started, you can resolve IKubeMQRider and use:

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));

CQ Modes (Commands vs Queries)

Request/response can run in either of two CQ modes, selected by the CqMode enum:

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

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:

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:

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

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.

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

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.

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

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:

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.

Endpoint Address Format

Send endpoints use the kubemq:// URI scheme. The path segment is the target channel name:

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:

// 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

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.

The application talks to MassTransit abstractions; the rider bridges them to the four native KubeMQ subsystems over gRPC.

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.

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Point the transport at it with cfg.Host("localhost", 50000).

Related topics