2.x: fix LambdaObserver not cancelling the upstream (#5170)

ReactiveX · Mar 10, 2017 · f059ded · f059ded
1 parent b5501c5
commit f059ded
Show file tree

Hide file tree

Showing 6 changed files with 139 additions and 13 deletions.
diff --git a/src/main/java/io/reactivex/Observable.java b/src/main/java/io/reactivex/Observable.java
@@ -10894,7 +10894,7 @@ public final <R> Observable<R> switchMap(Function<? super T, ? extends Observabl
     public final <R> Observable<R> switchMapSingle(@NonNull Function<? super T, ? extends SingleSource<? extends R>> mapper) {
         return ObservableInternalHelper.switchMapSingle(this, mapper);
     }
-    
+
     /**
      * Returns a new ObservableSource by applying a function that you supply to each item emitted by the source
      * ObservableSource that returns a SingleSource, and then emitting the item emitted by the most recently emitted
@@ -10925,7 +10925,7 @@ public final <R> Observable<R> switchMapSingle(@NonNull Function<? super T, ? ex
     public final <R> Observable<R> switchMapSingleDelayError(@NonNull Function<? super T, ? extends SingleSource<? extends R>> mapper) {
         return ObservableInternalHelper.switchMapSingleDelayError(this, mapper);
     }
-    
+
     /**
      * Returns a new ObservableSource by applying a function that you supply to each item emitted by the source
      * ObservableSource that returns an ObservableSource, and then emitting the items emitted by the most recently emitted

diff --git a/src/main/java/io/reactivex/internal/observers/LambdaObserver.java b/src/main/java/io/reactivex/internal/observers/LambdaObserver.java
@@ -47,6 +47,7 @@ public void onSubscribe(Disposable s) {
                 onSubscribe.accept(this);
             } catch (Throwable ex) {
                 Exceptions.throwIfFatal(ex);
+                s.dispose();
                 onError(ex);
             }
         }
@@ -59,6 +60,7 @@ public void onNext(T t) {
                 onNext.accept(t);
             } catch (Throwable e) {
                 Exceptions.throwIfFatal(e);
+                get().dispose();
                 onError(e);
             }
         }

diff --git a/src/main/java/io/reactivex/internal/operators/observable/ObservableInternalHelper.java b/src/main/java/io/reactivex/internal/operators/observable/ObservableInternalHelper.java
@@ -326,27 +326,27 @@ public static <T,R> Observable<R> switchMapSingleDelayError(Observable<T> source
             Function<? super T, ? extends SingleSource<? extends R>> mapper) {
         return source.switchMapDelayError(convertSingleMapperToObservableMapper(mapper), 1);
     }
-    
+
     private static <T, R> Function<T, Observable<R>> convertSingleMapperToObservableMapper(
             final Function<? super T, ? extends SingleSource<? extends R>> mapper) {
         ObjectHelper.requireNonNull(mapper, "mapper is null");
         return new ObservableMapper<T,R>(mapper);
     }
-    
+
     static final class ObservableMapper<T,R> implements Function<T,Observable<R>> {
-        
+
         final Function<? super T, ? extends SingleSource<? extends R>> mapper;
 
         ObservableMapper(Function<? super T, ? extends SingleSource<? extends R>> mapper) {
             this.mapper = mapper;
         }
-        
+
         @Override
         public Observable<R> apply(T t) throws Exception {
             return RxJavaPlugins.onAssembly(new SingleToObservable<R>(
                 ObjectHelper.requireNonNull(mapper.apply(t), "The mapper returned a null value")));
         }
-        
+
     }
 
 }
diff --git a/src/test/java/io/reactivex/internal/observers/LambdaObserverTest.java b/src/test/java/io/reactivex/internal/observers/LambdaObserverTest.java
@@ -26,6 +26,7 @@
 import io.reactivex.exceptions.*;
 import io.reactivex.functions.*;
 import io.reactivex.plugins.RxJavaPlugins;
+import io.reactivex.subjects.PublishSubject;
 
 public class LambdaObserverTest {
 
@@ -280,4 +281,65 @@ public void accept(Disposable s) throws Exception {
 
         assertEquals(Arrays.asList(1, 100), received);
     }
+
+    @Test
+    public void onNextThrowsCancelsUpstream() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        final List<Throwable> errors = new ArrayList<Throwable>();
+
+        ps.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer v) throws Exception {
+                throw new TestException();
+            }
+        }, new Consumer<Throwable>() {
+            @Override
+            public void accept(Throwable e) throws Exception {
+                errors.add(e);
+            }
+        });
+
+        assertTrue("No observers?!", ps.hasObservers());
+        assertTrue("Has errors already?!", errors.isEmpty());
+
+        ps.onNext(1);
+
+        assertFalse("Has observers?!", ps.hasObservers());
+        assertFalse("No errors?!", errors.isEmpty());
+
+        assertTrue(errors.toString(), errors.get(0) instanceof TestException);
+    }
+
+    @Test
+    public void onSubscribeThrowsCancelsUpstream() {
+        PublishSubject<Integer> ps = PublishSubject.create();
+
+        final List<Throwable> errors = new ArrayList<Throwable>();
+
+        ps.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer v) throws Exception {
+            }
+        }, new Consumer<Throwable>() {
+            @Override
+            public void accept(Throwable e) throws Exception {
+                errors.add(e);
+            }
+        }, new Action() {
+            @Override
+            public void run() throws Exception {
+            }
+        }, new Consumer<Disposable>() {
+            @Override
+            public void accept(Disposable s) throws Exception {
+                throw new TestException();
+            }
+        });
+
+        assertFalse("Has observers?!", ps.hasObservers());
+        assertFalse("No errors?!", errors.isEmpty());
+
+        assertTrue(errors.toString(), errors.get(0) instanceof TestException);
+    }
 }
diff --git a/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java b/src/test/java/io/reactivex/internal/operators/observable/ObservableSwitchTest.java
@@ -622,7 +622,7 @@ public void switchMapInnerCancelled() {
 
         assertFalse(pp.hasObservers());
     }
-    
+
     @Test
     public void switchMapSingleJustSource() {
         Observable.just(0)
@@ -635,7 +635,7 @@ public SingleSource<Integer> apply(Object v) throws Exception {
         .test()
         .assertResult(1);
     }
-    
+
     @Test
     public void switchMapSingleMapperReturnsNull() {
         Observable.just(0)
@@ -648,13 +648,13 @@ public SingleSource<Integer> apply(Object v) throws Exception {
         .test()
         .assertError(NullPointerException.class);
     }
-    
-    @Test(expected=NullPointerException.class)
+
+    @Test(expected = NullPointerException.class)
     public void switchMapSingleMapperIsNull() {
         Observable.just(0)
         .switchMapSingle(null);
     }
-    
+
     @Test
     public void switchMapSingleFunctionDoesntReturnSingle() {
         Observable.just(0)
@@ -698,7 +698,7 @@ public void accept(Integer n) throws Exception {
         .assertError(RuntimeException.class);
         assertTrue(completed.get());
     }
-    
+
     @Test
     public void scalarMap() {
         Observable.switchOnNext(Observable.just(Observable.just(1)))

diff --git a/src/test/java/io/reactivex/internal/subscribers/LambdaSubscriberTest.java b/src/test/java/io/reactivex/internal/subscribers/LambdaSubscriberTest.java
@@ -25,6 +25,7 @@
 import io.reactivex.functions.*;
 import io.reactivex.internal.subscriptions.BooleanSubscription;
 import io.reactivex.plugins.RxJavaPlugins;
+import io.reactivex.processors.PublishProcessor;
 
 public class LambdaSubscriberTest {
 
@@ -285,4 +286,65 @@ public void accept(Subscription s) throws Exception {
 
         assertEquals(Arrays.asList(1, 100), received);
     }
+
+    @Test
+    public void onNextThrowsCancelsUpstream() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        final List<Throwable> errors = new ArrayList<Throwable>();
+
+        ps.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer v) throws Exception {
+                throw new TestException();
+            }
+        }, new Consumer<Throwable>() {
+            @Override
+            public void accept(Throwable e) throws Exception {
+                errors.add(e);
+            }
+        });
+
+        assertTrue("No observers?!", ps.hasSubscribers());
+        assertTrue("Has errors already?!", errors.isEmpty());
+
+        ps.onNext(1);
+
+        assertFalse("Has observers?!", ps.hasSubscribers());
+        assertFalse("No errors?!", errors.isEmpty());
+
+        assertTrue(errors.toString(), errors.get(0) instanceof TestException);
+    }
+
+    @Test
+    public void onSubscribeThrowsCancelsUpstream() {
+        PublishProcessor<Integer> ps = PublishProcessor.create();
+
+        final List<Throwable> errors = new ArrayList<Throwable>();
+
+        ps.subscribe(new Consumer<Integer>() {
+            @Override
+            public void accept(Integer v) throws Exception {
+            }
+        }, new Consumer<Throwable>() {
+            @Override
+            public void accept(Throwable e) throws Exception {
+                errors.add(e);
+            }
+        }, new Action() {
+            @Override
+            public void run() throws Exception {
+            }
+        }, new Consumer<Subscription>() {
+            @Override
+            public void accept(Subscription s) throws Exception {
+                throw new TestException();
+            }
+        });
+
+        assertFalse("Has observers?!", ps.hasSubscribers());
+        assertFalse("No errors?!", errors.isEmpty());
+
+        assertTrue(errors.toString(), errors.get(0) instanceof TestException);
+    }
 }