# Subscription life-cycle

> Start, stop and test subscriptions.

[Layer 0: CloudEvents](/docs/layer-0/) · Blocking · Added in 0.7.0

Subscription models may also implement the `SubscriptionModelLifeCycle` interface (currently all blocking subscription models implement this). These subscription models support cancelling, pausing and resuming individual subscriptions. You can also stop an entire subscription model temporarily (`stop`) and restart it later (`start`). That “all” includes the models you might expect to be exceptions. A [synchronous subscription](/docs/layer-1/synchronous-subscriptions/#life-cycle) and a [push subscription](/docs/layer-0/push-subscription/#life-cycle) can be stopped, paused and shut down exactly like a MongoDB-backed one.

Calling `start()` on a model that’s already running is allowed. Starting a model is a goal, not a one-time transition, so a second call brings up whatever is still down and leaves the rest exactly as it was, including a subscription you paused yourself with `pauseSubscription(id)`. That’s what lets a leader election or a health check call `start()` without checking `isRunning()` first, which is what [`ManualStartSubscriptionModel`](#starting-subscriptions-manually) is for.

```java
// Fine to call on every tick, whether or not the model is already running
subscriptionModel.start(true);
```

`CompetingConsumerSubscriptionModel` used to throw `IllegalStateException` here. Every other subscription model already accepted a repeated `start()`, and now this one does too.

`Subscription.waitUntilStarted(..)` answers for the one `start()` the handle was created for. It’s not a live status check.

```java
Subscription subscription = subscriptionModel.subscribe("orders", e -> handleOrderEvent(e));
subscription.waitUntilStarted(Duration.ofSeconds(10)); // true


subscriptionModel.pauseSubscription("orders");
subscription.waitUntilStarted(Duration.ofSeconds(10)); // still true, "orders" is paused right now
```

Once a subscription has started, its handle keeps answering `true` even after you pause it, stop it, or it loses a competing consumer lock. Ask `isRunning(id)` and `isPaused(id)` when you want to know what’s happening right now.

A handle returns `false` for a subscription you still have to start yourself:

* a registration withheld by a [`ManualStartSubscriptionModel`](#starting-subscriptions-manually)
* one registered while a `PushSubscriptionModel`, a `SynchronousSubscriptionModel`, a `NativeMongoSubscriptionModel` or a `SpringMongoSubscriptionModel` is stopped
* a catch-up replay that `stop()` interrupted

Apart from the two MongoDB models, a start that failed and won’t be retried throws instead of returning.

`waitUntilStarted()` without a timeout keeps waiting for a subscription you still have to start yourself, so don’t call it on the thread that is going to call `start()`.

`NativeMongoSubscriptionModel` and `SpringMongoSubscriptionModel` also return `false` once the model is shut down, and when the change stream hasn’t opened before the timeout. The stream may not open, for example, while MongoDB can’t be reached or because the `RetryStrategy` gave up before it opened.

On the reactor stack, `waitUntilStarted()` returns a `Mono<Void>` instead of blocking. It completes once the subscription has started and errors if the start failed, so a subscription that hasn’t started yet is a `Mono` that hasn’t completed.

On `ReactorDurableSubscriptionModel`, the `Mono` fails with `CancellationException` when the subscription is cancelled before the durable model hands it to the model it wraps. Once that wrapped model has the subscription, the `Mono` completes or fails the same way as the wrapped model’s own `waitUntilStarted()`.

The durable model drives the subscription itself when the wrapped model isn’t a reactor `SubscriptionModel`. Then these fail the `Mono` with `CancellationException` when they come before the subscription started:

* a cancel
* a pause or a `stop()`, unless the subscription was made while the model was stopped

Note the difference between cancelling and pausing a subscription. Cancelling a subscription will *remove* it and it’s not possible to resume it again later. Pausing a subscription will temporarily pause the subscription, but it can later be resumed using the `resumeSubscription` method.

Many of the methods in the `SubscriptionModelLifeCycle` are good to have when you write integration tests. It’s often useful to e.g. write events to the event store *without* triggering all subscriptions listening to the events. The life cycle methods allow you to selectively start/stop individual subscriptions so that you can (integration) test them in isolation.

You don’t have to drive this by hand in your tests though. The `occurrent-testing-junit-jupiter-blocking` extension stops every subscription model by default and lets each test opt subscriptions back in, as described in [Integration Testing](/docs/testing/integration-testing/), including [using the subscription life cycle](/docs/testing/integration-testing/#using-the-subscription-life-cycle) and [stopping every subscription, then opting in](/docs/testing/integration-testing/#stopping-every-subscription-then-opting-in).

A subscription model may also implement `IntrospectableSubscriptions`, which adds `subscriptionIds()`, every id it knows about, running or paused. Not every subscription model does, so use the static `IntrospectableSubscriptions.findIn(subscriptionModel)`. It unwraps a chain of wrapping subscription models, a `DurableSubscriptionModel` wrapping a `CatchupSubscriptionModel` wrapping a `NativeMongoSubscriptionModel`, for example, until it finds one that implements it, or returns empty if nothing in the chain does. That’s what lets a caller holding a wrapped model ask what it’s subscribed to, without knowing its concrete type or how many layers deep the answer lives.

`ReplayAwareSubscriptions` is the same kind of capability interface, with two methods.

`isCatchingUp(subscriptionId)` answers whether a subscription is still replaying history or has handed over to live delivery. `isRunning(subscriptionId)` can’t tell you that, since it is true throughout a replay.

The catch-up models on both stacks implement it, including `CatchupThenPushSubscriptionModel`. `ReplayAwareSubscriptions.findIn(subscriptionModel)` unwraps a wrapping chain the same way as above, so a `DurableSubscriptionModel` wrapping a `CatchupSubscriptionModel` answers too.

Use it in a readiness probe when you run a catch-up model directly. Note that a saga’s timers ask exactly this question. They don’t fire until the replay has finished.

The second method, `listenForCatchup(subscriptionId, listener)`, tells a projection which part of a catch-up it’s in. A catch-up first reads the history that was already there, and then delivers whatever was written while it was reading.

The model calls the listener’s `catchupStarted` when a catch-up begins, and its `historyRead` once that catch-up has read the history that was already there. For a catch-up that’s stopped before then, the model never calls `historyRead`.

A projection that [records the appends it has applied](/docs/spring-boot/projection-annotation/#read-your-writes-for-an-asynchronous-projection) stops recording at `catchupStarted` and starts again at `historyRead`, so it still records the events written while the catch-up was reading.

When the live feed offers a copy of an event its replay already delivered, `CatchupThenPushSubscriptionModel` doesn’t deliver the copy and calls the listener’s `alreadyDeliveredByReplay(event)` instead. A projection that records its appends then records the append of that event, as long as the replay applied it.

Register the listener before you subscribe. If you register it after a catch-up has begun, it misses that catch-up’s `catchupStarted` call.

`listenForCatchup` returns whether the model calls the listener at all. Occurrent’s own models on both stacks do. The interface’s default implementation returns `false`, so implement it on a subscription model of your own that catches up.

When `listenForCatchup` returns `false`, poll `isCatchingUp` instead, as the [read-your-writes section](/docs/spring-boot/projection-annotation/#read-your-writes-for-an-asynchronous-projection) describes. A catch-up that starts and finishes between two polls goes unnoticed.

When a subscription model refuses a call, the exception names the reason as a type. `subscribe(..)` throws `DuplicateSubscriptionIdException` for an id this model instance already has, `UnsupportedSubscriptionFilterException` for a filter shape it cannot apply, and `UnsupportedStartAtException` for a start position it cannot resolve. The life-cycle methods throw `SubscriptionAlreadyRunningException`, `SubscriptionNotRunningException` and `UnknownSubscriptionException` the same way. All six are sealed under `SubscriptionRefusedException`, and each carries what it refused, the subscription id or the start position, as a typed accessor, so a catch can act on the specific refusal instead of parsing a message. This holds on every subscription model on both stacks, so the answer no longer depends on which model you happen to be running against.

## Starting Subscriptions Manually

`ManualStartSubscriptionModel` in `occurrent-subscription-api-blocking` wraps any subscription model so that subscribing only registers a subscription, and nothing runs until you start it yourself. No framework is required. `ManualStartSubscriptionModel.stoppedByDefault(subscriptionModel)` withholds every subscription, then `start()` or `resumeSubscription(id)` brings one up. A subscription running for the first time starts from wherever the wrapped model defaults to at the moment it’s started, so events written between registration and that moment do not reach it. For a registration that resolves to the model’s own default start position, only the three-argument `stoppedByDefault(SubscriptionModel, GlobalCheckpointSource, CheckpointStorage)` overload records that position at registration instead of at start. A registration that names its own position, a replay or an explicit checkpoint, still writes nothing there. It is also what the JUnit extension’s [stopped-by-default testing](/docs/testing/integration-testing/#stopping-every-subscription-then-opting-in) is built on.

* Java

  ```java
  var manual = ManualStartSubscriptionModel.stoppedByDefault(subscriptionModel);


  // Registered but withheld, nothing is delivered yet
  manual.subscribe("current-orders", System.out::println);


  // Start just this one once its dependencies are ready
  manual.resumeSubscription("current-orders");


  // Or bring up everything still withheld
  manual.start();
  ```

* Kotlin

  ```kotlin
  val manual = ManualStartSubscriptionModel.stoppedByDefault(subscriptionModel)


  // Registered but withheld, nothing is delivered yet
  manual.subscribe("current-orders") { println(it) }


  // Start just this one once its dependencies are ready
  manual.resumeSubscription("current-orders")


  // Or bring up everything still withheld
  manual.start()
  ```

> **Using Spring Boot?**
>
> `occurrent.subscription.mode=manual` wraps the starter’s model in a `ManualStartSubscriptionModel`. See [What the starter configures](/docs/spring-boot/what-the-starter-configures/#manual-start).