# Getting Started with MassTransit (/integrations/masstransit/tutorials/getting-started) The `MassTransit.KubeMQ` transport lets you use [MassTransit](https://masstransit.io) — the .NET distributed-application framework — with KubeMQ as the underlying message broker. It is a native gRPC client: MassTransit `Publish` maps to KubeMQ Events, `Send` maps to KubeMQ Queues, and request/response maps to KubeMQ Commands/Queries. This guide installs the transport and runs two end-to-end examples against a local KubeMQ broker: a fire-and-forget publish/subscribe flow and a point-to-point queue send. ## Setup & examples [#setup--examples] ### Prerequisites [#prerequisites] You need the following installed: | Requirement | Version | | ------------- | ------------------------------------------ | | .NET SDK | 8.0+ | | Docker | Any recent version | | KubeMQ broker | Running locally (started in the next step) | The transport connects to KubeMQ over gRPC. `KubeMQTransportOptions` defaults to `Host = "localhost"` and `Port = 50000`, and the examples below assume a **KubeMQ broker running at `localhost:50000`**. No other ports are required. ### Start KubeMQ [#start-kubemq] Run KubeMQ in Docker, exposing only the gRPC port: Port `50000` is the KubeMQ gRPC port. Because the MassTransit transport is a **native gRPC client** (not an HTTP connector), there is no connector enable flag to set — unlike HTTP-based integrations, you do not expose port `9090` or set any `CONNECTORS*_ENABLE` environment variable. Exposing the gRPC port is all that is needed; this matches the `Host`/`Port` defaults documented in the transport's configuration reference. If you already have KubeMQ running on a different host or port, you will adjust `cfg.Host(...)` in the code below. See the host configuration note at the end of this guide. ### Install the Transport [#install-the-transport] Create a console project and add the NuGet package: ```bash dotnet new console -n KubeMQ.Quickstart cd KubeMQ.Quickstart dotnet add package MassTransit.KubeMQ ``` The `MassTransit.KubeMQ` package targets `net8.0` and pulls in MassTransit `8.5.0` or newer as a transitive dependency. In your code, the transport requires two namespaces — the MassTransit core API and the KubeMQ transport namespace: ```csharp title="Program.cs" using MassTransit; using MassTransit.KubeMQTransport; ``` `MassTransit` provides `AddMassTransit`, `IConsumer`, and the bus APIs. `MassTransit.KubeMQTransport` provides the KubeMQ-specific entry points: `UsingKubeMQ`, the `UseVolatileEvents()` endpoint configurator, and `KubeMQRiderAccessor`. ### First Publish/Subscribe (Events) [#first-publishsubscribe-events] This first example wires up a consumer and publishes a volatile event to it. Replace the contents of `Program.cs`: ```csharp title="Program.cs" using MassTransit; using MassTransit.KubeMQTransport; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; var builder = Host.CreateApplicationBuilder(args); builder.Services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); // Configure a receive endpoint for volatile events (fire-and-forget, non-persistent) cfg.ReceiveEndpoint("events-basic-pubsub", e => { e.UseVolatileEvents(); }); }); }); var host = builder.Build(); await host.RunAsync(); // --- Message type --- public record OrderPlaced(string OrderId, string Product, decimal Amount, DateTime PlacedAt); // --- Consumer --- public class OrderPlacedConsumer : IConsumer { private readonly ILogger _logger; public OrderPlacedConsumer(ILogger logger) { _logger = logger; } public Task Consume(ConsumeContext context) { var msg = context.Message; _logger.LogInformation( "Received volatile event: OrderId={OrderId}, Product={Product}, Amount={Amount:C}", msg.OrderId, msg.Product, msg.Amount); return Task.CompletedTask; } } ``` `cfg.Host("localhost", 50000)` points the transport at your local broker. `cfg.ReceiveEndpoint("events-basic-pubsub", ...)` declares a receive endpoint, and `e.UseVolatileEvents()` configures that endpoint to subscribe to **non-persistent KubeMQ Events** rather than EventsStore or Queues. **About `UseVolatileEvents()`.** `IKubeMQReceiveEndpointConfigurator.UseVolatileEvents()` subscribes the endpoint to KubeMQ's fire-and-forget Events pattern. In MassTransit terms, this is the `Publish` side of pub/sub: every active subscriber receives the message, there is no persistence, and there is no acknowledgment. You publish through the bus with `bus.Publish(...)`, or directly through the KubeMQ rider with `KubeMQRiderAccessor.Current.PublishEventAsync(...)`: ```csharp title="Publishing a volatile event" var rider = KubeMQRiderAccessor.Current ?? throw new InvalidOperationException("KubeMQ rider not started."); await rider.PublishEventAsync( new OrderPlaced("ORD-0001", "Widget", 19.99m, DateTime.UtcNow), "events-basic-pubsub"); ``` Because Events are fire-and-forget, **subscribers must be active before you publish**. If no subscriber is listening on the channel when a message is sent, KubeMQ silently drops it — the message is not stored or replayed. If you need durable, replayable fan-out instead, use EventsStore. The example below delays publishing for a few seconds so the bus and subscription have time to start. ### Send to a Queue [#send-to-a-queue] The publish/subscribe flow above is fan-out and ephemeral. For point-to-point work where each message must be processed by exactly one consumer and **persisted in the queue until consumed**, use `Send` instead. MassTransit `Send` maps to KubeMQ Queues. Get a send endpoint from the bus using the `queue:` address and call `Send(...)`: ```csharp title="Sending to a queue" // Obtain a send endpoint for the "order-processing" queue var endpoint = await bus.GetSendEndpoint(new Uri("queue:order-processing")); // Each message is delivered to exactly one consumer and persists until consumed await endpoint.Send(new SubmitOrder { OrderId = "123" }); ``` The matching consumer implements `IConsumer`: ```csharp title="Queue consumer" public class OrderConsumer : IConsumer { public async Task Consume(ConsumeContext context) { // Process order -- message auto-ack'd on success } } ``` To receive from a queue, declare a receive endpoint with the queue name. Unlike Events, a queue endpoint does not call `UseVolatileEvents()`; the default endpoint behavior is a KubeMQ Queue consumer: ```csharp title="Registering a queue consumer" builder.Services.AddMassTransit(x => { x.AddConsumer(); x.UsingKubeMQ((context, cfg) => { cfg.Host("localhost", 50000); cfg.ReceiveEndpoint("order-processing", e => { e.ConfigureKubeMQ(t => { }); }); }); }); ``` Queue messages are acknowledgment-based: a message is ack'd after the consumer succeeds and nack'd on failure. Because they persist until consumed, the consumer does **not** need to be running at the moment you send — a key difference from volatile Events. ### Run It [#run-it] Run the project: ```bash dotnet run ``` When the publish/subscribe example runs, the bus starts, the `events-basic-pubsub` subscription comes up, and the background publisher then sends `OrderPlaced` events. The consumer logs each received event: ```text info: KubeMQ.Quickstart.OrderPlacedConsumer[0] Received volatile event: OrderId=ORD-0001, Product=Product-1, Amount=$19.99 info: KubeMQ.Quickstart.OrderPlacedConsumer[0] Received volatile event: OrderId=ORD-0002, Product=Product-2, Amount=$39.98 ``` If you see no `Received` lines, the most common cause is publishing before the subscriber is active — see the warning in the publish/subscribe step. Verify the broker is reachable on `localhost:50000` and that the publisher delays briefly after startup. ### Host Configuration Variations [#host-configuration-variations] This guide uses the simplest form, `cfg.Host("localhost", 50000)`. The transport supports several other ways to configure the broker connection: * **Host/port with options** — pass a configuration delegate to set `AuthToken`, `UseTls`, TLS file paths, `ConnectionTimeout`, and `ReconnectTimeout`. * **`kubemq://` URI** — e.g. `cfg.Host(new Uri("kubemq://kubemq-server:50000?authToken=my-token&tls=true"))`. * **`appsettings.json` via `IOptions`** — bind `Host`, `Port`, `AuthToken`, `UseTls`, `PollTimeoutSeconds`, `MaxPollMessages`, and timeouts from configuration. See the configuration how-to for the full set of options and examples. ### Next Steps [#next-steps]