collect - prevent multiple terminal emissions, add done to DeferredScalarSubscriber

Author: davidmoten

src/main/java/rx/internal/operators/DeferredScalarSubscriber.java

@@ -19,6 +19,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import rx.*;
+import rx.plugins.RxJavaHooks;
 
 /**
  * Base class for Subscribers that consume the entire upstream and signal
@@ -56,19 +57,31 @@ public abstract class DeferredScalarSubscriber<T, R> extends Subscriber<T> {
     /** Value will be emitted. */
     static final int HAS_REQUEST_HAS_VALUE = 3;
 
+    /** prevents multiple terminal emissions. */
+    protected boolean done;
+    
     public DeferredScalarSubscriber(Subscriber<? super R> actual) {
         this.actual = actual;
         this.state = new AtomicInteger();
     }
 
     @Override
     public void onError(Throwable ex) {
-        value = null;
-        actual.onError(ex);
+        if (!done) {
+            done = true;
+            value = null;
+            actual.onError(ex);
+        } else {
+            RxJavaHooks.onError(ex);
+        }
     }
 
     @Override
     public void onCompleted() {
+        if (done) {
+            return;
+        }
+        done = true;
         if (hasValue) {
             complete(value);
         } else {

src/main/java/rx/internal/operators/OnSubscribeCollect.java

@@ -63,12 +63,15 @@ public CollectSubscriber(Subscriber<? super R> actual, R initialValue, Action2<R
 
         @Override
         public void onNext(T t) {
+            if (done) {
+                return;
+            }
             try {
                 collector.call(value, t);
             } catch (Throwable ex) {
                 Exceptions.throwIfFatal(ex);
                 unsubscribe();
-                actual.onError(ex);
+                onError(ex);
             }
         }
 

src/main/java/rx/internal/operators/OnSubscribeReduce.java

@@ -22,6 +22,7 @@
 import rx.Observable.OnSubscribe;
 import rx.exceptions.Exceptions;
 import rx.functions.Func2;
+import rx.plugins.RxJavaHooks;
 
 public final class OnSubscribeReduce<T> implements OnSubscribe<T> {
 
@@ -57,6 +58,8 @@ static final class ReduceSubscriber<T> extends Subscriber<T> {
 
         static final Object EMPTY = new Object();
 
+        boolean done;
+        
         @SuppressWarnings("unchecked")
         public ReduceSubscriber(Subscriber<? super T> actual, Func2<T, T, T> reducer) {
             this.actual = actual;
@@ -68,6 +71,9 @@ public ReduceSubscriber(Subscriber<? super T> actual, Func2<T, T, T> reducer) {
         @SuppressWarnings("unchecked")
         @Override
         public void onNext(T t) {
+            if (done) {
+                return;
+            }
             Object o = value;
             if (o == EMPTY) {
                 value = t;
@@ -77,19 +83,28 @@ public void onNext(T t) {
                 } catch (Throwable ex) {
                     Exceptions.throwIfFatal(ex);
                     unsubscribe();
-                    actual.onError(ex);
+                    onError(ex);
                 }
             }
         }
 
         @Override
         public void onError(Throwable e) {
-            actual.onError(e);
+            if (!done) {
+                done = true;
+                actual.onError(e);
+            } else {
+                RxJavaHooks.onError(e);
+            }
         }
 
         @SuppressWarnings("unchecked")
         @Override
         public void onCompleted() {
+            if (done) {
+                return;
+            }
+            done = true;
             Object o = value;
             if (o != EMPTY) {
                 actual.onNext((T)o);

src/main/java/rx/internal/operators/OperatorAny.java

@@ -21,6 +21,7 @@
 import rx.exceptions.Exceptions;
 import rx.functions.Func1;
 import rx.internal.producers.SingleDelayedProducer;
+import rx.plugins.RxJavaHooks;
 
 /**
  * Returns an {@link Observable} that emits <code>true</code> if any element of
@@ -45,6 +46,9 @@ public Subscriber<? super T> call(final Subscriber<? super Boolean> child) {
 
             @Override
             public void onNext(T t) {
+                if (done) {
+                    return;
+                }
                 hasElements = true;
                 boolean result;
                 try {
@@ -53,7 +57,7 @@ public void onNext(T t) {
                     Exceptions.throwOrReport(e, this, t);
                     return;
                 }
-                if (result && !done) {
+                if (result) {
                     done = true;
                     producer.setValue(!returnOnEmpty);
                     unsubscribe();
@@ -64,7 +68,12 @@ public void onNext(T t) {
 
             @Override
             public void onError(Throwable e) {
-                child.onError(e);
+                if (!done) {
+                    done = true;
+                    child.onError(e);
+                } else {
+                    RxJavaHooks.onError(e);
+                }
             }
 
             @Override

src/test/java/rx/ObservableTests.java

@@ -939,62 +939,7 @@ public void testRangeWithScheduler() {
         inOrder.verifyNoMoreInteractions();
     }
 
-    @Test
-    public void testCollectToList() {
-        Observable<List<Integer>> o = Observable.just(1, 2, 3).collect(new Func0<List<Integer>>() {
-
-            @Override
-            public List<Integer> call() {
-                return new ArrayList<Integer>();
-            }
-            
-        }, new Action2<List<Integer>, Integer>() {
-
-            @Override
-            public void call(List<Integer> list, Integer v) {
-                list.add(v);
-            }
-        });
 
-        List<Integer> list =  o.toBlocking().last();
-
-        assertEquals(3, list.size());
-        assertEquals(1, list.get(0).intValue());
-        assertEquals(2, list.get(1).intValue());
-        assertEquals(3, list.get(2).intValue());
-        
-        // test multiple subscribe
-        List<Integer> list2 =  o.toBlocking().last();
-
-        assertEquals(3, list2.size());
-        assertEquals(1, list2.get(0).intValue());
-        assertEquals(2, list2.get(1).intValue());
-        assertEquals(3, list2.get(2).intValue());
-    }
-
-    @Test
-    public void testCollectToString() {
-        String value = Observable.just(1, 2, 3).collect(new Func0<StringBuilder>() {
-
-            @Override
-            public StringBuilder call() {
-                return new StringBuilder();
-            }
-            
-        }, new Action2<StringBuilder, Integer>() {
-
-            @Override
-            public void call(StringBuilder sb, Integer v) {
-                if (sb.length() > 0) {
-                    sb.append("-");
-                }
-                sb.append(v);
-            }
-        }).toBlocking().last().toString();
-
-        assertEquals("1-2-3", value);
-    }
-    
     @Test
     public void testMergeWith() {
         TestSubscriber<Integer> ts = new TestSubscriber<Integer>();