# Configuration Reference (/integrations/masstransit/reference/configuration)



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](/integrations/masstransit/reference/api); for exceptions, see [Error codes](/integrations/masstransit/reference/error-codes).

## Package [#package]

| Fact                   | Value                                     |
| ---------------------- | ----------------------------------------- |
| NuGet PackageId        | `MassTransit.KubeMQ`                      |
| Version                | `1.0.0`                                   |
| Target framework       | `net8.0`                                  |
| MassTransit dependency | `>= 8.5.0`                                |
| License                | Apache-2.0                                |
| Repository             | `github.com/kubemq-io/kubemq-masstransit` |

```bash title="terminal"
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]

`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}`.

<TypeTable
  type="{
  Host: { type: &#x22;string&#x22;, default: '&#x22;localhost&#x22;', description: &#x22;KubeMQ server hostname. Validated non-empty.&#x22; },
  Port: { type: &#x22;int&#x22;, default: &#x22;50000&#x22;, description: &#x22;KubeMQ gRPC port. Validated 1–65535.&#x22; },
  ClientId: { type: &#x22;string?&#x22;, default: &#x22;null&#x22;, description: &#x22;Client identifier for this connection. Auto-generated if null.&#x22; },
  AuthToken: { type: &#x22;string?&#x22;, default: &#x22;null&#x22;, description: &#x22;Authentication token.&#x22; },
  UseTls: { type: &#x22;bool&#x22;, default: &#x22;false&#x22;, description: &#x22;Enable TLS encryption.&#x22; },
  TlsCertFile: { type: &#x22;string?&#x22;, default: &#x22;null&#x22;, description: &#x22;TLS client certificate file path (PEM). Required when TLS is enabled.&#x22; },
  TlsKeyFile: { type: &#x22;string?&#x22;, default: &#x22;null&#x22;, description: &#x22;TLS client private key file path (PEM).&#x22; },
  TlsCaFile: { type: &#x22;string?&#x22;, default: &#x22;null&#x22;, description: &#x22;TLS CA certificate file path (PEM).&#x22; },
  PollTimeoutSeconds: { type: &#x22;int&#x22;, default: &#x22;5&#x22;, description: &#x22;Queue long-poll wait timeout. Validated 1–3600.&#x22; },
  MaxPollMessages: { type: &#x22;int&#x22;, default: &#x22;32&#x22;, description: &#x22;Maximum messages per poll batch. Validated 1–1024.&#x22; },
  DefaultCqMode: { type: &#x22;CqMode&#x22;, default: &#x22;Queries&#x22;, description: &#x22;Global CQ mode for request/response.&#x22; },
  ConnectionTimeout: { type: &#x22;TimeSpan&#x22;, default: &#x22;10s&#x22;, description: &#x22;Initial connection timeout. Must be positive.&#x22; },
  ReconnectTimeout: { type: &#x22;TimeSpan&#x22;, default: &#x22;60s&#x22;, description: &#x22;Reconnection wait timeout. Must be positive.&#x22; },
}"
/>

### Validation [#validation]

`KubeMQTransportOptions.Validate()` runs automatically during bus startup and throws `KubeMQTransportConfigurationException` for any invalid value. The exact messages are listed under [Error codes](/integrations/masstransit/reference/error-codes#validation-errors).

| Property             | Rule                       |
| -------------------- | -------------------------- |
| `Host`               | Non-empty / non-whitespace |
| `Port`               | `1`–`65535`                |
| `PollTimeoutSeconds` | `1`–`3600`                 |
| `MaxPollMessages`    | `1`–`1024`                 |
| `ConnectionTimeout`  | `> TimeSpan.Zero`          |
| `ReconnectTimeout`   | `> TimeSpan.Zero`          |

### Binding from appsettings.json [#binding-from-appsettingsjson]

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

```json title="appsettings.json"
{
  "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"
  }
}
```

```csharp title="Program.cs"
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](/integrations/masstransit/how-to/configuration).

## Host URI scheme [#host-uri-scheme]

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

### Connection URI [#connection-uri]

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

| Query parameter | Meaning                                 |
| --------------- | --------------------------------------- |
| `authToken`     | Authentication token                    |
| `tls`           | `true` or `false` to enable/disable TLS |

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

### Send address [#send-address]

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

```text
kubemq://host:port/channel-name
```

```csharp title="Program.cs"
// 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 [#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.

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

## Header and tag mapping [#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.

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

### Trace context tags [#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:

| Tag              | Maps to                  |
| ---------------- | ------------------------ |
| `MT-TraceParent` | W3C `traceparent` header |
| `MT-TraceState`  | W3C `tracestate` header  |

### Fault tags [#fault-tags]

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

| Tag                      | Description                              |
| ------------------------ | ---------------------------------------- |
| `MT-Fault-ExceptionType` | Exception type name                      |
| `MT-Fault-Message`       | Exception message                        |
| `MT-Fault-StackTrace`    | Exception stack trace                    |
| `MT-Fault-Timestamp`     | When the fault occurred                  |
| `MT-Fault-RetryCount`    | Number of retry attempts before faulting |

## Observability instruments [#observability-instruments]

### Metrics [#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.

| Instrument                          | Type      | Unit     | Description                       |
| ----------------------------------- | --------- | -------- | --------------------------------- |
| `kubemq.transport.poll.duration`    | Histogram | ms       | Duration of queue poll operations |
| `kubemq.transport.poll.messages`    | Histogram | messages | Messages received per poll batch  |
| `kubemq.transport.poll.empty`       | Counter   | polls    | Count of empty poll responses     |
| `kubemq.transport.send.duration`    | Histogram | ms       | Duration of queue send operations |
| `kubemq.transport.publish.duration` | Histogram | ms       | Duration of publish operations    |
| `kubemq.transport.errors`           | Counter   | errors   | Transport-level error count       |

### Health states [#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")`.

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

<Callout type="info">
  The `KubeMQConnectionContextSupervisor` exposes the same state via `IsReady` (`Ready`), `IsDegraded` (`Connecting` or `Reconnecting`), and `IsUnhealthy` (`Closed` or `Idle`) for programmatic checks.
</Callout>

For wiring up health, traces, and metrics end-to-end, see the [Observability guide](/integrations/masstransit/how-to/observability).

## See also [#see-also]

<Cards>
  <Card title="API" href="/integrations/masstransit/reference/api" description="Registration entry points, configurator interfaces, EventsStore positions, and the CqMode enum." />

  <Card title="Error codes" href="/integrations/masstransit/reference/error-codes" description="Exception hierarchy, validation messages, and CQ failure triggers." />
</Cards>