Ensure Observables catch errors when requesting data

JAVA-2048

Author: rozza

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

@@ -79,9 +79,9 @@ public void request(final long n) {
         }
 
         if (requestData) {
-            requestInitialData();
+            tryRequestInitialData();
         } else {
-            processResultsQueue();
+            tryProcessResultsQueue();
         }
     }
 
@@ -107,14 +107,14 @@ void addToQueue(final TResult result) {
         if (result != null) {
             resultsQueue.add(result);
         }
-        processResultsQueue();
+        tryProcessResultsQueue();
     }
 
     void addToQueue(final List<TResult> results) {
         if (results != null) {
             resultsQueue.addAll(results);
         }
-        processResultsQueue();
+        tryProcessResultsQueue();
     }
 
     void onError(final Throwable t) {
@@ -146,6 +146,22 @@ void onComplete() {
         }
     }
 
+    private void tryRequestInitialData() {
+        try {
+            requestInitialData();
+        } catch (Throwable t) {
+            onError(t);
+        }
+    }
+
+    private void tryProcessResultsQueue() {
+        try {
+            processResultsQueue();
+        } catch (Throwable t) {
+            onError(t);
+        }
+    }
+
     private void processResultsQueue() {
         boolean mustProcess = false;
 

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

@@ -242,6 +242,24 @@ class FlatteningSingleResultCallbackSubscriptionSpecification extends Specificat
         observer.assertTerminalEvent()
     }
 
+    def 'should call onError if the passed block errors'() {
+        given:
+        def observer = new TestObserver()
+        observeAndFlatten(new Block<SingleResultCallback<List<Integer>>>() {
+            @Override
+            void apply(final SingleResultCallback<List<Integer>> callback) {
+                throw new MongoException('failed');
+            }
+        }).subscribe(observer)
+
+        when:
+        observer.requestMore(1)
+
+        then:
+        observer.assertErrored()
+        observer.assertTerminalEvent()
+    }
+
     def getBlock() {
         new Block<SingleResultCallback<List<Integer>>>() {
 

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

@@ -372,6 +372,23 @@ class MongoIterableSubscriptionSpecification extends Specification {
 
         then:
         observer.assertTerminalEvent()
+        observer.assertErrored()
+    }
+
+    def 'should call onError if MongoIterable errors'() {
+        given:
+        def observer = new TestObserver()
+        observe(getMongoIterable(getFailingCursor(failImmediately))).subscribe(observer)
+
+        when:
+        observer.requestMore(3)
+
+        then:
+        observer.assertTerminalEvent()
+        observer.assertErrored()
+
+        where:
+        failImmediately << [true, false]
     }
 
     def getMongoIterable() {
@@ -394,4 +411,21 @@ class MongoIterableSubscriptionSpecification extends Specification {
             }
         }
     }
+
+    def getFailingCursor(boolean failImmediately) {
+        Mock(AsyncBatchCursor) {
+            def cursorResults = [[1, 2]]
+            def hasSetBatchSize = failImmediately
+            setBatchSize(_) >> {
+                if (!hasSetBatchSize) {
+                    hasSetBatchSize = true
+                } else {
+                    throw new MongoException("Failure")
+                }
+            }
+            next(_) >> {
+                it[0].onResult(cursorResults.isEmpty() ? null : cursorResults.remove(0), null)
+            }
+        }
+    }
 }

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

@@ -43,7 +43,6 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
         1 * block.apply(_)
     }
 
-
     def 'should call onComplete after all data has been consumed'() {
         given:
         SingleResultCallback<Integer> singleResultCallback
@@ -241,6 +240,24 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
         observer.assertTerminalEvent()
     }
 
+    def 'should call onError if the passed block errors'() {
+        given:
+        def observer = new TestObserver()
+        observe(new Block<SingleResultCallback<Integer>>() {
+            @Override
+            void apply(final SingleResultCallback<Integer> callback) {
+                throw new MongoException('failed');
+            }
+        }).subscribe(observer)
+
+        when:
+        observer.requestMore(1)
+
+        then:
+        observer.assertErrored()
+        observer.assertTerminalEvent()
+    }
+
     def getBlock() {
         getBlock(1)
     }