Push subscription

Layer 0: CloudEvents · Blocking

Use a push subscription when your application receives its events from a message broker, such as RabbitMQ or Kafka, instead of reading them from the event store. The application that writes the events forwards them to the broker, and a listener in your application receives them.

org.occurrent.subscription.push.blocking.PushSubscriptionModel never reads from an event store. Your listener hands it each event, and it passes the event to every registered subscription whose filter matches. It has no start position, checkpoint or replay of its own.

First include the dependency:

<dependency>
    <groupId>org.occurrent</groupId>
    <artifactId>occurrent-subscription-push-blocking</artifactId>
    <version>0.34.0</version>
</dependency>

Because it matches entirely in memory, new PushSubscriptionModel() refuses a subscription filter on the event data payload. Pass a DataFieldReader to its constructor to give it one, see Filtering on Payload Data.

Because PushSubscriptionModel implements the same Subscribable interface as every other subscription model, it plugs into the Subscription DSL, and into ProjectionRunner from Occurrent’s projection-dsl module, unchanged:

PushSubscriptionModel pushModel = new PushSubscriptionModel();

ProjectionRunner.agnostic(pushModel, cloudEventConverter)
        .project("order-status", orderStatusProjection(), repository);

A PushSubscriptionModel takes one projection or saga. Registering a second one on the same model fails at startup, and the error names both.

The reason is that your listener acknowledges each broker message once, so every projection on the model would have to share that one acknowledgement. If one of them kept failing on a message, the broker would keep sending that message to all of them. Once the broker gave up on it, none of them would get it.

So create one PushSubscriptionModel per projection, and give each its own queue, subscription or consumer group on the broker. A message that one projection can’t handle then holds up only that projection.

This limit applies to PushSubscriptionModel only. A subscription model that reads from the event store serves any number of subscriptions, since each one keeps track of its own position.

On the producer side, forward the stored CloudEvent to the broker as CloudEvents JSON, unchanged. On the listener side, reconstruct it from that CloudEvents JSON payload, hand it to acceptOrDefer(CloudEvent) and act on the RoutingOutcome it returns. In the example below, acknowledge(), redeliverLater(), applyFailurePolicy() and stopConsuming() stand for the calls your broker client makes:

void onMessage(byte[] body) {
    CloudEvent cloudEvent = EventFormatProvider.getInstance()
            .resolveFormat(JsonFormat.CONTENT_TYPE)
            .deserialize(body);
    switch (pushModel.acceptOrDefer(cloudEvent)) {
        case DELIVERED, FILTERED -> acknowledge();
        case DEFERRED, UNAVAILABLE -> redeliverLater();
        case NOT_DELIVERABLE -> applyFailurePolicy();
        case REFUSED -> stopConsuming();
    }
}

acceptOrDefer(CloudEvent) runs the registered handler synchronously, on the calling thread, when the event matches its filter. If the handler or the filter throws, the exception propagates back to your listener. Handle it the same way as NOT_DELIVERABLE. Consuming from a broker says what to do with each outcome.

redeliverLater() has to hand the message back to the broker, since not acknowledging it doesn’t get it delivered again.

Feeding it from a broker

Observing pushed events (Added in 0.34.0)

accept(..) returns normally for an event that no subscription takes, and it logs nothing about it. So each of these looks exactly like a saga or projection that got the event and ignored it:

a misconfigured queue binding
a missing declared event type
a typo in a type mapping

Pass a PushObserver to the constructor to tell them apart:

PushSubscriptionModel pushModel = new PushSubscriptionModel(DataFieldReader.refusing(),
        (cloudEvent, outcome) -> {
            if (outcome != RoutingOutcome.DELIVERED) {
                log.warn("Event {} of type {} was not delivered: {}", cloudEvent.getId(), cloudEvent.getType(), outcome);
            }
        });

accept(..) doesn’t throw for an event that no subscription takes, because the in-memory event store calls it too, as the listener you pass to an InMemoryEventStore. By then the event is already stored, so throwing would only make the write call fail, without undoing the write. ADR 104 has the reasoning, and the amendment to ADR 133 says why the in-memory event store is the only write path this covers.

