switch to rxjava2 TestSubscriber in Oplog tests and ensure termination of ChangeStream cursors

Author: jmoghisi

h2-backend/gradle.lockfile

@@ -10,6 +10,7 @@ io.netty:netty-common:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-handler:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-resolver:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-transport:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
+io.reactivex.rxjava2:rxjava:2.2.21=default,testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy-agent:1.10.20=testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy:1.10.20=testCompileClasspath,testRuntimeClasspath
 org.apiguardian:apiguardian-api:1.1.0=testCompileClasspath,testRuntimeClasspath

memory-backend/gradle.lockfile

@@ -9,6 +9,7 @@ io.netty:netty-common:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-handler:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-resolver:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-transport:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
+io.reactivex.rxjava2:rxjava:2.2.21=default,testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy-agent:1.10.20=testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy:1.10.20=testCompileClasspath,testRuntimeClasspath
 org.apiguardian:apiguardian-api:1.1.0=testCompileClasspath,testRuntimeClasspath

postgresql-backend/gradle.lockfile

@@ -18,6 +18,7 @@ io.netty:netty-common:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-handler:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-resolver:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-transport:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
+io.reactivex.rxjava2:rxjava:2.2.21=default,testCompileClasspath,testRuntimeClasspath
 junit:junit:4.13.1=testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy-agent:1.10.20=testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy:1.10.20=testCompileClasspath,testRuntimeClasspath

test-common/build.gradle

