Events (Publish)

Overview

MassTransit's IPublishEndpoint.Publish<T>() maps to KubeMQ Events — fire-and-forget fan-out where every active subscriber bound to the message type receives its own copy. Events are not persisted, no acknowledgment is returned, and the publisher does not wait for any consumer. This is the pattern for broadcast notifications, cache-invalidation signals, telemetry, and other "tell everyone, don't wait" cases.

→ For what an Event is and KubeMQ's pub/sub guarantees, see Events. This page documents the MassTransit transport's API surface only.

Internally the transport publishes via the KubeMQ SDK's SendEventAsync (an EventMessage); receive endpoints subscribe via SubscribeToEventsAsync. There is no queue, exchange, or topology to provision — the Events channel is created on first use.

A published event fans out over a KubeMQ Events channel to every active subscriber — fire-and-forget, no persistence.

API surface

When you call bus.Publish<T>(), the Events channel name is derived from the message type's full name — for a contract MyApp.Events.OrderSubmitted, the channel is MyApp.Events.OrderSubmitted. The : character in a type name is normalized to .. Channels are created on demand, so no topology is declared up front. Full naming rules live in the reference.

Usage

A publisher calls Publish<T>(); every consumer registered for that message type receives the event. The consumer is a plain IConsumer<T> — the same contract you would write for any other transport.

// Publishing an event — fans out to all active subscribers
await bus.Publish(new OrderSubmitted { OrderId = "123" });

public class OrderSubmittedConsumer : IConsumer<OrderSubmitted>
{
    public async Task Consume(ConsumeContext<OrderSubmitted> context)
    {
        // Handle event — fire-and-forget, no ack is sent back to the publisher
    }
}

Subscribing with UseVolatileEvents

A receive endpoint subscribes to non-persistent Events (rather than EventsStore) by calling UseVolatileEvents() in its configuration. Register the consumer with AddConsumer<T>, then attach a receive endpoint that opts into volatile event subscription.

using MassTransit;
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);

        // Subscribe to volatile (non-persistent) Events — not EventsStore
        cfg.ReceiveEndpoint("events-basic-pubsub", e =>
        {
            e.UseVolatileEvents();
        });
    });
});

var host = builder.Build();
await host.RunAsync();

// --- Message contract ---
public record OrderPlaced(string OrderId, string Product, decimal Amount, DateTime PlacedAt);

// --- Consumer ---
public class OrderPlacedConsumer(ILogger<OrderPlacedConsumer> logger) : IConsumer<OrderPlaced>
{
    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;
    }
}

UseVolatileEvents() is the distinguishing call for this pattern. Without it, an endpoint that subscribes to events would use the persistent EventsStore subscription path.

Publishing from a background service

In a hosted application you typically publish from a BackgroundService. Inject IBus and call bus.Publish<T>(), or — when you only have the KubeMQ rider — publish directly through KubeMQRiderAccessor.Current, which exposes PublishEventAsync(message, channelName, ct).

public class EventPublisher(
    ILogger<EventPublisher> logger,
    IHostApplicationLifetime lifetime) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        // Allow the bus and subscriptions to start before publishing —
        // events are volatile, so a subscriber must be active first.
        await Task.Delay(3000, stoppingToken);

        var rider = KubeMQRiderAccessor.Current
            ?? throw new InvalidOperationException("KubeMQ rider not started.");

        for (var counter = 1; counter <= 5; counter++)
        {
            var order = new OrderPlaced(
                $"ORD-{counter:D4}",
                $"Product-{counter}",
                19.99m * counter,
                DateTime.UtcNow);

            logger.LogInformation("Publishing volatile event: OrderId={OrderId}", order.OrderId);

            // Publish a volatile event to the named Events channel (fan-out)
            await rider.PublishEventAsync(order, "events-basic-pubsub", stoppingToken);

            await Task.Delay(1000, stoppingToken);
        }

        await Task.Delay(3000, stoppingToken);
        lifetime.StopApplication();
    }
}

Note the two publishing styles. bus.Publish<T>(message) routes by message type (the channel is derived from the type name). rider.PublishEventAsync<T>(message, channelName) publishes to an explicit channel name, which is useful when a background worker wants to target a specific endpoint channel directly. The 3-second startup delay is deliberate: it gives receive endpoints time to establish their subscriptions before the first event goes out, so nothing is dropped.

Consumer groups

By default, fan-out delivers each event to every subscriber. To load-balance events across a pool of competing consumers instead, give multiple receive endpoints the same endpoint name — KubeMQ treats them as one consumer group (via the subscription's Group), and each event is delivered to exactly one member.

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<SensorReadingConsumer>();

    x.UsingKubeMQ((context, cfg) =>
    {
        cfg.Host("localhost", 50000);

        // Two endpoints on the SAME channel form a consumer group.
        // KubeMQ distributes events across members — each event to one member only.
        cfg.ReceiveEndpoint("events-consumer-group", e =>
        {
            e.UseVolatileEvents();
        });

        cfg.ReceiveEndpoint("events-consumer-group", e =>
        {
            e.UseVolatileEvents();
        });
    });
});

The same grouping rule applies whether the members run in one process (as above) or scale out across instances — competing members sharing an endpoint name always load-balance the fan-out. To deliver every event to every subscriber instead, give each endpoint a distinct name.

Limitation: no delayed delivery

Delayed delivery is not supported for Events. Attempting to publish an event with a delay throws a KubeMQTransportConfigurationException:

Delayed delivery is not supported for Events. Use Queues.

Scheduling and delay are queue features. To defer a message, use Queues, which supports delayed delivery via ctx.Delay (mapped to QueueMessage.DelaySeconds) and guaranteed, persisted, point-to-point delivery.

Errors and next steps

• Diagnosing dropped events, consumer-group surprises, and KubeMQTransportConfigurationException → Error Handling & DLQ.
• The full exception hierarchy and transport-error model → reference/error-codes.