# Push subscription

> Feed events from a broker or any other source into a subscription model.

[Layer 0: CloudEvents](/docs/layer-0/) · 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:

```xml
<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](/docs/layer-0/eventstore-queries/#filtering-on-payload-data).

Because `PushSubscriptionModel` implements the same `Subscribable` interface as every other subscription model, it plugs into the [Subscription DSL](/docs/layer-1/subscribing-to-domain-events/), and into `ProjectionRunner` from Occurrent’s `projection-dsl` module, unchanged:

```java
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:

```java
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](/docs/stores/brokers/#consuming-from-a-broker) says what to do with each outcome.

`redeliverLater()` has to [hand the message back to the broker](/docs/stores/brokers/#handing-a-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:

```java
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](https://github.com/johanhaleby/occurrent/blob/main/doc/architecture/decisions/0104-an-undeliverable-push-event-is-refused-not-acknowledged.md) has the reasoning, and the amendment to [ADR 133](https://github.com/johanhaleby/occurrent/blob/main/doc/architecture/decisions/0133-a-broker-is-a-transport-for-the-push-feed-and-never-a-subscription-model.md) 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](/docs/stores/brokers/#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](/docs/layer-0/durable-subscriptions/) 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](/docs/stores/brokers/#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](/docs/stores/brokers/#consuming-from-a-broker) 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](/docs/layer-0/eventstore-queries/) or a [catch-up subscription](/docs/layer-0/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:

```java
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:

```java
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](https://github.com/johanhaleby/occurrent/blob/main/doc/architecture/decisions/0108-a-live-push-handler-runs-outside-the-handover-lock.md).

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:

```java
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](/docs/layer-1/synchronous-subscriptions/#life-cycle), and it’s what lets a test [stop every subscription model in the application and opt back in per test](/docs/testing/integration-testing/#stopping-every-subscription-then-opting-in), 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:

```java
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

  ```java
  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();
  }
  ```

* Kotlin

  ```kotlin
  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](/docs/stores/brokers/#consuming-from-a-broker) 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](/docs/stores/brokers/#handing-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](/docs/stores/brokers/#handing-a-message-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](/docs/layer-0/push-subscription/#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](/docs/stores/brokers/#handing-a-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](/docs/layer-0/push-subscription/#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](/docs/layer-0/push-subscription/#feeding-events-from-inside-the-projection) shows.

A listener that acknowledges RabbitMQ messages itself looks like this:

* Java

  ```java
  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);
      }
  }
  ```

* Kotlin

  ```kotlin
  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](/docs/stores/brokers/#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

  ```java
  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);
  ```

* Kotlin

  ```kotlin
  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)
  ```

> **Using Spring Boot?**
>
> A `@Projection` with `source = Source.PUSH` binds to a push source, and the starter runs the catch-up and reports its progress on a `PushCatchupStatus` bean. See [Fed by a push source](/docs/spring-boot/projection-annotation/#fed-by-a-push-source).