@@ -16,6 +16,8 @@ dependencies {
     api "org.mockito:mockito-core:latest.release"
     api "org.mockito:mockito-junit-jupiter:latest.release"
 
+    api group: 'io.reactivex.rxjava2', name: 'rxjava', version: 'latest.release'
+
     runtimeOnly group: 'ch.qos.logback', name: 'logback-classic', version: 'latest.release'
 
     testImplementation "org.testcontainers:testcontainers:latest.release"

test-common/gradle.lockfile

@@ -13,6 +13,7 @@ io.netty:netty-common:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-handler:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-resolver:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
 io.netty:netty-transport:4.1.60.Final=default,runtimeClasspath,testRuntimeClasspath
+io.reactivex.rxjava2:rxjava:2.2.21=default,compileClasspath,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 junit:junit:4.12=testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy-agent:1.10.20=compileClasspath,default,runtimeClasspath,testCompileClasspath,testRuntimeClasspath
 net.bytebuddy:byte-buddy:1.10.20=compileClasspath,default,runtimeClasspath,testCompileClasspath,testRuntimeClasspath

test-common/src/main/java/de/bwaldvogel/mongo/backend/AbstractOplogTest.java

@@ -39,6 +39,7 @@
 import com.mongodb.reactivestreams.client.Success;
 
 import de.bwaldvogel.mongo.oplog.OperationType;
+import io.reactivex.subscribers.TestSubscriber;
 
 public abstract class AbstractOplogTest extends AbstractTest {
 
@@ -239,24 +240,25 @@ public void testChangeStreamInsertAndUpdateFullDocumentLookup() {
             Document.parse("{'fullDocument.b': 1}")))
         );
 
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor =
-            collection.watch(pipeline).fullDocument(FullDocument.UPDATE_LOOKUP).cursor();
+        try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor =
+                 collection.watch(pipeline).fullDocument(FullDocument.UPDATE_LOOKUP).cursor()) {
 
-        for (int i = 1; i < numberOfDocs + 1; i++) {
-            Document doc = json(String.format("a: %d, b: 1", i));
-            collection.insertOne(doc);
-            collection.updateOne(eq("a", i), set("c", i * 10));
+            for (int i = 1; i < numberOfDocs + 1; i++) {
+                Document doc = json(String.format("a: %d, b: 1", i));
+                collection.insertOne(doc);
+                collection.updateOne(eq("a", i), set("c", i * 10));
 
-            ChangeStreamDocument<Document> insertDocument = cursor.next();
-            ChangeStreamDocument<Document> updateDocument = cursor.next();
+                ChangeStreamDocument<Document> insertDocument = cursor.next();
+                ChangeStreamDocument<Document> updateDocument = cursor.next();
 
-            assertThat(insertDocument.getFullDocument().get("a")).isEqualTo(i);
-            insert.add(insertDocument.getFullDocument());
+                assertThat(insertDocument.getFullDocument().get("a")).isEqualTo(i);
+                insert.add(insertDocument.getFullDocument());
 
-            assertThat(updateDocument.getFullDocument().get("a")).isEqualTo(i);
-            update.add(updateDocument.getFullDocument());
+                assertThat(updateDocument.getFullDocument().get("a")).isEqualTo(i);
+                update.add(updateDocument.getFullDocument());
 
-            changeStreamsResult.addAll(Arrays.asList(insertDocument, updateDocument));
+                changeStreamsResult.addAll(Arrays.asList(insertDocument, updateDocument));
+            }
         }
 
         assertThat(insert.size()).isEqualTo(numberOfDocs);
@@ -267,94 +269,107 @@ public void testChangeStreamInsertAndUpdateFullDocumentLookup() {
     @Test
     public void testChangeStreamUpdateDefault() {
         collection.insertOne(json("a: 1, b: 2, c: 3"));
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor();
-        collection.updateOne(eq("a", 1), json("$set: {b: 0, c: 10}"));
-        ChangeStreamDocument<Document> updateDocument = cursor.next();
-        Document fullDoc = updateDocument.getFullDocument();
-        assertThat(fullDoc).isNotNull();
-        assertThat(fullDoc.get("b")).isEqualTo(0);
-        assertThat(fullDoc.get("c")).isEqualTo(10);
-
-        collection.updateOne(eq("a", 1), unset("b"));
-        updateDocument = cursor.next();
-        fullDoc = updateDocument.getFullDocument();
-        assertThat(fullDoc).isNotNull();
-        assertThat(fullDoc.get("b")).isEqualTo("");
+        try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor()) {
+            collection.updateOne(eq("a", 1), json("$set: {b: 0, c: 10}"));
+            ChangeStreamDocument<Document> updateDocument = cursor.next();
+            Document fullDoc = updateDocument.getFullDocument();
+            assertThat(fullDoc).isNotNull();
+            assertThat(fullDoc.get("b")).isEqualTo(0);
+            assertThat(fullDoc.get("c")).isEqualTo(10);
+
+            collection.updateOne(eq("a", 1), unset("b"));
+            updateDocument = cursor.next();
+            fullDoc = updateDocument.getFullDocument();
+            assertThat(fullDoc).isNotNull();
+            assertThat(fullDoc.get("b")).isEqualTo("");
+        }
     }
 
     @Test
     public void testChangeStreamDelete() {
         collection.insertOne(json("_id: 1"));
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor();
-        collection.deleteOne(json("_id: 1"));
-        ChangeStreamDocument<Document> deleteDocument = cursor.next();
-        assertThat(deleteDocument.getDocumentKey().get("_id")).isEqualTo(new BsonInt32(1));
+        try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor()) {
+            collection.deleteOne(json("_id: 1"));
+            ChangeStreamDocument<Document> deleteDocument = cursor.next();
+            assertThat(deleteDocument.getDocumentKey().get("_id")).isEqualTo(new BsonInt32(1));
+        }
     }
 
     @Test
     public void testChangeStreamStartAfter() {
         collection.insertOne(json("a: 1")); // This is needed to initialize the collection in the server.
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor();
-        collection.insertOne(json("a: 2"));
-        collection.insertOne(json("a: 3"));
-        ChangeStreamDocument<Document> document = cursor.next();
-        BsonDocument resumeToken = document.getResumeToken();
-
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor2 = collection.watch().startAfter(resumeToken).cursor();
-        ChangeStreamDocument<Document> document2 = cursor2.next();
-        assertThat(document2.getFullDocument().get("a")).isEqualTo(3);
+        try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor()) {
+            collection.insertOne(json("a: 2"));
+            collection.insertOne(json("a: 3"));
+            ChangeStreamDocument<Document> document = cursor.next();
+            BsonDocument resumeToken = document.getResumeToken();
+
+            try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor2
+                     = collection.watch().startAfter(resumeToken).cursor()) {
+                ChangeStreamDocument<Document> document2 = cursor2.next();
+                assertThat(document2.getFullDocument().get("a")).isEqualTo(3);
+            }
+        }
     }
 
     @Test
     public void testChangeStreamResumeAfter() {
         collection.insertOne(json("a: 1"));
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor();
-        collection.insertOne(json("a: 2"));
-        collection.insertOne(json("a: 3"));
-        ChangeStreamDocument<Document> document = cursor.next();
-        BsonDocument resumeToken = document.getResumeToken();
-
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor2 = collection.watch().resumeAfter(resumeToken).cursor();
-        ChangeStreamDocument<Document> document2 = cursor2.next();
-        assertThat(document2.getFullDocument().get("a")).isEqualTo(3);
+        try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor();) {
+            collection.insertOne(json("a: 2"));
+            collection.insertOne(json("a: 3"));
+            ChangeStreamDocument<Document> document = cursor.next();
+            BsonDocument resumeToken = document.getResumeToken();
+
+            try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor2
+                     = collection.watch().resumeAfter(resumeToken).cursor()) {
+                ChangeStreamDocument<Document> document2 = cursor2.next();
+                assertThat(document2.getFullDocument().get("a")).isEqualTo(3);
+            }
+        }
     }
 
     @Test
     public void testChangeStreamResumeAfterTerminalEvent() {
         assertThrows(NoSuchElementException.class, () -> {
             MongoCollection<Document> col = db.getCollection("test-collection");
             ChangeStreamIterable<Document> watch = col.watch().fullDocument(FullDocument.UPDATE_LOOKUP).batchSize(1);
-            MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = watch.cursor();
-            col.insertOne(json("a: 1"));
-            cursor.next();
-
-            col.drop();
-
-            ChangeStreamDocument<Document> document = cursor.next();
-            BsonDocument resumeToken = document.getResumeToken();
-            cursor = watch.resumeAfter(resumeToken).cursor();
-            document = cursor.next();
-
-            assertThat(document).isNotNull();
-            assertThat(document.getOperationType()).isEqualTo(com.mongodb.client.model.changestream.OperationType.INVALIDATE);
-            cursor.next();
+            try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = watch.cursor();) {
+                col.insertOne(json("a: 1"));
+                cursor.next();
+
+                col.drop();
+
+                ChangeStreamDocument<Document> document = cursor.next();
+                BsonDocument resumeToken = document.getResumeToken();
+                try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> resumeAfterCursor
+                         = watch.resumeAfter(resumeToken).cursor();) {
+                    document = resumeAfterCursor.next();
+
+                    assertThat(document).isNotNull();
+                    assertThat(document.getOperationType())
+                        .isEqualTo(com.mongodb.client.model.changestream.OperationType.INVALIDATE);
+                    resumeAfterCursor.next();
+                }
+            }
         });
     }
 
     @Test
     public void testChangeStreamStartAtOperationTime() {
         collection.insertOne(json("a: 1"));
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor();
-        collection.insertOne(json("a: 2"));
-        collection.insertOne(json("a: 3"));
-        ChangeStreamDocument<Document> document = cursor.next();
-        BsonTimestamp startAtOperationTime = document.getClusterTime();
-
-        MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor2 = collection.watch().startAtOperationTime(startAtOperationTime).cursor();
-        ChangeStreamDocument<Document> document2 = cursor2.next();
-        assertThat(document2.getFullDocument().get("a")).isEqualTo(2);
-        document2 = cursor2.next();
-        assertThat(document2.getFullDocument().get("a")).isEqualTo(3);
+        try (MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor = collection.watch().cursor()) {
+            collection.insertOne(json("a: 2"));
+            collection.insertOne(json("a: 3"));
+            ChangeStreamDocument<Document> document = cursor.next();
+            BsonTimestamp startAtOperationTime = document.getClusterTime();
+
+            MongoChangeStreamCursor<ChangeStreamDocument<Document>> cursor2 = collection.watch().startAtOperationTime(startAtOperationTime).cursor();
+            ChangeStreamDocument<Document> document2 = cursor2.next();
+            assertThat(document2.getFullDocument().get("a")).isEqualTo(2);
+            document2 = cursor2.next();
+            assertThat(document2.getFullDocument().get("a")).isEqualTo(3);
+        }
     }
 
     @Test
