# Subscription model capabilities

> What a subscription model can do, and how wrapping models combines their capabilities.

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

A `SubscriptionModel` is a small interface with four things on it, subscribe, cancel, pause, and stop. Everything else, listing subscriptions, knowing whether a replay is still running, resuming at an explicit position, wrapping another model, exposing a checkpoint, is an optional capability a concrete model may or may not add. This is a different idea from the event store’s own `EventStoreCapability` (`STREAM` and `DCB`, see [Capabilities](/docs/extending/your-own-event-store/#capabilities)), which describes what an `EventStore` supports rather than what a `SubscriptionModel` does.

Most of these capabilities are mixins. Instead of casting the model to check for one, you call a small method that looks for it and hands back an answer if it finds one, which matters because the model you’re holding is usually several wrappers deep:

* Java

  ```java
  // subscriptionModel might be a bare NativeMongoSubscriptionModel, or three layers of
  // wrapper deep, CompetingConsumer(Catchup(Durable(Mongo))). Either way, this reaches
  // whichever layer actually knows the answer.
  Set<String> ids = IntrospectableSubscriptions.findIn(subscriptionModel)
          .map(IntrospectableSubscriptions::subscriptionIds)
          .orElse(Set.of());
  ```

* Kotlin

  ```kotlin
  // subscriptionModel might be a bare NativeMongoSubscriptionModel, or three layers of
  // wrapper deep, CompetingConsumer(Catchup(Durable(Mongo))). Either way, this reaches
  // whichever layer actually knows the answer.
  val ids = IntrospectableSubscriptions.findIn(subscriptionModel)
      .map { it.subscriptionIds() }
      .orElse(emptySet())
  ```

## The base every subscription model has

| Interface                    | Blocking | Reactor | Adds                                                                                                                                                                                                                                 |
| :--------------------------- | :------- | :------ | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `Subscribable`               | yes      | yes     | `subscribe(...)`, the read-and-react primitive every model starts from                                                                                                                                                               |
| `CancellableSubscriptions`   | yes      | yes     | `cancelSubscription(id)` alone, split out for a register-only model that has nothing to start, stop, or pause. No shipped model uses it on its own today, every one implements the fuller `SubscriptionModelLifeCycle` below instead |
| `SubscriptionModelLifeCycle` | yes      | yes     | extends `CancellableSubscriptions` with `start`, `stop`, `pauseSubscription`, `resumeSubscription`, `isRunning`, `isPaused`, and `shutdown`                                                                                          |
| `SubscriptionModel`          | yes      | yes     | `Subscribable` + `SubscriptionModelLifeCycle` together, what “a subscription model” means on either stack                                                                                                                            |

## Wrapping, and the `findIn(..)` probe

A model that wraps another one (`DurableSubscriptionModel` around a Mongo model, `CatchupSubscriptionModel` around that, and so on) implements `SubscriptionModelWrapper` on the blocking stack, which exposes `getWrappedSubscriptionModel()` and a recursive variant that walks all the way down. This is blocking-only, reactor has no equivalent interface, so a reactor wrapper has no generic way to hand back what it wraps.

That asymmetry is why the two stacks reach an optional capability differently. On the blocking stack, `IntrospectableSubscriptions`, `ReplayAwareSubscriptions`, and `RepositionableSubscriptions` each have a `findIn(subscriptionModel)` method that checks the model you hand it first, and only unwraps a `SubscriptionModelWrapper` layer by layer if that model itself doesn’t implement the capability. It searches the whole wrapper chain and can come back with nothing, an `Optional`, not a guaranteed conversion, which is why it’s named `findIn(..)` rather than `of(..)`. Which means the **outermost** implementer answers, not the innermost one. Hold `CompetingConsumer(Catchup(Durable(Mongo)))`, and if every layer happens to implement `IntrospectableSubscriptions`, `findIn(..)` returns `CompetingConsumerSubscriptionModel`’s own answer, never Mongo’s. On the reactor stack there is no probe at all for `IntrospectableSubscriptions` or `ReplayAwareSubscriptions`, check the concrete model you’re holding with `instanceof` instead, and note that the subscription DSL wrappers don’t forward either capability, so ask the subscription model itself rather than a DSL wrapper around it.

## Optional mixins

| Interface                             | Blocking | Reactor | Adds                                                                                                                                                                                                                                                                                                                                                                                                         | Reach it with                                                                              |
| :------------------------------------ | :------- | :------ | :----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | :----------------------------------------------------------------------------------------- |
| `IntrospectableSubscriptions`         | yes      | yes     | `subscriptionIds()`, every id the model knows, running or paused                                                                                                                                                                                                                                                                                                                                             | `findIn(model)` (blocking) / `instanceof` (reactor)                                        |
| `ReplayAwareSubscriptions`            | yes      | yes     | `isCatchingUp(id)`, true only while that subscription is still replaying history, which `isRunning(id)` can’t tell you since it stays true throughout a replay                                                                                                                                                                                                                                               | `findIn(model)` (blocking) / `instanceof` (reactor)                                        |
| `RepositionableSubscriptions`         | yes      | yes     | `resumeSubscription(id, startAt)`, resume at an explicit position instead of wherever the model stopped                                                                                                                                                                                                                                                                                                      | `findIn(model)`                                                                            |
| `QuietPositionReportingSubscriptions` | yes      | yes     | `addQuietPositionListener(..)`, tells a listener the position a subscription has read to when a read returned no event for it, which a durable model [saves as the checkpoint](/docs/layer-0/durable-subscriptions/). The MongoDB models implement it                                                                                                                                                        | `findIn(model)`                                                                            |
| `HistoryLossReportingSubscriptions`   | yes      | yes     | `addHistoryLossListener(..)`, asks a listener what to do with a subscription once the oplog has dropped its position, which is how a catch-up takes the subscription over. On the blocking stack it also tells the listener the position the subscription restarts from, before the restart. `NativeMongoSubscriptionModel`, `SpringMongoSubscriptionModel` and `ReactorMongoSubscriptionModel` implement it | `findIn(model)`                                                                            |
| `SubscriptionModelWrapper`            | yes      | no      | `getWrappedSubscriptionModel()` / `...Recursively()`, the delegate a wrapper sits on, what `findIn(..)` walks through                                                                                                                                                                                                                                                                                        | declare or cast                                                                            |
| `CheckpointAwareSubscriptionModel`    | yes      | yes     | `globalCheckpoint()`, on the reactor stack also `globalCheckpointAsOfNow()`, cloud events with a checkpoint you can store yourself, and `canResumeFrom(checkpoint)`, see below                                                                                                                                                                                                                               | declare the variable as this type, or cast, no `findIn(..)` probe, see below               |
| `Pushable`                            | yes      | yes     | `accept(cloudEvent)` / `accept(events)`. An `InMemoryEventStore` listener calls `accept(events)`, and a broker listener calls `PushSubscriptionModel.acceptOrDefer(cloudEvent)` instead, which `Pushable` doesn’t declare                                                                                                                                                                                    | implemented directly by the blocking and reactive `PushSubscriptionModel`, declare or cast |

`canResumeFrom(checkpoint)` returns `false` when the model no longer has the history back to that checkpoint, for example a MongoDB change-stream position older than the oldest entry in the oplog. It returns a `boolean` on blocking and a `Mono<Boolean>` on reactor, and returns `true` unless the model overrides it. A position catch-up, and the blocking time-based catch-up, call it when they resume from a stored checkpoint that holds a live start, before the resumed replay, and again once the replay is done, before it goes live.

`CheckpointAwareSubscriptionModel` and `Pushable` aren’t reachable through a wrapper-unwrapping probe the way the mixins reached with `findIn(model)` are. Nothing wraps a model and re-exposes checkpoint-awareness on demand. You get it because the concrete model you constructed implements it directly, so keep hold of that static type or cast to it. Those models are `NativeMongoSubscriptionModel`, `SpringMongoSubscriptionModel`, `DurableSubscriptionModel`, and their reactor equivalents.

## Typed views

| Interface                 | Blocking | Reactor              | Accepts                         | Reach it with                                                                                                                  |
| :------------------------ | :------- | :------------------- | :------------------------------ | :----------------------------------------------------------------------------------------------------------------------------- |
| `StreamSubscriptionModel` | yes      | no                   | `Filter`, `StartAt`             | `StreamSubscriptionModel.from(delegate)`                                                                                       |
| `DcbSubscriptionModel`    | yes      | yes, different shape | `DcbCriteria`, `DcbStartAt`     | `DcbSubscriptionModel.from(delegate)`                                                                                          |
| `FluxSubscriptionModel`   | n/a      | yes                  | `SubscriptionFilter`, `StartAt` | the reactor stack’s bare primitive, `subscribe(..)` returns a `Flux<CloudEvent>` directly rather than a tracked `Subscription` |

These aren’t optional mixins, everyone gets one, they’re translation layers you build explicitly around any `SubscriptionModel`. `Subscribable`’s generic `subscribe(..)` only accepts the marker type `SubscriptionFilter`. `StreamSubscriptionModel.from(delegate)` narrows that to the concrete `Filter` type on the blocking stack, and `DcbSubscriptionModel.from(delegate)` narrows it to `DcbCriteria`/`DcbStartAt` on both stacks. Reactor never needed a stream-typed facade to match, its bare `FluxSubscriptionModel` primitive already takes a `Filter` directly, since `Filter` implements `SubscriptionFilter`. The reactor `DcbSubscriptionModel` is a different shape from its blocking counterpart though. Its unnamed `subscribe(criteria, startAt)` returns a bare `Flux<CloudEvent>` the way `FluxSubscriptionModel` does, and only the named, life-cycle-managed `subscribe(id, criteria, startAt, action)` and `cancelSubscription(id)` require the delegate to also implement `Subscribable` and `SubscriptionModelLifeCycle`, checked only when one of those methods is actually called.

## Checkpoint fencing

`CheckpointWriteVersionSource`, blocking only, supplies the version a checkpoint-writing model stamps its own writes with. A model asks it right before every checkpoint write. A version that comes back becomes a not-older-than write condition, no version or no source at all becomes an unconditional write, so the model has one code path instead of choosing between two. It says nothing about leases or competing consumers itself. Wiring `CompetingConsumerCheckpointWriteVersionSource` (backed by `CompetingConsumerStrategy::fencingToken`) into `DurableSubscriptionModel` and `CatchupSubscriptionModel` connects the two, so a node that lost its lock and kept writing can’t overwrite a checkpoint a faster node already advanced, see [checkpoint fencing](/docs/layer-0/checkpoint-fencing/).

## `SubscriptionModelCapability`

0.33.0 adds a root marker interface, `SubscriptionModelCapability`, on both stacks, and every capability facet on this page extends it. That’s also the release where the method shown throughout this page is renamed from `of(Object)` to `findIn(SubscriptionModelCapability)`, the two changes ship together. The narrower parameter type doesn’t change what compiles at an existing call site (a `SubscriptionModel` reference still passes straight through), it only stops you probing something that was never part of this hierarchy in the first place.

## Asking the model instead of naming a facet

`SubscriptionModelCapability` also carries two default methods, `capability(type)` and `hasCapability(type)`, so any subscription model can answer for a capability handed to it as a `Class` rather than named directly in a static call:

* Java

  ```java
  Optional<IntrospectableSubscriptions> found = subscriptionModel.capability(IntrospectableSubscriptions.class);
  boolean canIntrospect = subscriptionModel.hasCapability(IntrospectableSubscriptions.class);
  ```

* Kotlin

  ```kotlin
  val found: IntrospectableSubscriptions? = subscriptionModel.capability<IntrospectableSubscriptions>()
  val canIntrospect = subscriptionModel.hasCapability<IntrospectableSubscriptions>()
  ```

Use `capability`/`hasCapability` when the `Class` you’re checking is itself a value, chosen by a caller further up the stack rather than written into the code doing the check. Use `findIn` when the facet is fixed at the call site, since `IntrospectableSubscriptions.findIn(model)` names both the search and its target in one call.

`capability(type)` runs the same search as `IntrospectableSubscriptions.findIn(model)` above, for whichever `Class` you pass it instead of one fixed facet. On the blocking stack it unwraps a `SubscriptionModelWrapper` chain until something implements `type`. On the reactor stack, which has no wrapper to unwrap, it checks the model itself.

Of the reactor models Occurrent ships, only `ReactorCatchupSubscriptionModel` and `ReactorStreamCatchupSubscriptionModel` behave differently, and only for `QuietPositionReportingSubscriptions`. For that type they return the capability of the model they wrap. `hasCapability(type)` gives the same answer as a `boolean`, for a caller that only needs to know whether the capability is there.

The Kotlin extensions come from the same [Subscription DSL](/docs/layer-1/subscribing-to-domain-events/) module as `streamSubscriptions`/`subscriptions`, on both stacks, as `capability<T>()` and `hasCapability<T>()`, inferring `T` from the type argument instead of taking a `Class`.

## Does composition order matter?

Three different questions hide behind “does order matter”, and they don’t share an answer.

Order does not matter for interfaces you declare on your own model. `class MySubscriptionModel implements SubscriptionModel, IntrospectableSubscriptions, ReplayAwareSubscriptions` behaves identically no matter which order you list the interfaces in. Java doesn’t care, and neither does anything in Occurrent that inspects the result.

Order does matter when composing wrappers, and the constraint is mostly the type system. Each wrapper’s constructor declares what it needs its delegate to already be. `DurableSubscriptionModel` and `CatchupSubscriptionModel` (and their reactor equivalents) both require a `CheckpointAwareSubscriptionModel` delegate, so each has to sit directly on something that already is one, the raw Mongo model, or another wrapper that also implements it (`DurableSubscriptionModel` does, which is what lets `CatchupSubscriptionModel` wrap it). `CompetingConsumerSubscriptionModel` and `ManualStartSubscriptionModel` only require a plain `SubscriptionModel`, so they compose further out, and in fact can’t compose any other way: `CompetingConsumerSubscriptionModel` doesn’t implement `CheckpointAwareSubscriptionModel`, so `CatchupSubscriptionModel` can’t wrap it even if you wanted it to.

Not every constraint is type-checked though. `ManualStartSubscriptionModel` has to be the outermost wrapper, and its own documentation states directly why. Withholding a subscription only works if nothing beneath it has already been handed the registration. Put it anywhere but outermost, and a catch-up layer below it would replay history to a handler nobody started yet. So wrap outward from whichever layer needs the most from its delegate (checkpoint-awareness first), then whatever only needs a plain `SubscriptionModel`, and put anything that has to see every subscription before the rest of the stack does, like `ManualStartSubscriptionModel`, last.

The outermost layer answers a probe. This is the same rule as the `findIn(..)` probe described above, restated here because it’s the detail that actually surprises people composing wrappers. Build `CompetingConsumer(Catchup(Durable(Mongo)))` and ask `ReplayAwareSubscriptions.findIn(..)` on the whole stack: `CompetingConsumerSubscriptionModel` doesn’t implement `ReplayAwareSubscriptions` itself, so the probe unwraps one layer and asks `CatchupSubscriptionModel`, which does, and that’s the answer you get, not whatever `Durable` or `Mongo` underneath it would have said. The first layer in the chain that implements the capability wins, working from the outside in.

> **Using Spring Boot?**
>
> [What the starter configures](/docs/spring-boot/what-the-starter-configures/#how-the-subscription-model-is-built) shows the stack of models the starter builds.