Pass your own DataFieldReader instead of DataFieldReader.refusing() if the subscription’s filter also needs to match on fields inside the event’s data.

The observer is called once for each event, after the subscription’s handler has run, or straight away when no handler runs. It’s called whether the handler succeeded or threw.

outcome is a RoutingOutcome with six values. It tells “the filter declined this event” apart from “nothing was there to receive it”.

Several of the outcomes involve a CatchupThenPushSubscriptionModel in front of the handler. That’s the wrapper that replays the event store before it delivers live events, and it’s described further down in this section.

Outcome	What happened
`DELIVERED`	The subscription’s filter accepted the event and the handler ran. Behind a `CatchupThenPushSubscriptionModel`, it can also mean the event was buffered for the replay to deliver, or had already been delivered. When the handler throws, the outcome is still `DELIVERED`, and the handler’s exception propagates out of `accept(..)` once the observer has been told.
`FILTERED`	The subscription’s filter looked at the event and declined it. Offering it to the same subscription again gets the same answer.
`UNAVAILABLE`	No filter was asked, because nothing is registered, the model is stopped, or the subscription is paused. Nothing is thrown.
`DEFERRED`	The filter accepted the event, but the `CatchupThenPushSubscriptionModel` in front of the handler didn’t deliver it. That happens, for example, when its catch-up was stopped, or when a broker listener called `acceptOrDefer(..)`, described below, before the replay finished. Nothing is thrown, and offering the event again later is safe.
`NOT_DELIVERABLE`	The filter threw instead of answering, or the `CatchupThenPushSubscriptionModel` in front of the handler refused the event, for example because its live buffer is full while the replay is still running. The exception propagates out of `accept(..)` either way.
`REFUSED`	The `CatchupThenPushSubscriptionModel` in front of the handler refused the event because its replay failed, and it refuses every event from then on. The exception propagates out of `accept(..)`.

Don’t base a broker acknowledgement on what the observer is told. The observer is told DELIVERED even when the handler threw, and, behind a CatchupThenPushSubscriptionModel, when accept(..) only buffered the event. Base it on the outcome that acceptOrDefer(..) returns, described below. Consuming from a broker says what to do with each outcome.

A broker listener that can redeliver a message should call acceptOrDefer(CloudEvent) instead of accept(..). It routes the event the same way, except that a CatchupThenPushSubscriptionModel in front that hasn’t reached live delivery yet refuses the event instead of buffering it, and reports DEFERRED.

Plain accept(..) buffers that event and reports DELIVERED before the handler has applied it. That’s right for the in-memory event store, because its listener never offers the same event a second time. When the buffer is full it throws instead, and while the catch-up is stopped it drops the event and reports DEFERRED.

The reported outcome is what the routing did with the event, even when a stop(), a pause or a resume happens at the same moment. The report reuses the routing’s own filter check and never checks again whether the model is running or the subscription is paused.

The table below shows what the observer is told, and what accept(..) throws, when the filter, the handler or the observer itself throws.

Thrower	What it throws	Outcome the observer is told	What `accept(..)` throws
Filter	`RuntimeException` or `AssertionError`	`NOT_DELIVERABLE`	The filter’s exception, once the observer has been told
Filter	A checked exception, which Kotlin code can throw without declaring it, or an `Error` other than `AssertionError`	None, the observer isn’t called	The filter’s exception
Handler	`RuntimeException` or `AssertionError`	`DELIVERED`	The handler’s exception, once the observer has been told
Handler	A checked exception, which Kotlin code can throw without declaring it, or an `Error` other than `AssertionError`	None, the observer isn’t called	The handler’s exception
Observer	Any `Exception`, checked ones included, or an `AssertionError`	Whichever outcome it was being told	Only the model’s own failure, if it has one. The observer’s exception is caught and logged.
Observer	An `Error` other than `AssertionError`	Whichever outcome it was being told	The model’s own failure for that event, from the filter, the handler or a catch-up refusal, with your `Error` attached to it with `addSuppressed(..)`. With no such failure, your `Error` itself.

