Made checkCompleted abstract in AbstractSubscriptions

There is no guarantee that there would be single calls to request before data arrives.
Therefore checkCompleted should not default to true.

Tracking completed state fixes a potential race conditions for Subscriptions to SingleResultCallbacks.

JAVA-1973

Author: rozza

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

@@ -93,9 +93,7 @@ void requestMoreData() {
     void postTerminate() {
     }
 
-    boolean checkCompleted() {
-        return true;
-    }
+    abstract boolean checkCompleted();
 
     boolean isTerminated() {
         return isTerminated;

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

@@ -25,6 +25,10 @@ class FlatteningSingleResultCallbackSubscription<TResult> extends AbstractSubscr
 
     private final Block<SingleResultCallback<List<TResult>>> block;
 
+    /* protected by `this` */
+    private boolean completed;
+    /* protected by `this` */
+
     public FlatteningSingleResultCallbackSubscription(final Block<SingleResultCallback<List<TResult>>> block,
                                                       final Observer<? super TResult> observer) {
         super(observer);
@@ -40,9 +44,17 @@ public void onResult(final List<TResult> result, final Throwable t) {
                 if (t != null) {
                     onError(t);
                 } else {
+                    synchronized (FlatteningSingleResultCallbackSubscription.this) {
+                        completed = true;
+                    }
                     addToQueue(result);
                 }
             }
         });
     }
+
+    @Override
+    boolean checkCompleted() {
+        return completed;
+    }
 }

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

@@ -23,6 +23,10 @@ class SingleResultCallbackSubscription<TResult> extends AbstractSubscription<TRe
 
     private final Block<SingleResultCallback<TResult>> block;
 
+    /* protected by `this` */
+    private boolean completed;
+    /* protected by `this` */
+
     public SingleResultCallbackSubscription(final Block<SingleResultCallback<TResult>> block,
                                             final Observer<? super TResult> observer) {
         super(observer);
@@ -38,9 +42,17 @@ public void onResult(final TResult result, final Throwable t) {
                 if (t != null) {
                     onError(t);
                 } else {
+                    synchronized (SingleResultCallbackSubscription.this) {
+                        completed = true;
+                    }
                     addToQueue(result);
                 }
             }
         });
     }
+
+    @Override
+    boolean checkCompleted() {
+        return completed;
+    }
 }

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

@@ -43,15 +43,28 @@ class FlatteningSingleResultCallbackSubscriptionSpecification extends Specificat
         1 * block.apply(_)
     }
 
-
     def 'should call onComplete after all data has been consumed'() {
         given:
-        def block = getBlock()
+        SingleResultCallback<List> listSingleResultCallback
         def observer = new TestObserver()
-        observeAndFlatten(block).subscribe(observer)
+        observeAndFlatten(new Block<SingleResultCallback<List>>() {
+            @Override
+            void apply(final SingleResultCallback<List> callback) {
+                listSingleResultCallback = callback
+            }
+        }).subscribe(observer)
+
+        when:
+        observer.requestMore(5)
+        observer.requestMore(5)
+
+        then:
+        observer.assertNoErrors()
+        observer.assertReceivedOnNext([])
+        observer.assertNoTerminalEvent()
 
         when:
-        observer.requestMore(10)
+        listSingleResultCallback.onResult([1, 2, 3, 4], null)
 
         then:
         observer.assertNoErrors()

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

@@ -46,12 +46,26 @@ class SingleResultCallbackSubscriptionSpecification extends Specification {
 
     def 'should call onComplete after all data has been consumed'() {
         given:
-        def block = getBlock()
+        SingleResultCallback<Integer> singleResultCallback
         def observer = new TestObserver()
-        observe(block).subscribe(observer)
+        observe(new Block<SingleResultCallback<Integer>>() {
+            @Override
+            void apply(final SingleResultCallback<Integer> callback) {
+                singleResultCallback = callback
+            }
+        }).subscribe(observer)
 
         when:
-        observer.requestMore(10)
+        observer.requestMore(5)
+        observer.requestMore(5)
+
+        then:
+        observer.assertNoErrors()
+        observer.assertReceivedOnNext([])
+        observer.assertNoTerminalEvent()
+
+        when:
+        singleResultCallback.onResult(1, null)
 
         then:
         observer.assertNoErrors()