@@ -372,9 +387,9 @@ public void testChangeStreamAndReplaceOneWithUpsertTrue() throws Exception {
         TestSubscriber<Document> findSubscriber = new TestSubscriber<>();
         asyncCollection.find(json("a:1")).subscribe(findSubscriber);
         findSubscriber.awaitTerminalEvent();
-        assertThat(findSubscriber.getSingleValue().get("a")).isEqualTo(1);
+        assertThat(findSubscriber.values().get(0).get("a")).isEqualTo(1);
 
-        streamSubscriber.awaitCount(1);
+        streamSubscriber.awaitCount(1).assertValueCount(1).cancel();
         assertThat(streamSubscriber.values().get(0).getOperationType().getValue()).isEqualTo("insert");
         assertThat(streamSubscriber.values().get(0).getFullDocument()).isEqualTo(findSubscriber.values().get(0));
     }
@@ -399,8 +414,8 @@ public void testSimpleChangeStreamWithFilter() throws Exception {
         insertSubscriber1.awaitTerminalEvent();
         insertSubscriber2.awaitTerminalEvent();
 
-        streamSubscriber.awaitCount(1);
-        assertThat(streamSubscriber.getSingleValue().getFullDocument().get("bu")).isEqualTo("abc");
+        streamSubscriber.awaitCount(1).assertValueCount(1).cancel();
+        assertThat(streamSubscriber.values().get(0).getFullDocument().get("bu")).isEqualTo("abc");
     }
 
 }