collect, DeferredScalarSubscriber - prevent multiple terminal emissions

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

@@ -0,0 +1,57 @@
+/**
+ * Copyright 2016 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
+ *
+ * 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. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package rx.internal.operators;
+
+import rx.Subscriber;
+import rx.plugins.RxJavaHooks;
+
+/**
+ * Supplements {@code DeferredScalarSubscriber} with defensive behaviour that ensures no emissions
+ * occur after a terminal event. If {@code onError} is called more than once then errors after the first
+ * are reported to {@code RxJavaHooks.onError}.
+ * 
+ * @param <T> source value type
+ * @param <R> result value type
+ */
+public abstract class DeferredScalarSubscriberSafe<T, R> extends DeferredScalarSubscriber<T,R> {
+
+    protected boolean done;
+
+    public DeferredScalarSubscriberSafe(Subscriber<? super R> actual) {
+        super(actual);
+    }
+
+    @Override
+    public void onError(Throwable ex) {
+        if (!done) {
+            done = true;
+            super.onError(ex);
+        } else {
+            RxJavaHooks.onError(ex);
+        }
+    }
+
+    @Override
+    public void onCompleted() {
+        if (done) {
+            return;
+        }
+        done = true;
+        super.onCompleted();
+    }
+
+}

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

@@ -50,7 +50,7 @@ public void call(Subscriber<? super R> t) {
         new CollectSubscriber<T, R>(t, initialValue, collector).subscribeTo(source);
     }
 
-    static final class CollectSubscriber<T, R> extends DeferredScalarSubscriber<T, R> {
+    static final class CollectSubscriber<T, R> extends DeferredScalarSubscriberSafe<T, R> {
 
         final Action2<R, ? super T> collector;
 
@@ -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/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>();