Configuration Reference

The complete configuration reference for the MassTransit.KubeMQ transport: package facts, the KubeMQTransportOptions POCO and its validation, appsettings.json binding, the kubemq:// host URI scheme, channel naming, header/tag mapping, and the observability instruments. Every value here is taken from the transport source. For the configurator method surface, see API; for exceptions, see Error codes.

Package

dotnet add package MassTransit.KubeMQ

The package depends on MassTransit with the version range [8.5.0,) and references the KubeMQ C# SDK (KubeMQ.Sdk). XML documentation is generated (GenerateDocumentationFile), so the configurator members surface as IntelliSense in your IDE.

KubeMQTransportOptions

KubeMQTransportOptions is the central configuration POCO. It is bindable to IOptions<T> and appsettings.json, and is passed via the configureOptions callback of UsingKubeMQ. The transport uses separate Host/Port properties for ergonomics and bridges them to the SDK's single Address string internally as {Host}:{Port}.

Validation

KubeMQTransportOptions.Validate() runs automatically during bus startup and throws KubeMQTransportConfigurationException for any invalid value. The exact messages are listed under Error codes.

Binding from appsettings.json

ConnectionTimeout and ReconnectTimeout are TimeSpan values, so they use the "hh:mm:ss" string format.

{
  "KubeMQ": {
    "Host": "kubemq-server.example.com",
    "Port": 50000,
    "AuthToken": "my-token",
    "UseTls": false,
    "PollTimeoutSeconds": 10,
    "MaxPollMessages": 64,
    "DefaultCqMode": "Queries",
    "ConnectionTimeout": "00:00:30",
    "ReconnectTimeout": "00:02:00"
  }
}

services.AddMassTransit(x =>
{
    x.UsingKubeMQ((ctx, cfg) =>
    {
        cfg.Host("kubemq-server.example.com", 50000);

        cfg.ReceiveEndpoint("orders", e =>
        {
            e.ConfigureKubeMQ(t => { });
        });
    }, options =>
    {
        options.MaxPollMessages = 64;
        options.PollTimeoutSeconds = 10;
    });
});

For the full walkthrough of every way to supply configuration, see the Configuration guide.

Host URI scheme

The Host(Uri) overload and send addresses use the kubemq:// scheme.

Connection URI

kubemq://host:port?authToken=<token>&tls=<true|false>

cfg.Host(new Uri("kubemq://kubemq-server:50000?authToken=my-token&tls=true"));

Send address

Send endpoints address a specific channel by path. The queue: shorthand resolves to the same channel:

kubemq://host:port/channel-name

// Full scheme
var endpoint = await bus.GetSendEndpoint(new Uri("kubemq://kubemq-server:50000/order-processing"));

// queue: shorthand
var endpoint = await bus.GetSendEndpoint(new Uri("queue:order-processing"));
await endpoint.Send(new SubmitOrder { OrderId = "123" });

Channel naming

KubeMQ channel names are derived from MassTransit endpoint and message-type names. The : character is replaced with . (so Namespace:Type becomes Namespace.Type), and IEndpointNameFormatter conventions (PascalCase, kebab-case, snake_case) are respected.

Header and tag mapping

MassTransit envelope headers are mapped to KubeMQ Tags with the MT- prefix. Custom user headers use the MT-Header-{name} form.

Trace context tags

W3C OpenTelemetry trace context is propagated automatically via two tags, integrating with MassTransit's ActivitySource ("MassTransit") for end-to-end distributed tracing:

Fault tags

When a message is faulted (after all retries are exhausted) and routed to {channel}_error, fault metadata is stored as tags:

Observability instruments

Metrics

The transport exposes instruments under the MassTransit.KubeMQ meter (version 1.0.0). Add the meter to your OpenTelemetry pipeline with AddMeter("MassTransit.KubeMQ") to collect them.

Health states

The transport maps the KubeMQ SDK ConnectionState to MassTransit health status. MassTransit registers these health checks automatically; expose them with app.MapHealthChecks("/health").

For wiring up health, traces, and metrics end-to-end, see the Observability guide.

See also