Merge pull request ReactiveX#476 from benjchristensen/bugfix-asyncsubject-empty

Don't emit null onComplete when no onNext received in AsyncSubject

Author: benjchristensen

rxjava-core/src/main/java/rx/subjects/AsyncSubject.java

@@ -16,6 +16,7 @@
 package rx.subjects;
 
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
 import rx.Observer;
@@ -85,6 +86,7 @@ public void unsubscribe() {
 
     private final ConcurrentHashMap<Subscription, Observer<? super T>> observers;
     private final AtomicReference<T> currentValue;
+    private final AtomicBoolean hasValue = new AtomicBoolean();
 
     protected AsyncSubject(OnSubscribeFunc<T> onSubscribe, ConcurrentHashMap<Subscription, Observer<? super T>> observers) {
         super(onSubscribe);
@@ -96,9 +98,9 @@ protected AsyncSubject(OnSubscribeFunc<T> onSubscribe, ConcurrentHashMap<Subscri
     public void onCompleted() {
         T finalValue = currentValue.get();
         for (Observer<? super T> observer : observers.values()) {
-            observer.onNext(finalValue);
-        }
-        for (Observer<? super T> observer : observers.values()) {
+            if (hasValue.get()) {
+                observer.onNext(finalValue);
+            }
             observer.onCompleted();
         }
     }
@@ -112,6 +114,7 @@ public void onError(Throwable e) {
 
     @Override
     public void onNext(T args) {
+        hasValue.set(true);
         currentValue.set(args);
     }
 }

rxjava-core/src/test/java/rx/subjects/AsyncSubjectTest.java

@@ -1,12 +1,12 @@
 /**
  * Copyright 2013 Netflix, Inc.
- *
+ * 
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -19,6 +19,7 @@
 import static org.mockito.Mockito.*;
 
 import org.junit.Test;
+import org.mockito.InOrder;
 import org.mockito.Mockito;
 
 import rx.Observer;
@@ -42,10 +43,6 @@ public void testNeverCompleted() {
         subject.onNext("two");
         subject.onNext("three");
 
-        assertNeverCompletedObserver(aObserver);
-    }
-
-    private void assertNeverCompletedObserver(Observer<String> aObserver) {
         verify(aObserver, Mockito.never()).onNext(anyString());
         verify(aObserver, Mockito.never()).onError(testException);
         verify(aObserver, Mockito.never()).onCompleted();
@@ -64,10 +61,6 @@ public void testCompleted() {
         subject.onNext("three");
         subject.onCompleted();
 
-        assertCompletedObserver(aObserver);
-    }
-
-    private void assertCompletedObserver(Observer<String> aObserver) {
         verify(aObserver, times(1)).onNext("three");
         verify(aObserver, Mockito.never()).onError(any(Throwable.class));
         verify(aObserver, times(1)).onCompleted();
@@ -89,10 +82,6 @@ public void testError() {
         subject.onError(new Throwable());
         subject.onCompleted();
 
-        assertErrorObserver(aObserver);
-    }
-
-    private void assertErrorObserver(Observer<String> aObserver) {
         verify(aObserver, Mockito.never()).onNext(anyString());
         verify(aObserver, times(1)).onError(testException);
         verify(aObserver, Mockito.never()).onCompleted();
@@ -110,15 +99,14 @@ public void testUnsubscribeBeforeCompleted() {
         subject.onNext("two");
 
         subscription.unsubscribe();
-        assertNoOnNextEventsReceived(aObserver);
+
+        verify(aObserver, Mockito.never()).onNext(anyString());
+        verify(aObserver, Mockito.never()).onError(any(Throwable.class));
+        verify(aObserver, Mockito.never()).onCompleted();
 
         subject.onNext("three");
         subject.onCompleted();
 
-        assertNoOnNextEventsReceived(aObserver);
-    }
-
-    private void assertNoOnNextEventsReceived(Observer<String> aObserver) {
         verify(aObserver, Mockito.never()).onNext(anyString());
         verify(aObserver, Mockito.never()).onError(any(Throwable.class));
         verify(aObserver, Mockito.never()).onCompleted();
@@ -146,4 +134,21 @@ public void call(AsyncSubject<Object> DefaultSubject) {
                 null
                 );
     }
+
+    @Test
+    public void testEmptySubjectCompleted() {
+        AsyncSubject<String> subject = AsyncSubject.create();
+
+        @SuppressWarnings("unchecked")
+        Observer<String> aObserver = mock(Observer.class);
+        subject.subscribe(aObserver);
+
+        subject.onCompleted();
+
+        InOrder inOrder = inOrder(aObserver);
+        inOrder.verify(aObserver, never()).onNext(null);
+        inOrder.verify(aObserver, never()).onNext(any(String.class));
+        inOrder.verify(aObserver, times(1)).onCompleted();
+        inOrder.verifyNoMoreInteractions();
+    }
 }