Handle null results for reactive applications

JAVA-3279

Author: jstewart148

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

@@ -26,6 +26,7 @@
 
 abstract class AbstractSubscription<TResult> implements Subscription {
     private static final Logger LOGGER = Loggers.getLogger("client");
+    private static final Object NULL_PLACEHOLDER = new Object();
     private final Observer<? super TResult> observer;
 
     /* protected by `this` */
@@ -36,7 +37,7 @@ abstract class AbstractSubscription<TResult> implements Subscription {
     private boolean isTerminated = false;
     /* protected by `this` */
 
-    private final ConcurrentLinkedQueue<TResult> resultsQueue = new ConcurrentLinkedQueue<TResult>();
+    private final ConcurrentLinkedQueue<Object> resultsQueue = new ConcurrentLinkedQueue<Object>();
 
     AbstractSubscription(final Observer<? super TResult> observer) {
         this.observer = observer;
@@ -109,14 +110,18 @@ synchronized long getRequested() {
     }
 
     void addToQueue(@Nullable final TResult result) {
-        if (result != null) {
+        if (result == null) {
+            resultsQueue.add(NULL_PLACEHOLDER);
+        } else {
             resultsQueue.add(result);
         }
     }
 
     void addToQueue(@Nullable final List<TResult> results) {
         if (results != null) {
-            resultsQueue.addAll(results);
+            for (TResult cur : results) {
+                addToQueue(cur);
+            }
         }
     }
 
@@ -134,8 +139,11 @@ void onError(final Throwable t) {
         }
     }
 
-    void onNext(final TResult next) {
+    private void onNext(@Nullable final TResult next) {
         if (!isTerminated()) {
+            if (next == null) {
+                throw new NullPointerException();
+            }
             try {
                 observer.onNext(next);
             } catch (Throwable t) {
@@ -145,7 +153,7 @@ void onNext(final TResult next) {
         }
     }
 
-    void onComplete() {
+    private void onComplete() {
         if (terminalAction()) {
             postTerminate();
             try {
@@ -173,6 +181,7 @@ void tryProcessResultsQueue() {
         }
     }
 
+    @SuppressWarnings("unchecked")
     private void processResultsQueue() {
         boolean mustProcess = false;
 
@@ -207,11 +216,11 @@ private void processResultsQueue() {
                 processedCount = 0;
 
                 while (localWanted > 0) {
-                    TResult item = resultsQueue.poll();
+                    Object item = resultsQueue.poll();
                     if (item == null) {
                         break;
                     } else {
-                        onNext(item);
+                        onNext(item == NULL_PLACEHOLDER ? null : (TResult) item);
                         localWanted -= 1;
                         processedCount += 1;
                     }

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

@@ -257,7 +257,7 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
         observe(new Block<SingleResultCallback<Void>>(){
             @Override
             void apply(final SingleResultCallback<Void> callback) {
-                callback.onResult(null, null)
+                callback.onResult(1, null)
             }
         }).subscribe(observer)
 
@@ -266,7 +266,7 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
 
         then:
         observer.assertNoErrors()
-        observer.assertReceivedOnNext([])
+        observer.assertReceivedOnNext([1])
         observer.assertTerminalEvent()
     }
 
@@ -289,6 +289,25 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
         observer.assertErrored()
     }
 
+    def 'should call onError on a null result'() {
+        given:
+        def observer = new TestObserver()
+        observe(new Block<SingleResultCallback<Void>>() {
+            @Override
+            void apply(final SingleResultCallback<Void> callback) {
+                callback.onResult(null, null);
+            }
+        }).subscribe(observer)
+
+        when:
+        observer.requestMore(1)
+
+        then:
+        observer.assertTerminalEvent()
+        observer.assertErrored()
+        observer.getOnErrorEvents().first() instanceof NullPointerException
+    }
+
     def 'should throw the exception if calling onComplete raises one'() {
         given:
         def observer = new TestObserver(new Observer(){