# Subscription annotations > Register handlers with @Subscription and the other subscription annotations. [Spring Boot](/docs/spring-boot/) · Added in 0.18.0 If using the [Spring Boot Starter](/docs/spring-boot/starter/) you have the option to start subscriptions using the `@Subscription` annotation (`org.occurrent.annotation.Subscription`). For example if you have a domain event like this: * Java ```java public sealed interface DomainEvent permits DomainEvent1, DomainEvent2, DomainEvent3 { UUID eventId(); ZonedDateTime timestamp(); } public record DomainEvent1(UUID eventId, ZonedDateTime timestamp, String someData1) implements DomainEvent { } public record DomainEvent2(UUID eventId, ZonedDateTime timestamp, String someData2) implements DomainEvent { } public record DomainEvent3(UUID eventId, ZonedDateTime timestamp, String someData3) implements DomainEvent { } ``` * Kotlin ```kotlin sealed interface DomainEvent { val eventId: UUID val timestamp: ZonedDateTime } data class DomainEvent1( override val eventId: UUID, override val timestamp: ZonedDateTime, val someData1: String ) : DomainEvent data class DomainEvent2( override val eventId: UUID, override val timestamp: ZonedDateTime, val someData2: String ) : DomainEvent data class DomainEvent3( override val eventId: UUID, override val timestamp: ZonedDateTime, val someData3: String ) : DomainEvent ``` you can create a subscription to *all* events like this: * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents") void printAllDomainEvents(DomainEvent e) { System.out.println("Received domain event %s".formatted(e)); } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents") fun printAllDomainEvents(e: DomainEvent) { println("Received domain event $e") } } ``` Note that subscriptions started by the `Subscription` annotation will make use of [competing consumers](/docs/layer-0/competing-consumers/) so that if you run multiple instances of the application one of them will receive the event(s). Since version 0.30.0, `@Subscription` is capability-neutral. On an event store that has both the stream and the DCB capability it delivers both stream-written and DCB-appended events, filtered by event type, with catch-up over the shared global position. Use `@StreamSubscription` when you want stream events only, and [`@DcbSubscription`](/docs/layer-1/dcb-with-domain-events/#subscribing-to-dcb-events) when you want DCB events only with tag-based filtering. `@StreamSubscription` is configured the same way as `@Subscription` described below, only scoped to stream events. Its `startAt` takes the nested `StreamSubscription.StartPosition`, where replaying from the start is `BEGINNING_OF_TIME` rather than `BEGINNING`. All of these are asynchronous. For a handler that runs synchronously on the writer thread, before `execute` returns, and can commit atomically with the write, use [`@SynchronousSubscription`](/docs/layer-1/synchronous-subscriptions/) instead. It is documented under [Synchronous Subscriptions](/docs/layer-1/synchronous-subscriptions/). ## Subscription Start Position When creating a subscription using the `@Subscription` annotation you can specify how it should behave when first starting (creating) the subscription as well as how it should be resumed when the application is restarted. Here’s an example: * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING) void printAllDomainEvents(DomainEvent e) { System.out.println("Received domain event %s".formatted(e)); } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING) fun printAllDomainEvents(e: DomainEvent) { println("Received domain event $e") } } ``` This will first replay all historic events from the beginning of time and then continue subscribing to new events continuously. You can also start at a specific date by using `startAtISO8601()` or `startAtTimeEpochMillis()`. Note that the example above will *start* replaying historic events from the beginning of time when the subscription is started the first time. However, once the subscription is resumed, e.g. on application restart, it’ll continue from the last received event. Here’s a description for each StartPosition: | StartPosition | Description | | ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `BEGINNING` | Start this subscription from the first event in the event store. | | `NOW` | Start this subscription from current time. | | `DEFAULT` | Start this subscription using the default behavior of the subscription model. Typically, this means that it’ll start from `NOW`, unless the subscription has already been started before, in which case the subscription will be started from its last known checkpoint. | If you want a different behavior when the application is restarted, configure a different `resumeBehavior()` (`@Subscription(id="mySubscription", resumeBehavior=..)`): Resume behavior: | ResumeBehavior | Description | | ------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `DEFAULT` | Use the default resume behavior of the underlying subscription model. For example, if the `StartPosition` is set to `StartPosition.BEGINNING`, and `ResumeBehavior` is set to `ResumeBehavior.DEFAULT`, then the subscription will *start* from the beginning of time the first time it’s run. Then, on application restart, it’ll continue from the last received event (the checkpoint for the subscription) on restart. | | `SAME_AS_START_AT` | Always start at the same position as specified by the `StartPosition`. I.e., even if there’s a checkpoint stored for the subscription, it’ll be ignored on application restart and the subscription will resume from the specified `StartPosition`. | The combination of start and resume behavior can enable various use cases. For example: | StartPosition | ResumeBehavior | Use Case | | ------------- | ------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `BEGINNING` | `DEFAULT` | Start a new subscription from “beginning of time”, but when the app is restarted, continue from the last processed event (i.e. don’t replay all historic events again). | | `BEGINNING` | `SAME_AS_START_AT` | The subscription will always (even on restart) subscribe to all historic events, effectively making this an in-memory subscription. Note that this subscription will be called on each instance of the application even though competing consumer behavior is configured. | | `NOW` | `SAME_AS_START_AT` | The subscription will always (even on restart) subscribe to events from “now”, ignoring the historic events. This will effectively make this an in-memory subscription. Note that this subscription will be called on each instance of the application even though competing consumer behavior is configured. | | `NOW` | `DEFAULT` | The subscription start subscribing to events from “now” when created, but continue (resume) from the last received event on restart. | | `DEFAULT` | `DEFAULT` | Same as above (this is the default behavior if start position and resume behavior is not specified). | ## Selective Events If `DomainEvent` is a sealed interface/class (as in the examples above), then all events implementing this interface/class will be received when subscribing to this event. You can of course subscribe to an individual event, such as `DomainEvent2`. But if you want to receive only some of the events that implement the `DomainEvent` interface, you can use `eventTypes()`. For example, if you want to subscribe on both `DomainEvent1` and `DomainEvent3` but handle them as a `DomainEvent`: * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents", eventTypes = {DomainEvent1.class, DomainEvent3.class}) void printSomeDomainEvents(DomainEvent e) { System.out.println("Received any of the specified domain events: %s".formatted(e)); } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents", eventTypes = [DomainEvent1::class, DomainEvent3::class]) fun printSomeDomainEvents(e: DomainEvent) { println("Received any of the specified domain events: $e") } } ``` The filter Occurrent derives from a sealed type names the declared type as well as the concrete types it permits. That only matters if you wrote a `CloudEventTypeMapper` that maps a whole hierarchy onto the type string of the type it was declared with, because such a subscription used to receive nothing at all. No mapper Occurrent ships stores an event under a sealed interface’s own type, so nothing changes for the default setup. If you declare a type listed as refused in [Deriving the Event Filter](/docs/layer-1/deriving-the-event-filter/), Spring Boot startup fails. The message names the type and points you at `eventTypes()`. The same section has a table of when each of the other places refuses one. ## Event Metadata Sometimes it can be useful to get the metadata associated with the received event. For this reason, you can add a parameter of type `org.occurrent.cloudevents.EventMetadata` to a method annotated with `@Subscription` or `@StreamSubscription`. It contains all extension properties added to the [CloudEvent](/docs/layer-0/cloudevents/#cloudevent-metadata), with typed accessors for the common ones: `streamId`, `streamVersion`, and `position` (the global sequence number, or `null` for a stream-written event on a store that does not record a position). Note that for an event delivered through the capability-neutral `@Subscription` or the DCB path, `streamId` is the internal generated partition id rather than a domain stream id, but it is always present. For example: * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents") void printAllDomainEvents(DomainEvent e, EventMetadata metadata) { String streamId = metadata.getStreamId(); long streamVersion = metadata.getStreamVersion(); Long position = metadata.getPosition(); Object myCustomValue = metadata.get("MyCustomValue"); ... } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents") fun printAllDomainEvents(e: DomainEvent, metadata: EventMetadata) { val streamId: String = metadata.streamId val streamVersion: Long = metadata.streamVersion val position: Long? = metadata.position val myCustomValue: Any? = metadata["MyCustomValue"] // ... } } ``` When you only need the stream id or stream version, annotate a handler parameter with `@StreamId` or `@StreamVersion` instead of taking the whole `EventMetadata`. `@StreamId` binds a `String`, `@StreamVersion` a `long` or `Long`, and they may appear in any order alongside the event and an optional `EventMetadata` parameter. This works on `@Subscription`, `@StreamSubscription`, and `@SynchronousSubscription` (the same internal-partition-id caveat above applies on the capability-neutral `@Subscription`). On `@DcbSubscription` these annotations are rejected at startup, since a DCB handler’s stream id and version are internal partition values rather than domain ones. * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents") void printAllDomainEvents(DomainEvent e, @StreamId String streamId, @StreamVersion long streamVersion) { ... } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents") fun printAllDomainEvents(e: DomainEvent, @StreamId streamId: String, @StreamVersion streamVersion: Long) { // ... } } ``` ## Subscription Startup Mode You can configure whether the subscription should start before the application is ready to receive requests. For example, it might be very important that a certain subscription is started before the first web request comes in: * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents", startupMode = StartupMode.WAIT_UNTIL_STARTED) void printAllDomainEvents(DomainEvent e) { System.out.println("Received domain event %s".formatted(e)); } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents", startupMode = StartupMode.WAIT_UNTIL_STARTED) fun printAllDomainEvents(e: DomainEvent) { println("Received domain event $e") } } ``` In other cases, such as when replaying a huge number of historic events it might be better to let the application start and let the processing of historic events happen in the background. * Java ```java @Component public class Example { @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING, startupMode = StartupMode.BACKGROUND) void printAllDomainEvents(DomainEvent e) { System.out.println("Received domain event %s".formatted(e)); } } ``` * Kotlin ```kotlin @Component class Example { @Subscription(id = "printAllDomainEvents", startAt = StartPosition.BEGINNING, startupMode = StartupMode.BACKGROUND) fun printAllDomainEvents(e: DomainEvent) { println("Received domain event $e") } } ``` Here’s a summary of the different startup modes: | StartupMode | Description | | -------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | | `DEFAULT` | Determine the startup mode based on the properties of the subscription (such as `startAt()` and `resumeBehavior()`). It’ll use `BACKGROUND` if the subscription needs to replay historic events before subscribing to new ones (e.g. if `startAt()` is `StartPosition.BEGINNING_OF_TIME`), otherwise `WAIT_UNTIL_STARTED` will be used. | | `WAIT_UNTIL_STARTED` | The subscription will wait until it’s started up fully before Spring continues starting the rest of the application. Most of the time this is recommended because otherwise there could be a small chance that a request is received by your application before the subscription has bootstrapped completely. This can lead to the subscription missing this event. This is only true if the subscription is brand new. As soon as the subscription has received an event that is stored in a `org.occurrent.subscription.api.blocking.CheckpointStorage`, it’ll never miss an event during startup. | | `BACKGROUND` | The subscription will NOT wait until it’s started up fully before Spring continues starting the rest of the application; instead, it will be started in the background. Typically, this is useful if you instruct the subscription to start at an earlier date (such as the beginning of time), and you have a lot of events to read before the subscription has caught up. In this case, you may wish to start the Spring application before the subscription has fully started (i.e., before all historic events have been replayed) because waiting for all events to replay takes too long. The subscription will then replay all historic events in the background before switching to continuous mode. | These modes decide what happens while the application starts. The subscription handlers of a bean built after startup never wait for their replay, whatever `startupMode` says, see [When Annotations Register](/docs/spring-boot/subscription-annotations/#when-annotations-register). ## Spring Advice on Handler Methods Occurrent calls a handler method through the bean’s Spring proxy, so `@Transactional`, or any other aspect on the method or its class, applies to every event the method receives: ```java @Component public class OrderStatusUpdater { @Transactional @Subscription(id = "orderStatus") public void onOrderPlaced(OrderPlaced orderPlaced) { // runs inside a transaction } } ``` This applies to `@Subscription`, `@StreamSubscription`, `@DcbSubscription` and `@SynchronousSubscription` alike. If the proxy can’t reach a handler method, Spring Boot startup fails with `SubscriptionHandlerNotInvocableException` instead of running the method without its advice: | Handler method | Why the proxy can’t reach it | Fix | | ------------------------------------------------------------------ | --------------------------------------------- | ------------------------------------------------------------------------------- | | Declared only on the class, on a bean behind a JDK interface proxy | The proxy implements only the interfaces | Declare the method on an interface, or set `spring.aop.proxy-target-class=true` | | `private`, on a bean behind a CGLIB proxy | A CGLIB proxy can’t override a private method | Make the method non-private | | `final`, on a bean behind a CGLIB proxy | A CGLIB proxy can’t override a final method | Remove `final` | | `static` | Calling it never goes through any proxy | Make it an instance method | A private or final handler on a bean that isn’t proxied at all still runs, since there’s no advice to lose. ## When Annotations Register Every `@Subscription`, `@StreamSubscription`, `@DcbSubscription`, `@SynchronousSubscription`, `@Projection`, `@Snapshot` and `@Saga` registers after all singleton beans have been created and before any `ApplicationRunner` or `CommandLineRunner` runs. A subscription that only receives new events therefore sees an event written by an `ApplicationRunner`, but misses one written while the beans are still being created, from a `@PostConstruct` method for example. A `@Lazy` bean whose declared type shows one of these annotations is built at that point too, so it registers with the others. A bean whose declared type doesn’t show the annotation registers when something first builds it. That happens when a `@Lazy` `@Bean` method returns an interface and the annotated method is only on the class that implements it: ```java @Configuration public class NotifierConfig { @Bean @Lazy public OrderNotifier orderNotifier() { // OrderNotifier is an interface return new EmailOrderNotifier(); // @Subscription is declared on EmailOrderNotifier } } ``` A `FactoryBean` registers the same way when nothing has asked for its object yet. A bean that nothing ever builds never registers its handlers. A `@Subscription`, `@StreamSubscription` or `@DcbSubscription` registered this way doesn’t wait for its replay, whatever `startupMode` says, since the application is already running. On the reactor stack, the same goes for a `@Projection` or `@Snapshot`. ### A Bean Built on a Reactor Non-Blocking Thread On the reactor stack the bean can be built on a Reactor non-blocking thread, for example when a WebFlux handler asks for it: ```java Mono.fromCallable(() -> applicationContext.getBean(OrderNotifier.class)) .subscribeOn(Schedulers.parallel()); ``` Reactor doesn’t allow blocking on such a thread. Subscribing a handler that starts at the beginning or at an explicit position can block while the starter reads the position stored for it. On any other thread, every handler subscribes on the thread that builds the bean. On a non-blocking thread, it depends on where the handler starts: | Start position | What happens | | :------------------------------------- | :------------------------------------------------------------------------------------------ | | The beginning, or an explicit position | Subscribes on `Schedulers.boundedElastic()`. | | `NOW` or `DEFAULT` | Subscribes on the thread that builds the bean, which works on the model the starter builds. | A handler that starts at `NOW` or `DEFAULT` subscribes on the calling thread because it starts from wherever the event feed is when it subscribes. Subscribing it later could skip what the caller writes in between. A handler that starts at the beginning or at an explicit position receives the same events no matter when it subscribes. When a subscription model other than the one the starter builds blocks while subscribing a `NOW` or `DEFAULT` handler, the bean fails to build instead, and the exception message says what to do. To fix it, build the bean on a thread that may block, with `Schedulers.boundedElastic()` instead of `Schedulers.parallel()` above, or start the handler at the beginning or at an explicit position. When the subscribe runs on `Schedulers.boundedElastic()`, there’s no caller to throw an exception to, so a failure there is logged at `ERROR`. After a failure that might be temporary, an unreachable storage for example, Occurrent tries the subscribe again until an attempt succeeds or the application context closes. The delay between attempts doubles from 100 ms up to 30 seconds, and the handler receives no events until an attempt succeeds. Some failures happen the same way on every attempt. Occurrent logs these once and gives up on the subscription. They include: * a duplicate subscription id, or a filter or start position the model refuses * a bean the context doesn’t have, has with another type, or has only as an abstract definition * a subscription model that was shut down * an `Error` If a bean the subscribe needs, `CheckpointStorage` for example, failed to build, Occurrent looks at the root cause of that build failure to decide whether to try again. The handlers on one bean subscribe one after the other. So when Occurrent gives up on one handler, it also gives up on the handlers after it, and the log names them. For a singleton bean, fix the cause and restart the application, since Spring builds a singleton only once. For a bean that isn’t a singleton, the next instance Spring builds tries to register its handlers again.