A filter can throw a RuntimeException or an AssertionError when the DataFieldReader you gave the model fails to read a field.

The model catches an Exception or AssertionError from the observer. So an observer that throws one can’t make the broker redeliver an event that was delivered, and doesn’t stop the rest of a batch.

If your observer throws an InterruptedException, the model sets the interrupt flag on the calling thread again.

The two constructors that take no observer use PushObserver.noop(). With it, accept(..) routes each event without working out a RoutingOutcome at all.

accept(Iterable<CloudEvent>) stops the batch at the first event where routing throws, whether its handler threw, its filter threw, or a CatchupThenPushSubscriptionModel in front refused it. The events after it are neither routed nor observed. An Error other than AssertionError from the observer stops the batch the same way.

No broker dependency is added by this module, you pick and wire up RabbitMQ, Kafka, or anything else yourself. The CloudEventConverter.toDomainEvent(...) call inside the projection runner needs the extension attributes your handlers rely on, so make sure the pushed CloudEvent carries at least streamid and streamversion, and position too if something downstream (such as a catch-up model) reads it.

Feeding the PushSubscriptionModel straight from an event store’s writes is supported only for the in-memory event store. Its listener calls accept(Iterable<CloudEvent>), and there a crash before the handler has run loses the event from the store too.

The PushSubscriptionModel keeps no record of which events a subscription has handled. So with a durable event store, such as MongoDB, the subscription never sees an event if the application crashes after the write has committed but before the handler has run. Use a durable subscription there, or forward the events to a broker and call acceptOrDefer(CloudEvent) from its listener.

When a broker feeds the model, don’t acknowledge a message just because accept(..) returned. The broker never sends an acknowledged message again, and accept(..) returns normally in cases where the handler hasn’t applied the event:

for an event no subscription takes
while the model is stopped or the subscription is paused
behind a CatchupThenPushSubscriptionModel, before the handler has applied the event, or without the handler ever applying it

Call acceptOrDefer(CloudEvent) instead. It returns the event’s RoutingOutcome, and you acknowledge the message only when that is RoutingOutcome.DELIVERED or RoutingOutcome.FILTERED. Consuming from a broker says what to do with the other outcomes. Here is when it returns them:

UNAVAILABLE for an event no subscription takes, and while the model is stopped or the subscription is paused.
DEFERRED, with a CatchupThenPushSubscriptionModel in front, for an event that arrives before that model has gone live, while its catch-up is stopped, or while an earlier delivery of the same event is still in progress on another thread.
REFUSED for every event once a catch-up in front has failed.
NOT_DELIVERABLE, with a CatchupThenPushSubscriptionModel in front, for an event whose getId() is null, since that model tells events apart by id. Handle it the same way as a handler that throws.

The RabbitMQ and Kafka sources do all of this for you.

A handler that feeds another PushSubscriptionModel from inside its own subscription has to act on what acceptOrDefer(..) returns, because nothing redelivers an event that the other model refuses. For example, throw on anything but DELIVERED or FILTERED, so the outer handler fails instead of returning as if the event had been handled. Calling accept(..) there is no safer, since it also returns normally for an event no subscription takes, and while the model is stopped or the subscription is paused.

Replaying the event store before going live

A push subscription only ever sees the live tail. A broker is not a log, so a new or rebuilt projection can’t be backfilled from the queue. Replay history from the event store first, with EventStore Queries or a catch-up subscription, and only then attach the push feed to keep the projection current.

CatchupThenPushSubscriptionModel automates that catch-up. Wrap it around the PushSubscriptionModel and give it the event store as the replay source.

On the first subscribe it replays the projection’s history in position order. Then it hands over to the live feed, and that switch from replay to live delivery is the handover. During the replay, acceptOrDefer(..) returns DEFERRED, so the broker delivers the event again, and accept(..) buffers it.

The handover de-duplicates the overlap by event id and source together. So nothing is lost or delivered twice across it, within the two limits described below:

