From 25f2838eb5e945876d15fa39994510efb182bc70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Wed, 23 Sep 2020 08:40:25 +0200 Subject: [PATCH] 2.x: Fix toFlowable(ERROR) not cancelling upon MBE --- .../flowable/FlowableOnBackpressureError.java | 1 + .../FlowableOnBackpressureErrorTest.java | 21 +++++++++++++++++++ 2 files changed, 22 insertions(+) diff --git a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureError.java b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureError.java index f60c43bc31..556e54b8c3 100644 --- a/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureError.java +++ b/src/main/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureError.java @@ -65,6 +65,7 @@ public void onNext(T t) { downstream.onNext(t); BackpressureHelper.produced(this, 1); } else { + upstream.cancel(); onError(new MissingBackpressureException("could not emit value due to lack of requests")); } } diff --git a/src/test/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureErrorTest.java b/src/test/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureErrorTest.java index 3be409e862..9afb864d40 100644 --- a/src/test/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureErrorTest.java +++ b/src/test/java/io/reactivex/internal/operators/flowable/FlowableOnBackpressureErrorTest.java @@ -13,11 +13,16 @@ package io.reactivex.internal.operators.flowable; +import static org.junit.Assert.*; + import org.junit.Test; import org.reactivestreams.Publisher; import io.reactivex.*; +import io.reactivex.exceptions.MissingBackpressureException; import io.reactivex.functions.Function; +import io.reactivex.subjects.PublishSubject; +import io.reactivex.subscribers.TestSubscriber; public class FlowableOnBackpressureErrorTest { @@ -50,4 +55,20 @@ public Object apply(Flowable f) throws Exception { } }, false, 1, 1, 1); } + + @Test + public void overflowCancels() { + PublishSubject ps = PublishSubject.create(); + + TestSubscriber ts = ps.toFlowable(BackpressureStrategy.ERROR) + .test(0L); + + assertTrue(ps.hasObservers()); + + ps.onNext(1); + + assertFalse(ps.hasObservers()); + + ts.assertFailure(MissingBackpressureException.class); + } }