Getting Started with MassTransit

Prerequisites

You need the following installed:

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

Run KubeMQ in Docker, exposing only the gRPC port:

docker run -d \  --name kubemq \  -p 50000:50000 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

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.

Install the Transport

Create a console project and add the NuGet package:

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:

using MassTransit;
using MassTransit.KubeMQTransport;

MassTransit provides AddMassTransit, IConsumer<T>, and the bus APIs. MassTransit.KubeMQTransport provides the KubeMQ-specific entry points: UsingKubeMQ, the UseVolatileEvents() endpoint configurator, and KubeMQRiderAccessor.

First Publish/Subscribe (Events)

This first example wires up a consumer and publishes a volatile event to it. Replace the contents of 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<OrderPlacedConsumer>();
    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<OrderPlaced>
{
    private readonly ILogger<OrderPlacedConsumer> _logger;

    public OrderPlacedConsumer(ILogger<OrderPlacedConsumer> logger)
    {
        _logger = logger;
    }

    public Task Consume(ConsumeContext<OrderPlaced> 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<T>(...), or directly through the KubeMQ rider with KubeMQRiderAccessor.Current.PublishEventAsync(...):

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");

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(...):

// 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<SubmitOrder>:

public class OrderConsumer : IConsumer<SubmitOrder>
{
    public async Task Consume(ConsumeContext<SubmitOrder> 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:

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderConsumer>();
    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 the project:

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:

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

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