Avoid calling batch cursor when the cursor is closed

JAVA-3710

Author: jstewart148

driver-async/src/main/com/mongodb/async/client/AbstractSubscription.java

@@ -69,6 +69,9 @@ public void request(final long n) {
         if (n < 1) {
             throw new IllegalArgumentException("Number requested must be > 0: " + n);
         }
+        if (isTerminated()) {
+            return;
+        }
 
         boolean requestData = false;
         synchronized (this) {
@@ -132,7 +135,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-async/src/main/com/mongodb/async/client/MongoIterableSubscription.java

@@ -78,7 +78,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-async/src/test/unit/com/mongodb/async/client/FlatteningSingleResultCallbackSubscriptionSpecification.groovy

@@ -279,7 +279,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'
         observer.assertTerminalEvent()
         observer.assertNoErrors()
     }
@@ -312,7 +313,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'
         observer.assertTerminalEvent()
         observer.assertErrored()
     }

driver-async/src/test/unit/com/mongodb/async/client/MongoIterableSubscriptionSpecification.groovy

@@ -464,7 +464,7 @@ class MongoIterableSubscriptionSpecification extends Specification {
 
         then:
         def ex = thrown(MongoException)
-        ex.message == 'exception calling onComplete'
+        ex.message == 'Subscription has already been terminated'
         observer.assertTerminalEvent()
         observer.assertNoErrors()
     }
@@ -497,7 +497,8 @@ class MongoIterableSubscriptionSpecification extends Specification {
 
         then:
         def ex = thrown(MongoException)
-        ex.message == 'exception calling onError'
+        ex.message == 'Subscription has already been terminated'
+        ex.cause.cause.message == 'exception calling onError'
         observer.assertTerminalEvent()
         observer.assertErrored()
     }

driver-async/src/test/unit/com/mongodb/async/client/SingleResultCallbackSubscriptionSpecification.groovy

@@ -318,7 +318,8 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
         def ex = thrown(MongoException)
         observer.assertNoErrors()
         observer.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'() {