# Events (Publish) (/integrations/masstransit/how-to/events)



## Overview [#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](/learn/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.

<Callout type="warn">
  Events are **volatile**. If no subscriber is active when an event is published, the message is **silently dropped** — there is no error and no retry. Always make sure subscribers are running *before* you publish. For durable, replayable fan-out where late subscribers can catch up, use [Events Store](/integrations/masstransit/how-to/events-store) instead.
</Callout>

<Mermaid
  chart="`
graph LR
P[&#x22;bus.Publish&lt;T&gt;()&#x22;]
R{{&#x22;MassTransit.KubeMQ<br/>transport&#x22;}}
E[&#x22;KubeMQ Events channel&#x22;]
S1[&#x22;Subscriber 1&#x22;]
S2[&#x22;Subscriber 2&#x22;]
S3[&#x22;Subscriber 3&#x22;]

P --> R
R -- &#x22;SendEventAsync (gRPC :50000)&#x22; --> E
E -- &#x22;fan-out&#x22; --> S1
E -- &#x22;fan-out&#x22; --> S2
E -- &#x22;fan-out&#x22; --> S3

class P,S1,S2,S3 external
class R aiway
class E broker
`"
/>

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

## API surface [#api-surface]

| Member                                                        | Where            | Purpose                                                                           |
| ------------------------------------------------------------- | ---------------- | --------------------------------------------------------------------------------- |
| `IPublishEndpoint.Publish<T>(message)`                        | bus              | Publish an event; channel is derived from the message type name                   |
| `IKubeMQReceiveEndpointConfigurator.UseVolatileEvents()`      | receive endpoint | Subscribe the endpoint to non-persistent Events (not EventsStore)                 |
| `IKubeMQRider.PublishEventAsync<T>(message, channelName, ct)` | rider            | Publish a volatile event to an explicit channel (useful from a background worker) |

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](/integrations/masstransit/reference/configuration).

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

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

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

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

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

```csharp title="Program.cs"
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 [#limitation-no-delayed-delivery]

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

```text title="error"
Delayed delivery is not supported for Events. Use Queues.
```

Scheduling and delay are queue features. To defer a message, use [Queues](/integrations/masstransit/how-to/queues), which supports delayed delivery via `ctx.Delay` (mapped to `QueueMessage.DelaySeconds`) and guaranteed, persisted, point-to-point delivery.

## Errors and next steps [#errors-and-next-steps]

* Diagnosing dropped events, consumer-group surprises, and `KubeMQTransportConfigurationException` → [Error Handling & DLQ](/integrations/masstransit/how-to/error-handling-dlq).
* The full exception hierarchy and transport-error model → [reference/error-codes](/integrations/masstransit/reference/error-codes).

<Cards>
  <Card title="Events Store (Durable Publish)" href="/integrations/masstransit/how-to/events-store" description="Durable, replayable fan-out where late subscribers catch up from a configurable start position." />

  <Card title="Queues (Send)" href="/integrations/masstransit/how-to/queues" description="Guaranteed point-to-point delivery with persistence, TTL, and delayed delivery." />

  <Card title="Concepts" href="/integrations/masstransit/concepts" description="Pattern mapping, channel naming, header mapping, and the rider architecture." />
</Cards>