Observable.scan no seed fix post-terminal behaviour (#4904)

ReactiveX · Dec 4, 2016 · 34b4b6b · 34b4b6b
1 parent d52cce9
commit 34b4b6b
Show file tree

Hide file tree

Showing 2 changed files with 101 additions and 1 deletion.
diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableScan.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableScan.java
@@ -19,6 +19,7 @@
 import io.reactivex.functions.BiFunction;
 import io.reactivex.internal.disposables.DisposableHelper;
 import io.reactivex.internal.functions.ObjectHelper;
+import io.reactivex.plugins.RxJavaPlugins;
 
 public final class ObservableScan<T> extends AbstractObservableWithUpstream<T, T> {
     final BiFunction<T, T, T> accumulator;
@@ -39,6 +40,8 @@ static final class ScanObserver<T> implements Observer<T>, Disposable {
         Disposable s;
 
         T value;
+
+        boolean done;
 
         ScanObserver(Observer<? super T> actual, BiFunction<T, T, T> accumulator) {
             this.actual = actual;
@@ -67,6 +70,9 @@ public boolean isDisposed() {
 
         @Override
         public void onNext(T t) {
+            if (done) {
+                return;
+            }
             final Observer<? super T> a = actual;
             T v = value;
             if (v == null) {
@@ -80,7 +86,7 @@ public void onNext(T t) {
                 } catch (Throwable e) {
                     Exceptions.throwIfFatal(e);
                     s.dispose();
-                    a.onError(e);
+                    onError(e);
                     return;
                 }
 
@@ -91,11 +97,20 @@ public void onNext(T t) {
 
         @Override
         public void onError(Throwable t) {
+            if (done) {
+                RxJavaPlugins.onError(t);
+                return;
+            }
+            done = true;
             actual.onError(t);
         }
 
         @Override
         public void onComplete() {
+            if (done) {
+                return;
+            }
+            done = true;
             actual.onComplete();
         }
     }

diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableScanTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableScanTest.java
@@ -19,16 +19,20 @@
 
 import java.util.*;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
 import io.reactivex.*;
 import io.reactivex.Observable;
 import io.reactivex.Observer;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.disposables.Disposables;
 import io.reactivex.exceptions.TestException;
 import io.reactivex.functions.*;
 import io.reactivex.observers.*;
+import io.reactivex.plugins.RxJavaPlugins;
 import io.reactivex.subjects.PublishSubject;
 
 public class ObservableScanTest {
@@ -300,4 +304,85 @@ public Object apply(Object a, Object b) throws Exception {
             }
         }, false, 1, 1, 0, 0);
     }
+
+    @Test
+    public void testScanFunctionThrowsAndUpstreamErrorsDoesNotResultInTwoTerminalEvents() {
+        final RuntimeException err = new RuntimeException();
+        final RuntimeException err2 = new RuntimeException();
+        final List<Throwable> list = new CopyOnWriteArrayList<Throwable>();
+        final Consumer<Throwable> errorConsumer = new Consumer<Throwable>() {
+            @Override
+            public void accept(Throwable t) throws Exception {
+                list.add(t);
+            }};
+        try {
+            RxJavaPlugins.setErrorHandler(errorConsumer);
+            Observable.unsafeCreate(new ObservableSource<Integer>() {
+                @Override
+                public void subscribe(Observer<? super Integer> o) {
+                    Disposable d = Disposables.empty();
+                    o.onSubscribe(d);
+                    o.onNext(1);
+                    o.onNext(2);
+                    o.onError(err2);
+                }})
+            .scan(new BiFunction<Integer,Integer,Integer>() {
+                @Override
+                public Integer apply(Integer t1, Integer t2) throws Exception {
+                    throw err;
+                }})
+            .test()
+            .assertError(err)
+            .assertValue(1);
+        } finally {
+            RxJavaPlugins.reset();
+        }
+    }
+
+    @Test
+    public void testScanFunctionThrowsAndUpstreamCompletesDoesNotResultInTwoTerminalEvents() {
+        final RuntimeException err = new RuntimeException();
+        Observable.unsafeCreate(new ObservableSource<Integer>() {
+            @Override
+            public void subscribe(Observer<? super Integer> o) {
+                Disposable d = Disposables.empty();
+                o.onSubscribe(d);
+                o.onNext(1);
+                o.onNext(2);
+                o.onComplete();
+            }})
+        .scan(new BiFunction<Integer,Integer,Integer>() {
+            @Override
+            public Integer apply(Integer t1, Integer t2) throws Exception {
+                throw err;
+            }})
+        .test()
+        .assertError(err)
+        .assertValue(1);
+    }
+
+    @Test
+    public void testScanFunctionThrowsAndUpstreamEmitsOnNextResultsInScanFunctionBeingCalledOnlyOnce() {
+        final RuntimeException err = new RuntimeException();
+        final AtomicInteger count = new AtomicInteger();
+        Observable.unsafeCreate(new ObservableSource<Integer>() {
+            @Override
+            public void subscribe(Observer<? super Integer> o) {
+                Disposable d = Disposables.empty();
+                o.onSubscribe(d);
+                o.onNext(1);
+                o.onNext(2);
+                o.onNext(3);
+            }})
+        .scan(new BiFunction<Integer,Integer,Integer>() {
+            @Override
+            public Integer apply(Integer t1, Integer t2) throws Exception {
+                count.incrementAndGet();
+                throw err;
+            }})
+        .test()
+        .assertError(err)
+        .assertValue(1);
+        assertEquals(1, count.get());
+    }
 }