Subscription model capabilities

Layer 0: CloudEvents · 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), 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

// 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

// 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. 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.

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

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

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 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 shows the stack of models the starter builds.