From 2caafb51f0aecc8d6d3bf531cf1c824ed732f4f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Mon, 7 Nov 2022 22:49:36 +0100 Subject: [PATCH 1/3] Fix cancellation order in ThrottleFirst --- .../operators/flowable/FlowableThrottleFirstTimed.java | 7 ++++--- .../operators/observable/ObservableThrottleFirstTimed.java | 4 ++-- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java index 223076dfc7..44499ec255 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java @@ -105,9 +105,10 @@ public void onNext(T t) { downstream.onNext(t); BackpressureHelper.produced(this, 1); } else { + upstream.cancel(); done = true; - cancel(); downstream.onError(MissingBackpressureException.createDefault()); + worker.dispose(); return; } @@ -122,10 +123,10 @@ public void onNext(T t) { onDropped.accept(t); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); - downstream.onError(ex); - worker.dispose(); upstream.cancel(); done = true; + downstream.onError(ex); + worker.dispose(); } } } diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java index aa0f5058fd..6bf3b9f119 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java @@ -32,7 +32,7 @@ public final class ObservableThrottleFirstTimed extends AbstractObservableWit public ObservableThrottleFirstTimed( ObservableSource source, - long timeout, + long timeout, TimeUnit unit, Scheduler scheduler, Consumer onDropped) { @@ -102,9 +102,9 @@ public void onNext(T t) { onDropped.accept(t); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); + upstream.dispose(); downstream.onError(ex); worker.dispose(); - upstream.dispose(); } } } From f70c79de215aad65b25c5192d83098bddaf0b19d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Mon, 7 Nov 2022 22:59:32 +0100 Subject: [PATCH 2/3] Bump vanniktech maven publish --- build.gradle | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build.gradle b/build.gradle index ec0e5b9dde..2b80cfab4d 100644 --- a/build.gradle +++ b/build.gradle @@ -8,7 +8,7 @@ plugins { id("me.champeau.gradle.jmh") version "0.5.3" id("com.github.hierynomus.license") version "0.16.1" id("biz.aQute.bnd.builder") version "6.3.1" - id("com.vanniktech.maven.publish") version "0.19.0" + id("com.vanniktech.maven.publish") version "0.20.0" id("org.beryx.jar") version "1.2.0" } From c9e05b38e6aab552679589b2828d7cea36cd8c06 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?D=C3=A1vid=20Karnok?= Date: Mon, 7 Nov 2022 23:03:52 +0100 Subject: [PATCH 3/3] Undo bump, seems to be gradle issue --- build.gradle | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/build.gradle b/build.gradle index 2b80cfab4d..ec0e5b9dde 100644 --- a/build.gradle +++ b/build.gradle @@ -8,7 +8,7 @@ plugins { id("me.champeau.gradle.jmh") version "0.5.3" id("com.github.hierynomus.license") version "0.16.1" id("biz.aQute.bnd.builder") version "6.3.1" - id("com.vanniktech.maven.publish") version "0.20.0" + id("com.vanniktech.maven.publish") version "0.19.0" id("org.beryx.jar") version "1.2.0" }