tailable-cursors.adoc

Infinite Streams with Tailable Cursors

By default, MongoDB automatically closes a cursor when the client exhausts all results supplied by the cursor. Closing a cursor on exhaustion turns a stream into a finite stream. For capped collections, you can use a Tailable Cursor that remains open after the client consumed all initially returned data.

Tip	Capped collections can be created with MongoOperations.createCollection. To do so, provide the required CollectionOptions.empty().capped()…​.

Tailable cursors can be consumed with both, the imperative and the reactive MongoDB API. It is highly recommended to use the reactive variant, as it is less resource-intensive. However, if you cannot use the reactive API, you can still use a messaging concept that is already prevalent in the Spring ecosystem.

Tailable Cursors with MessageListener

Listening to a capped collection using a Sync Driver creates a long running, blocking task that needs to be delegated to a separate component. In this case, we need to first create a MessageListenerContainer, which will be the main entry point for running the specific SubscriptionRequest. Spring Data MongoDB already ships with a default implementation that operates on MongoTemplate and is capable of creating and running Task instances for a TailableCursorRequest.

The following example shows how to use tailable cursors with MessageListener instances:

Example 1. Tailable Cursors with MessageListener instances

MessageListenerContainer container = new DefaultMessageListenerContainer(template);
container.start();                                                                  (1)

MessageListener<Document, User> listener = System.out::println;                     (2)

TailableCursorRequest request = TailableCursorRequest.builder()
  .collection("orders")                                                             (3)
  .filter(query(where("value").lt(100)))                                            (4)
  .publishTo(listener)                                                              (5)
  .build();

container.register(request, User.class);                                            (6)

// ...

container.stop();                                                                   (7)

1. Starting the container intializes the resources and starts Task instances for already registered SubscriptionRequest instances. Requests added after startup are ran immediately.

2. Define the listener called when a Message is received. The Message#getBody() is converted to the requested domain type. Use Document to receive raw results without conversion.

3. Set the collection to listen to.

4. Provide an optional filter for documents to receive.

5. Set the message listener to publish incoming Messages to.

6. Register the request. The returned Subscription can be used to check the current Task state and cancel it to free resources.

7. Do not forget to stop the container once you are sure you no longer need it. Doing so stops all running Task instances within the container.

Reactive Tailable Cursors

Using tailable cursors with a reactive data types allows construction of infinite streams. A tailable cursor remains open until it is closed externally. It emits data as new documents arrive in a capped collection.

Tailable cursors may become dead, or invalid, if either the query returns no match or the cursor returns the document at the “end” of the collection and the application then deletes that document. The following example shows how to create and use an infinite stream query:

Example 2. Infinite Stream queries with ReactiveMongoOperations

Flux<Person> stream = template.tail(query(where("name").is("Joe")), Person.class);

Disposable subscription = stream.doOnNext(person -> System.out.println(person)).subscribe();

// …

// Later: Dispose the subscription to close the stream
subscription.dispose();

Spring Data MongoDB Reactive repositories support infinite streams by annotating a query method with @Tailable. This works for methods that return Flux and other reactive types capable of emitting multiple elements, as the following example shows:

Example 3. Infinite Stream queries with ReactiveMongoRepository

public interface PersonRepository extends ReactiveMongoRepository<Person, String> {

  @Tailable
  Flux<Person> findByFirstname(String firstname);

}

Flux<Person> stream = repository.findByFirstname("Joe");

Disposable subscription = stream.doOnNext(System.out::println).subscribe();

// …

// Later: Dispose the subscription to close the stream
subscription.dispose();