Simpler fix at the flux level only

Author: rozza

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

@@ -21,29 +21,21 @@
 
 import java.util.List;
 
+import static com.mongodb.reactivestreams.client.internal.MongoOperationPublisher.sinkToCallback;
+
 /**
  * <p>This class is not part of the public API and may be removed or changed at any time</p>
  */
 public class BatchCursor<T> implements AutoCloseable {
 
     private final AsyncBatchCursor<T> wrapped;
-    private volatile boolean cursorClosed = false;
 
     public BatchCursor(final AsyncBatchCursor<T> wrapped) {
         this.wrapped = wrapped;
     }
 
     public Publisher<List<T>> next() {
-        return Mono.create(sink -> wrapped.next(
-                (result, t) -> {
-                    if (t != null && !cursorClosed) {
-                        sink.error(t);
-                    } else if (result == null) {
-                        sink.success();
-                    } else {
-                        sink.success(result);
-                    }
-                }));
+        return Mono.create(sink -> wrapped.next(sinkToCallback(sink)));
     }
 
     public void setBatchSize(final int batchSize) {
@@ -59,7 +51,6 @@ public boolean isClosed() {
     }
 
     public void close() {
-        cursorClosed = true;
         wrapped.close();
     }
 

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

@@ -32,6 +32,7 @@ class BatchCursorFlux<T> implements Publisher<T> {
     private final BatchCursorPublisher<T> batchCursorPublisher;
     private final AtomicBoolean inProgress = new AtomicBoolean(false);
     private final AtomicLong demandDelta = new AtomicLong(0);
+    private final AtomicBoolean fluxCancelled = new AtomicBoolean(false);
     private volatile BatchCursor<T> batchCursor;
     private FluxSink<T> sink;
 
@@ -72,6 +73,11 @@ public void subscribe(final Subscriber<? super T> subscriber) {
         .subscribe(subscriber);
     }
 
+    private void cancelled() {
+        fluxCancelled.set(true);
+        closeCursor();
+    }
+
     private void closeCursor() {
         if (batchCursor != null) {
             batchCursor.close();
@@ -85,12 +91,13 @@ private void recurseCursor(){
             } else {
                 batchCursor.setBatchSize(calculateBatchSize(sink.requestedFromDownstream()));
                 Mono.from(batchCursor.next())
-                        .doOnCancel(this::closeCursor)
                         .doOnError((e) -> {
                             try {
                                 closeCursor();
                             } finally {
-                                sink.error(e);
+                                if (!fluxCancelled.get()) {
+                                    sink.error(e);
+                                }
                             }
                         })
                         .doOnSuccess(results -> {