# API Reference (/integrations/masstransit/reference/api) 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](/integrations/masstransit/reference/configuration); for exceptions, see [Error codes](/integrations/masstransit/reference/error-codes). ## Registration entry points [#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` | | `AddKubeMQRider((ctx, k) => …)` | The bus already uses another transport and you want KubeMQ as a supplementary rider | ```csharp title="KubeMQBusRegistrationExtensions.cs" // KubeMQ as the primary transport (sets up an InMemory base bus + KubeMQ rider) public static void UsingKubeMQ( this IBusRegistrationConfigurator configurator, Action? configure = null); // Same, with IOptions binding for transport options public static void UsingKubeMQ( this IBusRegistrationConfigurator configurator, Action configure, Action? configureOptions = null); // KubeMQ as a supplementary rider on a bus that already has a base transport public static void AddKubeMQRider( this IBusRegistrationConfigurator configurator, Action? 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 [#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? configure = null)` | Configure the host connection by hostname/port | | `Host` | `void Host(Uri hostAddress, Action? 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? configure = null)` | Enable weighted priority queue channels | | `ReceiveEndpoint` | `void ReceiveEndpoint(string queueName, Action configureEndpoint)` | Declare and configure a receive endpoint | | `ConfigureEndpoints` | `void ConfigureEndpoints(IBusRegistrationContext context)` | **Not supported** — throws `KubeMQTransportConfigurationException`. Declare each endpoint explicitly with `ReceiveEndpoint(...)` instead | ```csharp title="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 [#host-configurator] The optional `Action` passed to `Host(...)` sets connection-level properties (all are write-only setters): `ClientId`, `AuthToken`, `UseTls`, `TlsCertFile`, `TlsKeyFile`, `TlsCaFile`, `ConnectionTimeout`, and `ReconnectTimeout`. ```csharp title="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 [#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 configure)` | Configure the EventsStore subscription start position | | `ConfigureKubeMQ` | `void ConfigureKubeMQ(Action configure)` | Configure transport-specific endpoint options (expiration, native DLQ) | | `UseVolatileEvents` | `void UseVolatileEvents()` | Subscribe to non-persistent Events (not EventsStore) for this endpoint | ```csharp title="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 [#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 [#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. ```csharp title="IPriorityQueueConfigurator.cs" // Default weights: high=3, normal=2, low=1 void SetWeights(int highWeight, int normalWeight, int lowWeight); ``` ```csharp title="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 [#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 | ```csharp title="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)](/integrations/masstransit/how-to/events-store) for full guidance on persistent publish and replay. ## CqMode enum [#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](/integrations/masstransit/how-to/commands-queries) for selecting and overriding the CQ mode. ## See also [#see-also]