The configurator API surface for the MassTransit.KubeMQ transport: the registration entry points and every configurator interface, from the bus factory down to per-endpoint transport options, plus the EventsStore subscription positions and the CqMode enum. For options values and validation, see Configuration; for exceptions, see Error codes.

Registration entry points

The transport registers inside AddMassTransit(...). Three extension methods are defined on IBusRegistrationConfigurator.

Method	Use it when
`UsingKubeMQ((ctx, cfg) => …)`	KubeMQ is the application's primary transport
`UsingKubeMQ(configure, options => …)`	Primary transport, with `KubeMQTransportOptions` bound from `appsettings.json` / `IOptions<T>`
`AddKubeMQRider((ctx, k) => …)`	The bus already uses another transport and you want KubeMQ as a supplementary rider

KubeMQBusRegistrationExtensions.cs

// KubeMQ as the primary transport (sets up an InMemory base bus + KubeMQ rider)
public static void UsingKubeMQ(
    this IBusRegistrationConfigurator configurator,
    Action<IBusRegistrationContext, IKubeMQBusFactoryConfigurator>? configure = null);

// Same, with IOptions binding for transport options
public static void UsingKubeMQ(
    this IBusRegistrationConfigurator configurator,
    Action<IBusRegistrationContext, IKubeMQBusFactoryConfigurator> configure,
    Action<KubeMQTransportOptions>? configureOptions = null);

// KubeMQ as a supplementary rider on a bus that already has a base transport
public static void AddKubeMQRider(
    this IBusRegistrationConfigurator configurator,
    Action<IRiderRegistrationContext, IKubeMQRiderConfigurator>? configure = null);

UsingKubeMQ internally calls UsingInMemory() to provide the IBusControl lifecycle, then attaches a KubeMQ rider that manages connections and receive transports. AddKubeMQRider adds only the rider, leaving your existing base transport intact.

Bus-factory configurator

IKubeMQBusFactoryConfigurator is the cfg argument inside UsingKubeMQ((ctx, cfg) => …). It configures the host connection and bus-wide behavior.

Member	Signature	Purpose
`Host`	`void Host(string host, int port = 50000, Action<IKubeMQHostConfigurator>? configure = null)`	Configure the host connection by hostname/port
`Host`	`void Host(Uri hostAddress, Action<IKubeMQHostConfigurator>? configure = null)`	Configure the host from a `kubemq://` URI
`UseCommandsForRequestResponse`	`void UseCommandsForRequestResponse()`	Set the global CQ mode to Commands for request/response
`UseEventsStore`	`void UseEventsStore()`	Enable EventsStore for all publish endpoints (global)
`UsePriorityQueues`	`void UsePriorityQueues(Action<IPriorityQueueConfigurator>? configure = null)`	Enable weighted priority queue channels
`ReceiveEndpoint`	`void ReceiveEndpoint(string queueName, Action<IKubeMQReceiveEndpointConfigurator> configureEndpoint)`	Declare and configure a receive endpoint
`ConfigureEndpoints`	`void ConfigureEndpoints(IBusRegistrationContext context)`	Not supported — throws `KubeMQTransportConfigurationException`. Declare each endpoint explicitly with `ReceiveEndpoint(...)` instead

Program.cs

x.UsingKubeMQ((ctx, cfg) =>
{
    cfg.Host("kubemq-server.example.com", 50000, h =>
    {
        h.AuthToken = "my-secret-token";
        h.UseTls = true;
        h.ConnectionTimeout = TimeSpan.FromSeconds(30);
    });

    cfg.UseCommandsForRequestResponse();   // global CQ mode -> Commands
    cfg.UseEventsStore();                   // global persistent publish
    cfg.UsePriorityQueues();                // default weights 3:2:1

    // Declare each receive endpoint explicitly — ConfigureEndpoints() is not supported.
    cfg.ReceiveEndpoint("orders", e =>
    {
        e.ConfigureKubeMQ(t => { });
    });
});

Host configurator

The optional Action<IKubeMQHostConfigurator> passed to Host(...) sets connection-level properties (all are write-only setters): ClientId, AuthToken, UseTls, TlsCertFile, TlsKeyFile, TlsCaFile, ConnectionTimeout, and ReconnectTimeout.

Program.cs

cfg.Host("kubemq-server.example.com", 50000, h =>
{
    h.ClientId = "order-service";
    h.AuthToken = "my-secret-token";
    h.UseTls = true;
    h.TlsCertFile = "/certs/client.pem";
    h.TlsKeyFile = "/certs/client-key.pem";
    h.TlsCaFile = "/certs/ca.pem";
    h.ConnectionTimeout = TimeSpan.FromSeconds(30);
    h.ReconnectTimeout = TimeSpan.FromSeconds(120);
});

Receive-endpoint configurator

IKubeMQReceiveEndpointConfigurator is the e argument inside ReceiveEndpoint(name, e => …). It tunes polling, selects the messaging pattern, and overrides CQ mode for the endpoint.

