Fix cancellation order in ThrottleFirst (ReactiveX#7484)

* Fix cancellation order in ThrottleFirst * Bump vanniktech maven publish * Undo bump, seems to be gradle issue
Desislav-Petrov · Nov 26, 2022 · 42b366d · 42b366d
1 parent 7a2792a
commit 42b366d
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 5 deletions.
diff --git a/...ain/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableThrottleFirstTimed.java b/...ain/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/...java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java b/...java/io/reactivex/rxjava3/internal/operators/observable/ObservableThrottleFirstTimed.java
@@ -32,7 +32,7 @@ public final class ObservableThrottleFirstTimed<T> extends AbstractObservableWit
 
     public ObservableThrottleFirstTimed(
             ObservableSource<T> source,
-            long timeout, 
+            long timeout,
             TimeUnit unit,
             Scheduler scheduler,
             Consumer<? super T> 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();
                 }
             }
         }