Remove nested subscription from within Mono.

Author: rozza

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/BatchCursorPublisher.java

@@ -123,21 +123,17 @@ public TimeoutMode getTimeoutMode() {
 
     public Publisher<T> first() {
         return batchCursor(this::asAsyncFirstReadOperation)
-                .flatMap(batchCursor -> Mono.create(sink -> {
+                .flatMap(batchCursor -> {
                     batchCursor.setBatchSize(1);
-                    Mono.from(batchCursor.next())
+                    return Mono.from(batchCursor.next())
                             .doOnTerminate(batchCursor::close)
-                            .doOnError(sink::error)
-                            .doOnSuccess(results -> {
+                            .flatMap(results -> {
                                 if (results == null || results.isEmpty()) {
-                                    sink.success();
-                                } else {
-                                    sink.success(results.get(0));
+                                    return Mono.empty();
                                 }
-                            })
-                            .contextWrite(sink.contextView())
-                            .subscribe();
-                }));
+                                return Mono.fromCallable(() -> results.get(0));
+                            });
+                });
     }
 
     @Override