# Blocking subscription models > Subscription models for MongoDB (native driver and Spring), and in-memory. [Layer 0: CloudEvents](/docs/layer-0/) · Blocking · Modules: `occurrent-subscription-mongodb-native-blocking`, `occurrent-subscription-mongodb-spring-blocking`, `occurrent-subscription-inmemory` These are the *non-durable* [blocking subscription implementations](/docs/layer-0/blocking-subscriptions/): **MongoDB** * [Blocking subscription using the “native” Java MongoDB driver](/docs/layer-0/blocking-subscription-models/#blocking-subscription-using-the-native-java-mongodb-driver) * [Blocking subscription using Spring MongoTemplate](/docs/layer-0/blocking-subscription-models/#blocking-subscription-using-spring-mongotemplate) > **Note** > > It’s important to recognize that MongoDB subscriptions are using the [oplog](https://docs.mongodb.com/manual/core/replica-set-oplog/) so you need to make sure that you have enough oplog capacity to support your use case. You can also read more about this [here](https://docs.mongodb.com/manual/changeStreams/#startafter-for-change-streams). Typically this shouldn’t be a problem, but if you have subscribers that risk falling behind, you may consider piping the events to e.g. Kafka and leverage it for these types of subscriptions. Note that you can always use a [catch-up subscription](/docs/layer-0/catch-up-subscription/) to recover if this happens. **In-Memory** * [In-Memory subscription](/docs/layer-0/blocking-subscription-models/#inmemory-subscription) By “non-durable” we mean implementations that doesn’t store the checkpoint in a durable storage automatically.\ It might be that the datastore does this automatically *or* that [checkpoint storage](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage) is not required for your use case. If the datastore *doesn’t* support storing the checkpoint automatically, a subscription will typically implement the `org.occurrent.subscription.api.blocking.CheckpointAwareSubscriptionModel` interface (since these types of subscriptions needs to be aware of the checkpoint). Typically, if you want the stream to continue where it left off on application restart you want to store away the checkpoint. You can do this anyway you like, but for most cases you probably want to look into implementations of `org.occurrent.subscription.api.blocking.CheckpointAwareSubscriptionModel`. These subscriptions can be combined with a [checkpoint storage](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage) implementation to store the checkpoint in a durable datastore. Occurrent provides a [utility](/docs/layer-0/durable-subscriptions/) that combines a `CheckpointAwareSubscriptionModel` and a `CheckpointStorage` (see [here](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage)) to automatically store the checkpoint\ *after each processed event*. You can tweak how often the checkpoint should be persisted in the configuration. ## Blocking Subscription using the “Native” Java MongoDB Driver Uses the vanilla Java MongoDB synchronous driver (no Spring dependency is required). To get started first include the following dependency: * Maven ```xml org.occurrent occurrent-subscription-mongodb-native-blocking 0.34.0 ``` * Gradle ```groovy compile 'org.occurrent:occurrent-subscription-mongodb-native-blocking:0.34.0' ``` * SBT ```scala libraryDependencies += "org.occurrent" % "occurrent-subscription-mongodb-native-blocking" % "0.34.0" ``` * Grape ```groovy @Grab(group='org.occurrent', module='occurrent-subscription-mongodb-native-blocking', version='0.34.0') ``` * Leiningen ```clojure [org.occurrent/occurrent-subscription-mongodb-native-blocking "0.34.0"] ``` * Buildr ```ruby 'org.occurrent:occurrent-subscription-mongodb-native-blocking:jar:0.34.0' ``` * Ivy ```xml ``` Then create a new instance of `NativeMongoSubscriptionModel` and start subscribing: * Java ```java MongoDatabase database = mongoClient.getDatabase("some-database"); // Create the blocking subscription SubscriptionModel subscriptionModel = new NativeMongoSubscriptionModel(database, "eventCollection", TimeRepresentation.DATE, Executors.newCachedThreadPool(), RetryStrategy.retry().fixed(200)); // Now you can create subscription instances that subscribes to new events as they're written to an EventStore subscriptionModel.subscribe("mySubscriptionId", cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent)); // You can later cancel the subscription by calling: subscriptionModel.cancelSubscription("mySubscriptionId"); ``` * Kotlin ```kotlin val database = mongoClient.getDatabase("some-database") // Create the blocking subscription val subscriptionModel = NativeMongoSubscriptionModel(database, "eventCollection", TimeRepresentation.DATE, Executors.newCachedThreadPool(), RetryStrategy.retry().fixed(200)) // Now you can create subscriptions instances that subscribes to new events as they're written to an EventStore subscriptionModel.subscribe("mySubscriptionId") { cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent) } // You can later cancel the subscription by calling: subscriptionModel.cancelSubscription("mySubscriptionId") ``` > **Note** > > NativeMongoSubscriptionModel can be imported from the “org.occurrent.subscription.mongodb.nativedriver.blocking” package. There are a few things to note here that needs explaining. First we have the `TimeRepresentation.DATE` that is passed as the third constructor argument which you can read more about [here](/docs/stores/choosing-an-event-store/#mongodb-time-representation). Secondly we have the `Executors.newCachedThreadPool()`. A thread will be created from this executor for each call to “subscribe” (i.e. for each subscription). Make sure that you have enough threads to cover all your subscriptions or the “SubscriptionModel” may hang. Last we have the [RetryStrategy](/docs/stores/retry/#retry-configuration-blocking) which defines what should happen if there’s e.g. a connection issue during the life-time of a subscription or if subscription fails to process a cloud event (i.e. the `action` throws an exception). Note that you can provide a [filter](/docs/layer-0/blocking-subscriptions/#blocking-subscription-filters), [start position](/docs/layer-0/blocking-subscriptions/#blocking-subscription-start-position) and [checkpoint persistence](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage) for this subscription implementation. ## Blocking Subscription using Spring MongoTemplate An implementation that uses Spring’s [MongoTemplate](https://docs.spring.io/spring-data/mongodb/docs/current/api/org/springframework/data/mongodb/core/MongoTemplate.html) for event subscriptions. First include the following dependency: * Maven ```xml org.occurrent occurrent-subscription-mongodb-spring-blocking 0.34.0 ``` * Gradle ```groovy compile 'org.occurrent:occurrent-subscription-mongodb-spring-blocking:0.34.0' ``` * SBT ```scala libraryDependencies += "org.occurrent" % "occurrent-subscription-mongodb-spring-blocking" % "0.34.0" ``` * Grape ```groovy @Grab(group='org.occurrent', module='occurrent-subscription-mongodb-spring-blocking', version='0.34.0') ``` * Leiningen ```clojure [org.occurrent/occurrent-subscription-mongodb-spring-blocking "0.34.0"] ``` * Buildr ```ruby 'org.occurrent:occurrent-subscription-mongodb-spring-blocking:jar:0.34.0' ``` * Ivy ```xml ``` Then create a new instance of `SpringMongoSubscriptionModel` and start subscribing: * Java ```java MongoTemplate mongoTemplate = ... // Create the blocking subscription SubscriptionModel subscriptionModel = new SpringMongoSubscriptionModel(mongoTemplate, "eventCollectionName", TimeRepresentation.RFC_3339_STRING); // Now you can create subscriptions instances that subscribes to new events as they're written to an EventStore // Typically you do this after the Spring application context has finished loading. For example by subscribing to // the (org.springframework.boot.context.event.ApplicationStartedEvent) or in a method annotated with (@PostConstruct) subscriptionModel.subscribe("mySubscriptionId", cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent)); // You can later cancel the subscription by calling: subscriptionModel.cancelSubscription("mySubscriptionId"); ``` * Kotlin ```kotlin val mongoTemplate : MongoTemplate = ... // Create the blocking subscription val subscriptionModel = SpringMongoSubscriptionModel(mongoTemplate, "eventCollectionName", TimeRepresentation.RFC_3339_STRING) // Now you can create subscriptions instances that subscribes to new events as they're written to an EventStore subscriptionModel.subscribe("mySubscriptionId") { cloudEvent -> doSomethingWithTheCloudEvent(cloudEvent) } // You can later cancel the subscription by calling: subscriptionModel.cancelSubscription("mySubscriptionId") ``` > **Note** > > SpringMongoSubscriptionModel can be imported from the “org.occurrent.subscription.mongodb.spring.blocking” package. The “eventCollectionName” specifies the event collection in MongoDB where events are stored. It’s important that this collection is the same as the collection used by the `EventStore` implementation. Secondly, we have the `TimeRepresentation.RFC_3339_STRING` that is passed as the third constructor argument, which you can read more about [here](/docs/stores/choosing-an-event-store/#mongodb-time-representation). It’s also very important that this is configured the same way as the `EventStore`. Each subscription reads its change stream with a cursor of its own, on a thread from the model’s executor. When the change stream fails, for example because the connection to MongoDB is lost, the model opens it again from the position the subscription had reached, and retries that with its `RetryStrategy`. When it comes to retries, if the “action” fails (i.e. if the higher-order function you provide when calling `subscribe` throws an exception), either using something like [Spring Retry](https://github.com/spring-projects/spring-retry) or the [Occurrent Retry Module](/docs/stores/retry/#retry-configuration-blocking). By default, all subscription models will use the Occurrent retry module with exponential backoff starting with 100 ms and progressively go up to max 2 seconds wait time between each retry when reading/saving/deleting the checkpoint. You can customize this by passing an instance of `RetryStrategy` to the `SpringMongoSubscriptionModel` constructor. When the action still throws after the `RetryStrategy` gives up, the model opens the change stream again at the position before that event and delivers the event again, so the event isn’t skipped. The same `RetryStrategy` decides how many times the model opens the change stream again. It counts every reopen since the subscription was started or resumed, including those after a lost connection. When it gives up, the model logs an error, and the subscription receives nothing more until you pause and resume it. With `RetryStrategy.none()` that happens the first time the action throws. If you want to disable the Occurrent retry module, pass `RetryStrategy.none()` to the `SpringMongoSubscriptionModel` constructor and then handle retries anyway you find fit. For example, let’s say you want to use `spring-retry`, and you have a simple Spring bean that writes each cloud event to a repository: ```java @Component public class WriteToRepository { private final SomeRepository someRepository; public WriteToRepository(SomeRepository someRepository) { this.someRepository = someRepository; } public void write(CloudEvent cloudEvent) { someRepository.persist(cloudEvent); } } ``` And you want to subscribe to all “GameStarted” events and write them to the repository: ```java WriteToRepository writeToRepository = ... subscriptionModel.subscribe("gameStartedLog", writeToRepository::write); ``` But if the connection to `someRepository` is flaky you can add [Spring Retry](https://github.com/spring-projects/spring-retry) so allow for retry with exponential backoff: ```java @Component public class WriteToRepository { private final SomeRepository someRepository; public WriteToRepository(SomeRepository someRepository) { this.someRepository = someRepository; } @Retryable(backoff = @Backoff(delay = 200, multiplier = 2, maxDelay = 30000)) public void write(CloudEvent cloudEvent) { someRepository.persist(cloudEvent); } } ``` Don’t forget to add `@EnableRetry` in to your Spring Boot application as well. Note that you can provide a [filter](/docs/layer-0/blocking-subscriptions/#blocking-subscription-filters), [start position](/docs/layer-0/blocking-subscriptions/#blocking-subscription-start-position) and [checkpoint persistence](/docs/layer-0/blocking-subscriptions/#blocking-subscription-checkpoint-storage) for this subscription implementation. ### Restart Subscription when Oplog Lost From 0.34.0, a subscription whose filter matches no events for a long time no longer needs automatic restart. When a read of the change stream returns no event for it, `SpringMongoSubscriptionModel` moves its position forward anyway, so a resume after a pause opens at a position the oplog still has. A [`DurableSubscriptionModel`](/docs/layer-0/durable-subscriptions/) wrapping the model also saves that position as the checkpoint once a minute, so a restart or a lease moving to another node resumes from it too. You still need automatic restart when: * the subscription was paused, the application was down, or no node held the subscription’s lease, for longer than the oplog keeps history * no `DurableSubscriptionModel` wraps the model, and the subscription restarts from a position you stored yourself * the save is off (`neverSaveQuietPosition()`), or its interval (`saveQuietPositionEvery(Duration)`) is longer than the oplog keeps history * the persist predicate declined the last event before the subscription went quiet, as `EveryN` with `n` above 1 can, or the subscription started from a `StartAt` of your own and the predicate hasn’t stored a position yet When the oplog no longer has the history a subscription needs to resume, the subscription model can restart the subscription from the current time instead. This section uses `SpringMongoSubscriptionModel` as the example, but `NativeMongoSubscriptionModel` and the reactive `ReactorMongoSubscriptionModel` work the same way, with the same `restartSubscriptionsOnChangeStreamHistoryLost(true)` setting on their config. A subscription runs into this whenever it opens its change stream at a position the oplog no longer has. That happens when an application restarts with a subscription configured to start from such a position. It also happens when a subscription is resumed, or its change stream restarts, after being paused or disconnected for longer than the oplog keeps history. A subscription that hasn’t handled an event yet gets the same handling. Automatic restart is off when you create the model yourself, because a subscription that restarts from the current time can miss events. That happens when the subscriptions run in a different process than the writes, and the event store gets many writes. It’s safe when the subscription runs in the same process as the writes, as long as the subscription starts before the application accepts writes. To turn it on when you create the model yourself: ```java var subscriptionModel = new SpringMongoSubscriptionModel(mongoTemplate, SpringMongoSubscriptionModelConfig.withConfig("events", TimeRepresentation.RFC_3339_STRING).restartSubscriptionsOnChangeStreamHistoryLost(true)); ``` With automatic restart turned off, the model logs an error and removes the subscription, so its id can be subscribed again. With automatic restart turned on and a [`DurableSubscriptionModel`](/docs/layer-0/durable-subscriptions/) wrapping the model, the position the subscription restarts from is stored as its checkpoint before the restart, unless another node has written the checkpoint with a newer lease. So a process restart before the next event doesn’t open the change stream at the lost position again. When the lost position is the live start a [catch-up subscription](/docs/layer-0/catch-up-subscription/#resuming-a-replay-from-a-stored-checkpoint) handed over, the model asks the catch-up first, whichever way automatic restart is set. The catch-up takes the subscription over, and the model pauses it instead of restarting or removing it. An alternative approach to restarting automatically is to use a [catch-up subscription](/docs/layer-0/catch-up-subscription/) and restart the subscription from an earlier date. ### Deferring Startup (Added in 0.32.0) By default a `SpringMongoSubscriptionModel` starts itself as soon as it’s constructed, and if you register it as a Spring bean, Spring starts it again on refresh (it implements `SmartLifecycle`). Set `autoStartup(false)` on the config to skip both: ```java var config = SpringMongoSubscriptionModelConfig .withConfig("events", TimeRepresentation.RFC_3339_STRING) .autoStartup(false); var subscriptionModel = new SpringMongoSubscriptionModel(mongoTemplate, config); ``` A subscription registered on a model created this way comes back paused, the same as if you’d called `stop()` right after construction. No change stream opens until you call `start()` or `resumeSubscription(id)` yourself. `subscribe(..)` asks MongoDB for its current operation time on the model’s executor, without waiting for the answer. A subscription from `StartAt.now()`, or with no `StartAt`, starts at that time. So it receives the events written from shortly after `subscribe(..)` returns, including those written before `start()`. Use `autoStartup(false)` to bring subscriptions up under your own control, behind a leader election or a health check for example, or in a test that wants to choose which subscriptions actually run before the context finishes starting. ## Tuning the MongoDB change stream (Added in 0.31.0) Both blocking MongoDB subscription models let you tune the change stream they read from. Both options are opt-in. Leave them unset and you get the driver and server defaults, which is what every subscription did before these existed, so upgrading changes nothing on its own. `batchSize` sets how many change-stream documents the server returns per batch. A larger batch means fewer round trips to the server, which helps throughput on a high-volume subscription such as an outbox, and costs more memory per batch. Values in the low hundreds work well for high throughput, but the right number depends on your event size and load. `maxAwaitTime` bounds how long the server holds an idle cursor before returning a batch, possibly an empty one. A shorter wait delivers events sooner and costs more `getMore` round trips while the stream is quiet. A longer wait keeps an idle cursor waiting and reduces that chatter. Somewhere between 200 ms and 1000 ms suits most workloads. ```java var config = NativeMongoSubscriptionModelConfig.withConfig() .batchSize(500) .maxAwaitTime(Duration.ofMillis(500)); var subscriptionModel = new NativeMongoSubscriptionModel(database, "events", TimeRepresentation.DATE, executor, config); ``` The two models don’t offer the same options. `NativeMongoSubscriptionModel` has both. `SpringMongoSubscriptionModelConfig` has `maxAwaitTime` only: ```java var config = SpringMongoSubscriptionModelConfig .withConfig("events", TimeRepresentation.RFC_3339_STRING) .maxAwaitTime(Duration.ofMillis(500)); ``` Use `NativeMongoSubscriptionModel` when you need `batchSize`. `ReactorMongoSubscriptionModel` offers neither. ## InMemory Subscription If you’re using the [InMemory EventStore](/docs/stores/in-memory-event-store/) you can use the “InMemorySubscriptionModel” to subscribe to new events. For add the dependency: * Maven ```xml org.occurrent occurrent-subscription-inmemory 0.34.0 ``` * Gradle ```groovy compile 'org.occurrent:occurrent-subscription-inmemory:0.34.0' ``` * SBT ```scala libraryDependencies += "org.occurrent" % "occurrent-subscription-inmemory" % "0.34.0" ``` * Grape ```groovy @Grab(group='org.occurrent', module='occurrent-subscription-inmemory', version='0.34.0') ``` * Leiningen ```clojure [org.occurrent/occurrent-subscription-inmemory "0.34.0"] ``` * Buildr ```ruby 'org.occurrent:occurrent-subscription-inmemory:jar:0.34.0' ``` * Ivy ```xml ``` Then you can use it like this: * Java ```java InMemorySubscriptionModel subscriptionModel = new InMemorySubscriptionModel(); InMemoryEventStore inMemoryEventStore = new InMemoryEventStore(inMemorySubscriptionModel); subscriptionModel.subscribe("subscription1", System.out::println); // You can later cancel the subscription by calling: subscriptionModel.cancelSubscription("mySubscriptionId"); ``` * Kotlin ```kotlin val subscriptionModel = new InMemorySubscriptionModel() val inMemoryEventStore = new InMemoryEventStore(inMemorySubscriptionModel) subscriptionModel.subscribe("subscription1", ::println) // You can later cancel the subscription by calling: subscriptionModel.cancelSubscription("mySubscriptionId") ``` By default this model 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). > **Using Spring Boot?** > > The starter turns automatic restart on. See [What the starter configures](/docs/spring-boot/what-the-starter-configures/#restarting-when-the-oplog-is-lost).