Member	Signature	Purpose
`PollTimeoutSeconds`	`int { set; }`	Override queue long-poll timeout for this endpoint
`MaxPollMessages`	`int { set; }`	Override max messages per poll batch for this endpoint
`UseCommands`	`void UseCommands()`	Force CQ mode to Commands for this endpoint only
`UseQueries`	`void UseQueries()`	Force CQ mode to Queries for this endpoint only
`UseEventsStore`	`void UseEventsStore()`	Enable EventsStore for this endpoint
`UseEventsStoreSubscription`	`void UseEventsStoreSubscription(Action<IEventsStoreSubscriptionConfigurator> configure)`	Configure the EventsStore subscription start position
`ConfigureKubeMQ`	`void ConfigureKubeMQ(Action<IKubeMQEndpointTransportConfigurator> configure)`	Configure transport-specific endpoint options (expiration, native DLQ)
`UseVolatileEvents`	`void UseVolatileEvents()`	Subscribe to non-persistent Events (not EventsStore) for this endpoint

Program.cs

cfg.ReceiveEndpoint("order-processing", e =>
{
    e.PollTimeoutSeconds = 10;
    e.MaxPollMessages = 64;
    e.UseQueries();   // this endpoint uses Queries even if the global mode is Commands

    e.ConfigureKubeMQ(k =>
    {
        k.ExpirationSeconds = 3600;        // messages expire after 1 hour
        k.UseNativeDlq(5, "orders-dlq");   // dead-letter after 5 receive attempts
    });
});

Endpoint transport configurator

IKubeMQEndpointTransportConfigurator is the k argument inside ConfigureKubeMQ(k => …). It exposes transport-level knobs that don't have a MassTransit-agnostic equivalent.

Member	Signature	Purpose
`ExpirationSeconds`	`int { set; }`	Message expiration (TTL) in seconds
`UseNativeDlq`	`void UseNativeDlq(int maxReceiveCount, string? dlqChannel = null)`	Route to KubeMQ's native dead-letter queue after `maxReceiveCount` deliveries; optional explicit DLQ channel
`UseEventsStore`	`void UseEventsStore()`	Enable EventsStore for this endpoint via the transport configurator

Priority-queue configurator

IPriorityQueueConfigurator is the optional argument to UsePriorityQueues(...). It sets the weighted round-robin polling weights for the _high / _normal / _low channels. Higher weights poll more messages per cycle.

IPriorityQueueConfigurator.cs

// Default weights: high=3, normal=2, low=1
void SetWeights(int highWeight, int normalWeight, int lowWeight);

Program.cs

cfg.UsePriorityQueues(p =>
{
    p.SetWeights(highWeight: 5, normalWeight: 3, lowWeight: 1);
});

Priority queues create three channels per endpoint — {queue}_high, {queue}_normal, {queue}_low — and messages are polled from them via weighted round-robin based on these values.

EventsStore subscription positions

IEventsStoreSubscriptionConfigurator is the opts argument inside UseEventsStoreSubscription(opts => …). It controls where a subscriber begins reading from a persisted EventsStore channel. If no method is called, the subscription starts from new (only messages published after the subscription starts).

Method	Signature	Behavior
(none — default)	—	StartFromNew — receive only messages published after subscribing
`StartFromFirst`	`void StartFromFirst()`	Replay from the first stored message, then continue with new ones
`StartFromLast`	`void StartFromLast()`	Start from the most recent stored message, then continue
`StartFromSequence`	`void StartFromSequence(long sequence)`	Start from a specific sequence number onward
`StartFromTime`	`void StartFromTime(DateTimeOffset time)`	Start from messages stored at or after a point in time
`StartFromTimeDelta`	`void StartFromTimeDelta(int seconds)`	Start from a relative offset in seconds before now

Program.cs

cfg.ReceiveEndpoint("audit-events", e =>
{
    e.UseEventsStore();
    e.UseEventsStoreSubscription(opts =>
    {
        opts.StartFromFirst();                                 // full replay
        // opts.StartFromSequence(42);                         // from a checkpoint
        // opts.StartFromTime(DateTimeOffset.UtcNow.AddHours(-1));
        // opts.StartFromTimeDelta(3600);                      // last hour
    });
});

See Events Store (Durable Publish) for full guidance on persistent publish and replay.

CqMode enum

The DefaultCqMode option and the per-endpoint UseCommands() / UseQueries() overrides resolve to the CqMode enum, which selects whether request/response runs over KubeMQ Queries or Commands.

Member	Value	Semantics
`Queries`	`0`	Default. Request carries data, response carries data.
`Commands`	`1`	Request carries data, response is an execution acknowledgment only.

See Commands & Queries for selecting and overriding the CQ mode.

API Reference

Registration entry points

Bus-factory configurator

Host configurator

Receive-endpoint configurator

Endpoint transport configurator

Priority-queue configurator

EventsStore subscription positions

CqMode enum

See also

Configuration

Error codes

On this page