Test: updated reactive syncadapter to use reactor

Author: rozza

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SingleResultSubscriber.java

@@ -20,9 +20,12 @@
 import com.mongodb.lang.Nullable;
 import com.mongodb.reactivestreams.client.AggregatePublisher;
 import org.bson.conversions.Bson;
+import reactor.core.publisher.Mono;
 
 import java.util.concurrent.TimeUnit;
 
+import static com.mongodb.ClusterFixture.TIMEOUT_DURATION;
+
 class SyncAggregateIterable<T> extends SyncMongoIterable<T> implements AggregateIterable<T> {
     private final AggregatePublisher<T> wrapped;
 
@@ -33,9 +36,7 @@ class SyncAggregateIterable<T> extends SyncMongoIterable<T> implements Aggregate
 
     @Override
     public void toCollection() {
-        SingleResultSubscriber<Void> subscriber = new SingleResultSubscriber<>();
-        wrapped.toCollection().subscribe(subscriber);
-        subscriber.get();
+        Mono.from(wrapped.toCollection()).block(TIMEOUT_DURATION);
     }
 
     @Override

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SyncAggregateIterable.java

@@ -25,6 +25,9 @@
 import com.mongodb.session.ServerSession;
 import org.bson.BsonDocument;
 import org.bson.BsonTimestamp;
+import reactor.core.publisher.Mono;
+
+import static com.mongodb.ClusterFixture.TIMEOUT_DURATION;
 
 class SyncClientSession implements ClientSession {
     private final com.mongodb.reactivestreams.client.ClientSession wrapped;
@@ -131,16 +134,12 @@ public void startTransaction(final TransactionOptions transactionOptions) {
 
     @Override
     public void commitTransaction() {
-        SingleResultSubscriber<Void> subscriber = new SingleResultSubscriber<>();
-        wrapped.commitTransaction().subscribe(subscriber);
-        subscriber.get();
+        Mono.from(wrapped.commitTransaction()).block(TIMEOUT_DURATION);
     }
 
     @Override
     public void abortTransaction() {
-        SingleResultSubscriber<Void> subscriber = new SingleResultSubscriber<>();
-        wrapped.abortTransaction().subscribe(subscriber);
-        subscriber.get();
+        Mono.from(wrapped.abortTransaction()).block(TIMEOUT_DURATION);
     }
 
     @Override

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SyncClientSession.java

@@ -22,9 +22,12 @@
 import com.mongodb.lang.Nullable;
 import com.mongodb.reactivestreams.client.MapReducePublisher;
 import org.bson.conversions.Bson;
+import reactor.core.publisher.Mono;
 
 import java.util.concurrent.TimeUnit;
 
+import static com.mongodb.ClusterFixture.TIMEOUT_DURATION;
+
 @SuppressWarnings("deprecation")
 class SyncMapReduceIterable<T> extends SyncMongoIterable<T> implements MapReduceIterable<T> {
     private final MapReducePublisher<T> wrapped;
@@ -36,9 +39,7 @@ class SyncMapReduceIterable<T> extends SyncMongoIterable<T> implements MapReduce
 
     @Override
     public void toCollection() {
-        SingleResultSubscriber<Void> subscriber = new SingleResultSubscriber<>();
-        wrapped.toCollection().subscribe(subscriber);
-        subscriber.get();
+        Mono.from(wrapped.toCollection()).block(TIMEOUT_DURATION);
     }
 
     @Override

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SyncMapReduceIterable.java

@@ -26,9 +26,11 @@
 import com.mongodb.connection.ClusterDescription;
 import org.bson.Document;
 import org.bson.conversions.Bson;
+import reactor.core.publisher.Mono;
 
 import java.util.List;
 
+import static com.mongodb.ClusterFixture.TIMEOUT_DURATION;
 import static java.util.Objects.requireNonNull;
 
 public class SyncMongoClient implements MongoClient {
@@ -45,20 +47,12 @@ public MongoDatabase getDatabase(final String databaseName) {
 
     @Override
     public ClientSession startSession() {
-        SingleResultSubscriber<com.mongodb.reactivestreams.client.ClientSession> subscriber = createSubscriber();
-        wrapped.startSession().subscribe(subscriber);
-        return new SyncClientSession(requireNonNull(subscriber.get()), this);
-    }
-
-    private <TResult> SingleResultSubscriber<TResult> createSubscriber() {
-        return new SingleResultSubscriber<>();
+        return new SyncClientSession(requireNonNull(Mono.from(wrapped.startSession()).block(TIMEOUT_DURATION)), this);
     }
 
     @Override
     public ClientSession startSession(final ClientSessionOptions options) {
-        SingleResultSubscriber<com.mongodb.reactivestreams.client.ClientSession> subscriber = createSubscriber();
-        wrapped.startSession(options).subscribe(subscriber);
-        return new SyncClientSession(requireNonNull(subscriber.get()), this);
+        return new SyncClientSession(requireNonNull(Mono.from(wrapped.startSession(options)).block(TIMEOUT_DURATION)), this);
     }
 
     @Override