Fix possible onErrorDropped logged message

Caused by using `doOnError` rather than using `subscribe` and
passing the handler in there.

JAVA-5284 JAVA-5266

Author: rozza

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

@@ -87,29 +87,28 @@ private void recurseCursor(){
                 batchCursor.setBatchSize(calculateBatchSize(sink.requestedFromDownstream()));
                 Mono.from(batchCursor.next(() -> sink.isCancelled()))
                         .doOnCancel(this::closeCursor)
-                        .doOnError((e) -> {
-                            try {
-                                closeCursor();
-                            } finally {
-                                sink.error(e);
-                            }
-                        })
-                        .doOnSuccess(results -> {
-                            if (!results.isEmpty()) {
-                                results
-                                        .stream()
-                                        .filter(Objects::nonNull)
-                                        .forEach(sink::next);
-                                calculateDemand(-results.size());
-                            }
-                            if (batchCursor.isClosed()) {
-                                sink.complete();
-                            } else {
-                                inProgress.set(false);
-                                recurseCursor();
-                            }
-                        })
-                        .subscribe();
+                        .subscribe(results -> {
+                                    if (!results.isEmpty()) {
+                                        results
+                                                .stream()
+                                                .filter(Objects::nonNull)
+                                                .forEach(sink::next);
+                                        calculateDemand(-results.size());
+                                    }
+                                    if (batchCursor.isClosed()) {
+                                        sink.complete();
+                                    } else {
+                                        inProgress.set(false);
+                                        recurseCursor();
+                                    }
+                                },
+                                e -> {
+                                    try {
+                                        closeCursor();
+                                    } finally {
+                                        sink.error(e);
+                                    }
+                                });
                 }
         }
     }