PushSubscriptionModel pushModel = new PushSubscriptionModel();
CatchupThenPushSubscriptionModel model = new CatchupThenPushSubscriptionModel(eventStore, pushModel, checkpointStorage);

ProjectionRunner.agnostic(model, cloudEventConverter)
        .project("order-status", orderStatusProjection(), repository);

Two settings control the handover, and both have defaults.

The first, dedupCacheSize, sets how many recently delivered events the handover remembers by id and source. That’s how an event that both the replay and the live feed deliver reaches your handler only once. The handover keeps two of these caches. One holds events the replay delivered, and one holds events delivered live. dedupCacheSize is the size of each.

At the default of 10000 the two caches together hold up to 20000 events, twice the number you set.

If the replay and the live feed overlap by more than dedupCacheSize events, an event can reach your handler twice, so make sure that applying an event a second time doesn’t change your read model.

The second, maxBufferedEvents, caps the live events buffered during the replay, at 100000 by default. It is a fail-loud ceiling rather than a throttle. Only accept(..) fills it, since acceptOrDefer(..) refuses an event during the replay instead of buffering it.

When the buffer is full, the in-memory event store’s write call throws, even though the event is already stored. The PushSubscriptionModel’s accept(events) stops at that event and doesn’t route any of the later events from the same write either, so the subscription can miss all of them. Pass CatchupThenLiveOptions to change either:

CatchupThenPushSubscriptionModel model = new CatchupThenPushSubscriptionModel(
        eventStore, pushModel, checkpointStorage, new CatchupThenLiveOptions(50_000, 200_000));

Once the handover is live, the handler can be called concurrently. Handler calls are no longer serialized behind the handover’s own lock, so a listener container configured with more than one delivering thread runs your handler on all of them at once, and the handler must tolerate that. A single-threaded listener, the common case, sees no change. The reasoning, including the measured throughput this buys, is in ADR 0108.

The default projection materializer has no locking of its own, so once the handler can run concurrently, two threads handling different events for the same projection key can both read the same state, and the second save then overwrites the first without telling you. Pass a RetryStrategy to Projections.materializedView(..), or to Projections.domainEventFeed(..) for the domain-event variant below, so a losing write re-reads and re-applies instead of being overwritten. That only helps once your store detects the conflict and throws, an optimistic-locking failure or a unique-key violation, since a retry that never sees an exception has nothing to retry. Keeping the delivering side single-threaded, the default, is the other way to avoid this.

Live-resume stays the broker’s job. The model persists no live position watermark. It only records that the catch-up finished, in the checkpointStorage you pass. That record is the catch-up marker. A restart that finds it skips the replay and lets the broker redeliver whatever the consumer had not yet acknowledged. Pass null and it records nothing, so every restart replays the history.

Delivery is therefore at-least-once, so the projection must tolerate seeing the same event twice. Correctness across a restart depends on the broker retaining the backlog for an offline consumer (a durable queue with a preserved offset). If the consumer is offline longer than the broker retains, rebuild the projection.

Only stream and capability-agnostic subscriptions can catch up this way.

The catch-up marker is kept per subscription id. The model’s cancelSubscription(id) deletes it, so subscribing the same id again replays the history.

On the blocking stack cancelSubscription(id) deletes the marker before it cancels anything else. If the delete throws, the subscription stays as it was, paused or running, and you can call cancelSubscription(id) again.

On the reactor stack the subscription is cancelled right away, and the Mono<Void> that cancelSubscription(id) returns completes once the marker is deleted. Wait for it when a restart must not skip the history, and call cancelSubscription(id) again if it fails.

Your own code starts the catch-up, so it also sees how it ends. On the blocking stack CatchupProjectionFeed.catchUp() returns when the replay has finished and throws when it fails. On the reactor stack the Mono it returns completes or errors instead. The blocking DomainEventFeed can also be asked at any time:

boolean readyToServe() {
    if (feed.refusesPermanently()) {
        throw new IllegalStateException("The order-status catch-up failed");
    }
    return feed.isReadyForLiveDelivery();
}

