Avoid calling batch cursor when the cursor is closed

JAVA-3710

Author: jstewart148

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

@@ -79,6 +79,9 @@ public void request(final long n) {
                     + "argument is <= 0."));
             return;
         }
+        if (isTerminated()) {
+            return;
+        }
 
         boolean requestData = false;
         synchronized (this) {
@@ -142,7 +145,7 @@ void onError(final Throwable t) {
                 throw MongoException.fromThrowableNonNull(t1);
             }
         } else {
-            throw MongoException.fromThrowableNonNull(t);
+            throw new MongoException("Subscription has already been terminated", t);
         }
     }
 

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

@@ -74,7 +74,7 @@ void postTerminate() {
     void requestMoreData() {
         boolean mustRead = false;
         synchronized (this) {
-            if (!isReading && !isTerminated() && batchCursor != null) {
+            if (!isReading && !isTerminated() && batchCursor != null && !batchCursor.isClosed()) {
                 isReading = true;
                 mustRead = true;
             }

driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/FlatteningSingleResultCallbackSubscriptionSpecification.groovy

@@ -265,7 +265,8 @@ class FlatteningSingleResultCallbackSubscriptionSpecification extends Specificat
 
         then:
         def ex = thrown(MongoException)
-        ex.message == 'exception calling onComplete'
+        ex.message == 'Subscription has already been terminated'
+        ex.cause.cause.message == 'exception calling onComplete'
         subscriber.assertTerminalEvent()
         subscriber.assertNoErrors()
     }
@@ -298,7 +299,8 @@ class FlatteningSingleResultCallbackSubscriptionSpecification extends Specificat
 
         then:
         def ex = thrown(MongoException)
-        ex.message == 'exception calling onError'
+        ex.message == 'Subscription has already been terminated'
+        ex.cause.cause.message == 'exception calling onError'
         subscriber.assertTerminalEvent()
         subscriber.assertErrored()
     }

driver-reactive-streams/src/test/unit/com/mongodb/reactivestreams/client/internal/SingleResultCallbackSubscriptionSpecification.groovy

@@ -302,7 +302,8 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
         def ex = thrown(MongoException)
         subscriber.assertNoErrors()
         subscriber.assertTerminalEvent()
-        ex.message == 'exception calling onComplete'
+        ex.message == 'Subscription has already been terminated'
+        ex.cause.cause.message == 'exception calling onComplete'
     }
 
     def 'should throw the exception if calling onError raises one'() {