Filter null values for Observables

JAVA-3382

Author: jstewart148

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

@@ -26,7 +26,6 @@
 
 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` */
@@ -37,7 +36,7 @@ abstract class AbstractSubscription<TResult> implements Subscription {
     private boolean isTerminated = false;
     /* protected by `this` */
 
-    private final ConcurrentLinkedQueue<Object> resultsQueue = new ConcurrentLinkedQueue<Object>();
+    private final ConcurrentLinkedQueue<TResult> resultsQueue = new ConcurrentLinkedQueue<TResult>();
 
     AbstractSubscription(final Observer<? super TResult> observer) {
         this.observer = observer;
@@ -110,9 +109,7 @@ synchronized long getRequested() {
     }
 
     void addToQueue(@Nullable final TResult result) {
-        if (result == null) {
-            resultsQueue.add(NULL_PLACEHOLDER);
-        } else {
+        if (result != null) {
             resultsQueue.add(result);
         }
     }
@@ -139,11 +136,8 @@ void onError(final Throwable t) {
         }
     }
 
-    private void onNext(@Nullable final TResult next) {
+    private void onNext(final TResult next) {
         if (!isTerminated()) {
-            if (next == null) {
-                throw new NullPointerException();
-            }
             try {
                 observer.onNext(next);
             } catch (Throwable t) {
@@ -216,11 +210,11 @@ private void processResultsQueue() {
                 processedCount = 0;
 
                 while (localWanted > 0) {
-                    Object item = resultsQueue.poll();
+                    TResult item = resultsQueue.poll();
                     if (item == null) {
                         break;
                     } else {
-                        onNext(item == NULL_PLACEHOLDER ? null : (TResult) item);
+                        onNext(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(1, null)
+                callback.onResult(null, null)
             }
         }).subscribe(observer)
 
@@ -266,7 +266,7 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
 
         then:
         observer.assertNoErrors()
-        observer.assertReceivedOnNext([1])
+        observer.assertReceivedOnNext([])
         observer.assertTerminalEvent()
     }
 
@@ -289,25 +289,6 @@ 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(){