isReadyForLiveDelivery() is true once the feed is live, and refusesPermanently() is true once its catch-up has failed. A replay that’s still running answers false to both.

Life-cycle (Added in 0.32.0)

Like every other subscription model, a PushSubscriptionModel can be stopped, started, paused per subscription, cancelled and shut down. There’s no feed behind it holding events back, so an event handed to accept(..) while the model is stopped, or a subscription is paused, never reaches that handler, and resuming does not replay it. That’s dropped, not deferred. shutdown() is one-way. A model that’s been shut down stays that way. This is the same contract SynchronousSubscriptionModel runs on, and it’s what lets a test stop every subscription model in the application and opt back in per test, a PushSubscriptionModel included.

CatchupThenPushSubscriptionModel gained the same full life cycle in 0.32.0, on both stacks, and stopping it now also interrupts a catch-up replay that’s already running rather than only the live feed behind it. Starting it again replays the interrupted history from the beginning, and start(false) leaves that replay for resumeSubscription(..) to pick up on its own.

Feeding domain events instead of CloudEvents (Layer 1)

If your listener already hands you domain events, for example because a broker message converter deserializes them for you, a projection that takes CloudEvents needs each event turned into a CloudEvent and back into a domain event. That’s an extra serialize and deserialize for every event. Pass the domain events to the projection directly instead, and nothing is converted between your listener and the read model.

Projections.domainEventFeed(projection, repository) is the live-only feed. On the blocking stack it returns a MaterializedView<E>, call update(event) or update(metadata, event). On the reactor stack it returns a BiFunction<EventMetadata, E, Mono<Void>>. Either way it applies a domain event straight into the read model:

MaterializedView<OrderEvent> feed = Projections.domainEventFeed(orderStatusProjection(), repository);

@RabbitListener(queues = "orders")
public void onMessage(OrderEvent event) {
    feed.update(event);   // no CloudEvent conversion
}

For a new or rebuilt projection that also needs to catch up, use CatchupProjectionFeed. Live events are still applied as domain events, and only the one-time catch-up reads the event store and converts each replayed event once.

CatchupProjectionFeed.create(...) takes the feed’s id, the projection, its repository, the event store, the CloudEvent converter, a function that returns an event’s id, and the CheckpointStorage that records that the catch-up finished. Your listener passes each event to acceptAndWait(event), called accept(event) before 0.34.0, or to acceptAndWait(metadata, event) when the message also includes the stream id, version or position. catchUp() replays the projection’s history and then goes live, or goes live right away when the CheckpointStorage records that an earlier catch-up finished. On the blocking stack it runs on the thread that calls it, so call it from a thread other than the listener’s:

Java
Kotlin

CatchupProjectionFeed<OrderEvent> feed = CatchupProjectionFeed.create(
        "order-status", orderStatusProjection(), repository,
        eventStore, cloudEventConverter, OrderEvent::eventId, checkpointStorage);

@RabbitListener(queues = "orders")
public void onMessage(OrderEvent event) {
    feed.acceptAndWait(event);   // returns once the event is applied
}

// Once the listener is wired, on a thread other than the listener's
void startFeed() {
    feed.catchUp();
}

val feed = CatchupProjectionFeed.create(
    "order-status", orderStatusProjection(), repository,
    eventStore, cloudEventConverter, OrderEvent::eventId, checkpointStorage)

@RabbitListener(queues = ["orders"])
fun onMessage(event: OrderEvent) {
    feed.acceptAndWait(event)   // returns once the event is applied
}

// Once the listener is wired, on a thread other than the listener's
fun startFeed() {
    feed.catchUp()
}

Live events during the catch-up

catchUp() starts holding live events before it reads the history. An event passed to acceptAndWait(...) during the replay waits in memory, and acceptAndWait(...) doesn’t return, or its Mono doesn’t complete, until the event is applied. Once the last replayed event is applied, the feed applies the held events in the order they arrived and then goes live. The same holds for a catch-up started on a feed that’s already live.

