I would like to leverage from ChangeStream usage to organize versioning/audit in my application. ChangeStreamDocuments are to be intercepted after any of insert*/save*/delete* operations is called using Spring Data/Mongo Repository.

The following code snippet is used to iterate over collection's ChangeStreamDocuments.

try (MongoCursor<Document> cursor = collection.watch().iterator()) {
    while (cursor.hasNext()) {
        store(cursor.next());
    }
}

The issue is that this code sticks after the first operation (new entity insertion) if the DB is empty. An infinite loop is run inside cursor.hasNext().

I could only come up with an asynchronous solution which opens a persistent cursor against the DB and applies a Block function on each new incoming event. The biggest drawback of this solution is that each persistent cursor consumes a connection from the pool.

Block<ChangeStreamDocument<Document>> persistBlock = changeStreamDocument -> store(changeStreamDocument);

collection.watch().forEach(persistBlock);
   

Please, advise what could be done in this case and whether my code should work as expected, but has a bug.