Skip to content
OccurrentOccurrent

Subscription life-cycle

Layer 0: CloudEvents · 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 and a push subscription 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 is for.

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

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
  • 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, including using the subscription life cycle and 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 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 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

Section titled “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 is built on.

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();