An event can arrive both ways, once from the replay and once from the listener. The feed compares the ids your id function returns, OrderEvent::eventId above, and skips a live event whose id the replay already applied. acceptAndWait(...) then returns normally, so the listener acknowledges the message as usual. The feed remembers the last 10000 replayed ids, so a copy that arrives after the feed went live is skipped too. An older copy is applied again, so applying an event twice must leave the read model unchanged.

The feed holds at most 100000 events waiting to be applied, the maxBufferedEvents of CatchupThenLiveOptions. Once that many are waiting, it refuses the next one.

On the blocking stack only events that arrive while the feed isn’t live count toward that limit, because a live event is applied on the calling thread. On the reactor stack every event the feed has taken in and not yet applied counts, live or not.

The 10000 is a separate number, dedupCacheSize, and it doesn’t limit how many events wait. It sizes two lists of ids, one for the events the replay applied and one for those applied live, so the feed also skips a live event it already applied live, such as a message the broker delivered again. Pass a CatchupThenLiveOptions to create(...) to change either number.

Occurrent’s own RabbitMQ and Kafka modules call DomainEventFeed.acceptOrDeferCloudEvent(...) rather than acceptAndWait(...), because a broker can deliver a message again. That call never holds an event, and the modules stop fetching messages until the feed is live. A message that arrives anyway gets DEFERRED back and goes back to the broker unacknowledged, so no event waits in memory.

Both modules hand such a message back to the broker, the RabbitMQ module at most once per poll interval.

A listener you write yourself can do the same with acceptOrDefer(...). It doesn’t wait for the catch-up. Once the feed is live it applies the event and returns DELIVERED.

Before that, and while a replay runs, acceptOrDefer(...) returns DEFERRED right away and keeps the event nowhere. Don’t acknowledge the message then, and hand it back to the broker so it’s delivered again.

acceptAndWait(...) waits until the event is applied, so a caller that has nothing to deliver the event again needs it. A broker listener can call it too, for as long as the broker lets a message stay unacknowledged, covered in when acceptAndWait(...) returns.

On the reactor stack, once the feed is live, the Mono that acceptOrDefer(...) returns errors with an IllegalStateException when 100000 events are already waiting to be applied, so hand the message back to the broker then too. It errors the same way when the projection itself calls it, or a callback such as replayStarted() does. Call acceptAndWait(...) there instead, as feeding events from inside the projection shows. The blocking acceptOrDefer(...) never reaches that limit, since it applies a live event on the calling thread.

When `acceptAndWait(...)` returns

Called from a listener, acceptAndWait(...) returns, or its Mono completes, only once the event has been applied. So a listener that acknowledges the message when acceptAndWait(...) returns never acknowledges an event that is only held in memory.

A call from inside the projection, or from a callback such as replayStarted() that the feed calls on the projection, is different on the reactor stack. Its Mono completes once the event is queued, as feeding events from inside the projection shows.

A listener that acknowledges RabbitMQ messages itself looks like this:

Java
Kotlin

void onDelivery(Channel channel, long deliveryTag, OrderEvent event) throws IOException {
    try {
        feed.acceptAndWait(event);
        channel.basicAck(deliveryTag, false);
    } catch (RuntimeException e) {
        // Not applied, so RabbitMQ puts the message back on the queue
        channel.basicNack(deliveryTag, false, true);
    }
}

fun onDelivery(channel: Channel, deliveryTag: Long, event: OrderEvent) {
    try {
        feed.acceptAndWait(event)
        channel.basicAck(deliveryTag, false)
    } catch (e: RuntimeException) {
        // Not applied, so RabbitMQ puts the message back on the queue
        channel.basicNack(deliveryTag, false, true)
    }
}

When the event is not applied, acceptAndWait(...) throws an IllegalStateException, or its Mono errors with one. That happens when:

the catch-up was stopped before the feed went live
the event was fed after such a stop and before the next catch-up
the catch-up failed
the feed already holds 100000 events waiting to be applied, or the number set in CatchupThenLiveOptions

On the blocking stack it also happens when:

