# Events & Events Store (/integrations/spring-boot/how-to/events-and-events-store)



## Overview [#overview]

KubeMQ provides two publish/subscribe patterns, both exposed through the Spring Boot starter:

* **Events** — fire-and-forget, multicast delivery. An event is fanned out to every active subscriber on the channel and is **not persisted**. If no subscriber is connected when the event is published, it is lost.
* **Events Store** — the same pub/sub model, but messages are **persisted to the broker** and assigned a monotonically increasing sequence number. New or reconnecting subscribers can **replay** historical events from a chosen starting point.

In the Spring adapter both patterns are **push-based subscriptions**: you annotate a bean method with `@KubeMQEventListener` or `@KubeMQEventStoreListener`, and the auto-configured listener container opens a streaming gRPC subscription to the broker and invokes your method for each delivered message. Publishing is done through the injectable `KubeMQTemplate`.

This page documents the Spring API for these two patterns. For the underlying broker semantics, see the core [Events](/learn/events) and [Events Store](/learn/events-store) concepts.

<Mermaid
  chart="`
flowchart LR
  Pub[&#x22;KubeMQTemplate<br/>sendEvent / sendEventStore&#x22;] -->|gRPC :50000| Broker[&#x22;KubeMQ Broker&#x22;]
  Broker -->|push| L1[&#x22;@KubeMQEventListener&#x22;]
  Broker -->|push| L2[&#x22;@KubeMQEventStoreListener<br/>(with replay)&#x22;]
  Store[(&#x22;Events Store<br/>persisted log&#x22;)] --- Broker

  class Pub,L1,L2 client
  class Broker broker
  class Store store
`"
/>

*The template publishes events; the broker pushes to annotated listeners, and Events Store persists the log for replay.*

<Callout type="info">
  The starter speaks gRPC to the broker on port `50000`. Start a local KubeMQ broker with Docker — the gRPC API listens on `50000` and the shared HTTP/REST and dashboard endpoints on `9090`:

  <RunKubeMQ ports="[50000, 9090]" />

  Point the application at the broker in `application.yml`:

  ```yaml title="application.yml"
  kubemq:
    address: ${KUBEMQ_ADDRESS:localhost:50000}
    client-id: spring-events-basic-pubsub
  ```
</Callout>

## Publishing Events [#publishing-events]

Inject `KubeMQTemplate` and call `sendEvent(channel, data)`. The template serializes the payload through the configured message converter (falling back to `byte[]` / `String` when none is set) and publishes a fire-and-forget event.

<Tabs groupId="spring-language" items="['Java', 'Kotlin']">
  <Tab value="Java">
    ```java title="BasicPubSubRunner.java"
    @Component
    public class BasicPubSubRunner implements ApplicationRunner {

        private static final Logger log = LoggerFactory.getLogger(BasicPubSubRunner.class);

        private final KubeMQTemplate template;

        public BasicPubSubRunner(KubeMQTemplate template) {
            this.template = template;
        }

        @Override
        public void run(ApplicationArguments args) throws InterruptedException {
            // Allow annotated listeners to subscribe before sending.
            Thread.sleep(500);
            for (int i = 1; i <= 3; i++) {
                template.sendEvent("spring-events.basic-pubsub", "Hello KubeMQ #" + i);
                log.info("Published event to spring-events.basic-pubsub: Hello KubeMQ #{}", i);
            }
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="PublishRunner.kt"
    @Component
    class PublishRunner(private val template: KubeMQTemplate) : ApplicationRunner {

        private val log = LoggerFactory.getLogger(javaClass)

        override fun run(args: ApplicationArguments) = runBlocking {
            for (i in 1..3) {
                // Coroutine extension from the Kotlin starter.
                template.sendEventSuspend("spring-events.basic-pubsub", "Hello KubeMQ #$i")
                log.info("Published event #{}", i)
            }
        }
    }
    ```
  </Tab>
</Tabs>

To attach metadata, pass a tags map with the `sendEvent(channel, data, tags)` overload:

```java title="Publishing with tags"
template.sendEvent(
        "spring-events.basic-pubsub",
        "Hello KubeMQ",
        Map.of("source", "order-service", "region", "us"));
```

### Async publishing [#async-publishing]

For non-blocking publishing, `sendEventAsync` returns a `CompletableFuture<Void>` that completes when the broker has accepted the event. It also has a tags overload.

```java title="Async publish"
CompletableFuture<Void> future =
        template.sendEventAsync("spring-events.basic-pubsub", "Hello KubeMQ");

future.thenRun(() -> log.info("Event accepted by broker"));
```

## Subscribing to Events [#subscribing-to-events]

Annotate a bean method with `@KubeMQEventListener` and accept an `EventMessageReceived`. The starter opens the subscription on application startup and invokes the method for each event delivered to the channel.

```java title="BasicPubSubListener.java"
@Component
public class BasicPubSubListener {

    private static final Logger log = LoggerFactory.getLogger(BasicPubSubListener.class);

    @KubeMQEventListener(channels = "spring-events.basic-pubsub")
    public void onEvent(EventMessageReceived event) {
        byte[] body = event.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Received event: channel={} body={}", event.getChannel(), text);
    }
}
```

The `@KubeMQEventListener` annotation supports these attributes. All string attributes accept SpEL expressions and `${...}` property placeholders, so channels and tuning can be externalized to configuration.

<TypeTable
  type="{
  channels: {
    description: 'Channel names to subscribe to.',
    type: 'String[]',
    required: true,
  },
  group: {
    description: 'Consumer group for shared (load-balanced) subscriptions. Empty means no group.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  concurrency: {
    description: 'Number of concurrent message processors. Resolved as an integer at runtime.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  containerFactory: {
    description: 'Bean name of a custom KubeMQListenerContainerFactory.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  autoStartup: {
    description: 'Whether this listener auto-starts with the application context.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  errorHandler: {
    description: 'Bean name of a custom org.springframework.util.ErrorHandler.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  id: {
    description: 'Unique identifier for this listener endpoint.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
}"
/>

### Consumer groups [#consumer-groups]

When several listeners share the same `group`, the broker load-balances events across the group members instead of fanning out to all of them — exactly one member receives each event. This is how you scale out processing horizontally. The example below pairs two identical listeners in the `my-group` group.

```java title="ConsumerGroupListenerA.java"
@Component
public class ConsumerGroupListenerA {

    private static final Logger log = LoggerFactory.getLogger(ConsumerGroupListenerA.class);

    @KubeMQEventListener(channels = "spring-events.consumer-group", group = "my-group")
    public void onEvent(EventMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("[Listener-A] Received: {}", body);
    }
}
```

`ConsumerGroupListenerB` is identical except for its log prefix and joins the same group. With both running, the events published to `spring-events.consumer-group` are split between the two listeners.

### Multiple subscribers (fanout) [#multiple-subscribers-fanout]

Without a `group`, every listener on a channel receives a copy of each event. Two beans subscribed to `spring-events.multi-sub` both receive every message.

```java title="SubscriberOne.java"
@Component
public class SubscriberOne {

    private static final Logger log = LoggerFactory.getLogger(SubscriberOne.class);

    @KubeMQEventListener(channels = "spring-events.multi-sub")
    public void onEvent(EventMessageReceived event) {
        String body = event.getBody() != null
                ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
        log.info("[Subscriber-1] Received: {}", body);
    }
}
```

### Wildcard subscriptions [#wildcard-subscriptions]

A channel pattern ending in `>` matches all sub-channels at and below that prefix, so one listener can consume an entire channel hierarchy. The listener below receives events published to `spring-events.orders.us`, `spring-events.orders.eu`, `spring-events.orders.asia`, and any other `spring-events.orders.*` channel.

```java title="WildcardSubscribeListener.java"
@Component
public class WildcardSubscribeListener {

    private static final Logger log = LoggerFactory.getLogger(WildcardSubscribeListener.class);

    @KubeMQEventListener(channels = "spring-events.orders.>")
    public void onEvent(EventMessageReceived event) {
        byte[] body = event.getBody();
        String text = body != null ? new String(body, StandardCharsets.UTF_8) : "<empty>";
        log.info("Wildcard received: channel={} body={}", event.getChannel(), text);
    }
}
```

### Cancelling a subscription [#cancelling-a-subscription]

Annotation-driven listeners are managed by the application lifecycle, but you can also subscribe imperatively against the underlying `PubSubClient` and cancel the subscription when you no longer need it. Build an `EventsSubscription`, subscribe, and call `cancel()` to stop receiving events.

```java title="CancelSubscriptionRunner.java"
EventsSubscription subscription = EventsSubscription.builder()
        .channel("spring-events.cancel-sub")
        .onReceiveEventCallback(event -> {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Received: {}", body);
        })
        .onErrorCallback(err -> log.error("Subscription error: {}", err.getMessage()))
        .build();

pubSubClient.subscribeToEvents(subscription);
log.info("Subscription started on spring-events.cancel-sub");

template.sendEvent("spring-events.cancel-sub", "Before cancel"); // received

subscription.cancel();
log.info("Subscription cancelled");

template.sendEvent("spring-events.cancel-sub", "After cancel (not received)"); // dropped
```

## Publishing to Events Store [#publishing-to-events-store]

Publishing to Events Store is identical to publishing events, but uses `sendEventStore`. The broker persists each message to the channel's durable log and assigns it a sequence number, making it available for later replay.

```java title="StartNewOnlyRunner.java"
@Component
public class StartNewOnlyRunner implements ApplicationRunner {

    private final KubeMQTemplate template;

    public StartNewOnlyRunner(KubeMQTemplate template) {
        this.template = template;
    }

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {
        Thread.sleep(500);
        for (int i = 1; i <= 5; i++) {
            template.sendEventStore("spring-events-store.start-new-only", "New-only event #" + i);
        }
    }
}
```

A tags overload (`sendEventStore(channel, data, tags)`) and an async variant (`sendEventStoreAsync`, returning `CompletableFuture<Void>`) are available, mirroring the plain events API.

```java title="Async events-store publish"
CompletableFuture<Void> future =
        template.sendEventStoreAsync("spring-events-store.start-new-only", "Hello");
```

## Subscribing with Replay [#subscribing-with-replay]

`@KubeMQEventStoreListener` subscribes to a persisted channel and receives `EventStoreMessageReceived` instances, which expose the broker-assigned `getSequence()` in addition to the body and tags. The `subscriptionType` attribute controls where the replay begins.

<TypeTable
  type="{
  channels: {
    description: 'Channel names to subscribe to.',
    type: 'String[]',
    required: true,
  },
  subscriptionType: {
    description: 'Replay start type: StartNewOnly, StartFromFirst, StartFromLast, StartAtSequence, StartAtTime, or StartAtTimeDelta.',
    type: 'String',
    default: 'StartNewOnly',
  },
  subscriptionValue: {
    description: 'Value associated with the subscription type (sequence number or timestamp). Resolved as a long at runtime.',
    type: 'String',
    default: '0',
  },
  group: {
    description: 'Consumer group for shared subscriptions.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  concurrency: {
    description: 'Number of concurrent message processors.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  autoStartup: {
    description: 'Whether this listener auto-starts with the application context.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
  id: {
    description: 'Unique identifier for this listener endpoint.',
    type: 'String',
    default: '&#x22;&#x22;',
  },
}"
/>

### Start types [#start-types]

Each `subscriptionType` maps to a runnable example in the repository. The `StartAtSequence`, `StartAtTime`, and `StartAtTimeDelta` types additionally read `subscriptionValue`.

| `subscriptionType` | Behavior                                                     | Example module                  |
| ------------------ | ------------------------------------------------------------ | ------------------------------- |
| `StartNewOnly`     | Only events published after the subscription starts.         | `events-store-start-new-only`   |
| `StartFromFirst`   | Replay from the first persisted event (sequence 1).          | `events-store-start-from-first` |
| `StartFromLast`    | Deliver only the last persisted event, then continue live.   | `events-store-start-from-last`  |
| `StartAtSequence`  | Replay starting at `subscriptionValue` (a sequence number).  | `events-store-replay-sequence`  |
| `StartAtTime`      | Replay starting at `subscriptionValue` (an epoch timestamp). | `events-store-replay-time`      |
| `StartAtTimeDelta` | Replay events from the last `subscriptionValue` seconds.     | `events-store-time-delta`       |

<Tabs groupId="es-start-type" items="['StartNewOnly', 'StartFromFirst', 'StartFromLast', 'StartAtSequence', 'StartAtTime', 'StartAtTimeDelta']">
  <Tab value="StartNewOnly">
    ```java title="StartNewOnlyListener.java"
    @Component
    public class StartNewOnlyListener {

        private static final Logger log = LoggerFactory.getLogger(StartNewOnlyListener.class);

        @KubeMQEventStoreListener(
                channels = "spring-events-store.start-new-only",
                subscriptionType = "StartNewOnly")
        public void onEvent(EventStoreMessageReceived event) {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Received (new-only): sequence={} body={}", event.getSequence(), body);
        }
    }
    ```
  </Tab>

  <Tab value="StartFromFirst">
    ```java title="StartFromFirstListener.java"
    @Component
    public class StartFromFirstListener {

        private static final Logger log = LoggerFactory.getLogger(StartFromFirstListener.class);

        @KubeMQEventStoreListener(
                channels = "spring-events-store.start-from-first",
                subscriptionType = "StartFromFirst")
        public void onEvent(EventStoreMessageReceived event) {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Received (from-first): sequence={} body={}", event.getSequence(), body);
        }
    }
    ```
  </Tab>

  <Tab value="StartFromLast">
    ```java title="StartFromLastListener.java"
    @Component
    public class StartFromLastListener {

        private static final Logger log = LoggerFactory.getLogger(StartFromLastListener.class);

        @KubeMQEventStoreListener(
                channels = "spring-events-store.start-from-last",
                subscriptionType = "StartFromLast")
        public void onEvent(EventStoreMessageReceived event) {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Received (from-last): sequence={} body={}", event.getSequence(), body);
        }
    }
    ```
  </Tab>

  <Tab value="StartAtSequence">
    ```java title="ReplayFromSequenceListener.java"
    @Component
    public class ReplayFromSequenceListener {

        private static final Logger log = LoggerFactory.getLogger(ReplayFromSequenceListener.class);

        @KubeMQEventStoreListener(
                channels = "spring-events-store.replay-sequence",
                subscriptionType = "StartAtSequence",
                subscriptionValue = "5")
        public void onEvent(EventStoreMessageReceived event) {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Replayed from sequence: sequence={} body={}", event.getSequence(), body);
        }
    }
    ```
  </Tab>

  <Tab value="StartAtTime">
    ```java title="ReplayFromTimeListener.java"
    @Component
    public class ReplayFromTimeListener {

        private static final Logger log = LoggerFactory.getLogger(ReplayFromTimeListener.class);

        // subscriptionValue is externalized; resolves to an epoch timestamp at runtime.
        @KubeMQEventStoreListener(
                channels = "spring-events-store.replay-time",
                subscriptionType = "StartAtTime",
                subscriptionValue = "${kubemq.replay.start-time:0}")
        public void onEvent(EventStoreMessageReceived event) {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Replayed from time: sequence={} body={}", event.getSequence(), body);
        }
    }
    ```
  </Tab>

  <Tab value="StartAtTimeDelta">
    ```java title="TimeDeltaListener.java"
    @Component
    public class TimeDeltaListener {

        private static final Logger log = LoggerFactory.getLogger(TimeDeltaListener.class);

        // Replay everything from the last 60 seconds.
        @KubeMQEventStoreListener(
                channels = "spring-events-store.time-delta",
                subscriptionType = "StartAtTimeDelta",
                subscriptionValue = "60")
        public void onEvent(EventStoreMessageReceived event) {
            String body = event.getBody() != null
                    ? new String(event.getBody(), StandardCharsets.UTF_8) : "<empty>";
            log.info("Received (time-delta 60s): sequence={} body={}", event.getSequence(), body);
        }
    }
    ```
  </Tab>
</Tabs>

## Fluent Builder Alternative [#fluent-builder-alternative]

For publishing, the template also exposes a fluent builder. `template.newEvent(data)` and `template.newEventStore(data)` return a `KubeMQEventMessageBuilder` on which you set the channel, tags, and optional metadata before calling `send()` (or `sendAsync()` for a `CompletableFuture<Void>`). This reads well when attaching several tags or metadata at the call site.

```java title="FluentBuildersRunner.java"
// Fire-and-forget event with a tag.
template.newEvent("Tagged event")
        .toChannel("spring-fluent.events")
        .withTag("source", "fluent-builder")
        .send();

// Persisted events-store message with metadata and async send.
template.newEventStore("Audit record")
        .toChannel("spring-fluent.events-store")
        .withMetadata("v1")
        .withTag("kind", "audit")
        .sendAsync();
```

## Kotlin Coroutines [#kotlin-coroutines]

The Kotlin starter (`kubemq-spring-boot-starter-kotlin`) adds suspend extension functions that bridge the template's `CompletableFuture`-based async methods to coroutines. `sendEventSuspend` and `sendEventStoreSuspend` suspend until the broker accepts the message and respect structured-concurrency cancellation. Each has an overload that accepts a tags `Map<String, String>`.

```kotlin title="PublishService.kt"
@Service
class PublishService(private val template: KubeMQTemplate) {

    suspend fun publishOrder(order: String) {
        // Fire-and-forget event.
        template.sendEventSuspend("spring-events.orders", order)

        // Persisted events-store message with tags.
        template.sendEventStoreSuspend(
            "spring-events-store.orders",
            order,
            mapOf("source" to "order-service"),
        )
    }
}
```

## Events vs Events Store [#events-vs-events-store]

<Callout type="info">
  Choose **Events** for real-time, ephemeral notifications where losing a message to a momentarily disconnected subscriber is acceptable — telemetry, live UI updates, presence. Choose **Events Store** when you need durability and replay: late-joining consumers, audit trails, or rebuilding state from history.

  |                                 | Events                         | Events Store                                |
  | ------------------------------- | ------------------------------ | ------------------------------------------- |
  | Persistence                     | None (in-memory fanout)        | Persisted to the broker                     |
  | Delivery to offline subscribers | Lost                           | Replayable later                            |
  | Replay                          | Not supported                  | Choose a start point via `subscriptionType` |
  | Publish API                     | `sendEvent` / `sendEventAsync` | `sendEventStore` / `sendEventStoreAsync`    |
  | Listener                        | `@KubeMQEventListener`         | `@KubeMQEventStoreListener`                 |
</Callout>

## Next Steps [#next-steps]

<Cards>
  <Card title="Getting Started" href="/integrations/spring-boot/tutorials/getting-started" description="Add the starter, configure a broker, and send and receive your first message." />

  <Card title="Queues" href="/integrations/spring-boot/how-to/queues" description="Durable point-to-point messaging with competing consumers." />

  <Card title="Commands & Queries" href="/integrations/spring-boot/how-to/commands-and-queries" description="Synchronous request-response with the template and handler annotations." />

  <Card title="Reference" href="/integrations/spring-boot/reference/configuration" description="Configuration properties, the KubeMQTemplate API, and listener annotations." />
</Cards>