the waiting thread was interrupted
another delivery of the same event was still running
acceptAndWait(...) was called from inside the projection, a view or another callback of the same feed while the feed wasn’t live, since the thread would then wait for work it is holding up itself

In that case, don’t acknowledge the message. Hand it back to the broker instead, with basicNack and requeue as in the example above, or on Kafka as handing a message back to the broker describes.

A long replay keeps the listener waiting. A Kafka consumer that waits past its max.poll.interval.ms, 5 minutes by default, is taken out of its group and the record is delivered again. RabbitMQ delivers a message again once a consumer has held on to it without acknowledging it for longer than consumer_timeout, 30 minutes by default. In both cases the event is delivered again rather than lost.

Feeding events from inside the projection

A projection or a view can call its own feed, for example to pass in an OrderReadyToShip for each OrderPaid it handles.

The feed doesn’t store an event fed this way. It only applies it to the projection. A replay reads the event store, so it never sees that event, and a read model rebuilt from history doesn’t have it.

So for an event the read model has to keep, do one of these instead:

Apply its effect in the projection itself, when only this read model needs it.
Write it to the event store through your application service, when other parts of the system should see it too. It then reaches the feed like any other event.

If you do call the feed from the projection, the two stacks behave differently.

On the blocking stack, once the feed is live, the event is applied on the same thread before acceptAndWait(...) returns. During a replay, acceptAndWait(...) throws an IllegalStateException instead:

If the exception leaves the projection, the catch-up fails. A failed catch-up writes no catch-up marker, the record that the catch-up finished, so the next start replays the same history and fails the same way.
If the projection catches the exception, the event isn’t applied.

The projection can’t ask the feed whether a replay is running. A view that implements ReplayAware is told when a replay starts and ends, so it can skip the call during a replay, but then a read model rebuilt by a replay lacks every event it skipped.

On the reactor stack, acceptAndWait(...) called from inside the projection, or from a callback such as replayStarted(), isn’t refused. Its Mono completes once the event is queued rather than once it’s applied. The feed applies one event at a time, so the new event is applied after the projection or the callback returns, in the order it was fed. An event fed during the replay is applied after the replayed history, before the catch-up marker is written.

On the reactor stack, when the projection or one of its callbacks calls acceptOrDeferCloudEvent(...) on its DomainEventFeed with an event that matches the feed’s filter, the Mono errors with an IllegalStateException. Call acceptAndWaitCloudEvent(...) there instead. Its Mono completes once the event is queued.

The feed recognizes the call as coming from the projection when the projection returns it as part of its own Mono, or subscribes to it on the thread the feed called the projection on, for example by blocking on it. If the projection switches to another thread and blocks on the call there, it waits forever. In the example below, the function the feed calls for each event returns the call as part of its own Mono:

Java
Kotlin

Function<OrderEvent, Mono<Void>> orderStatus = Projections.reactiveUpdate(orderStatusProjection(), repository);
Function<OrderEvent, Mono<Void>> fold = event -> {
    Mono<Void> applied = orderStatus.apply(event);
    if (event instanceof OrderPaid paid) {
        // Part of the Mono the fold returns, so the feed knows the call comes from the projection
        return applied.then(feed.acceptAndWait(new OrderReadyToShip(paid.orderId())));
    }
    return applied;
};
feed = CatchupProjectionFeed.create("order-status", fold, Filter.all(),
        eventStore, cloudEventConverter, OrderEvent::eventId, checkpointStorage);

val orderStatus = Projections.reactiveUpdate(orderStatusProjection(), repository)
val fold = Function<OrderEvent, Mono<Void>> { event ->
    val applied = orderStatus.apply(event)
    if (event is OrderPaid) {
        // Part of the Mono the fold returns, so the feed knows the call comes from the projection
        applied.then(feed.acceptAndWait(OrderReadyToShip(event.orderId())))
    } else {
        applied
    }
}
feed = CatchupProjectionFeed.create("order-status", fold, Filter.all(),
    eventStore, cloudEventConverter, OrderEvent::eventId, checkpointStorage)