Fix cancellation order in ThrottleFirst (#7484)

akarnokd · web-flow · commit d6d8869c9f4d · 2022-11-08T14:39:15.000+01:00
* Fix cancellation order in ThrottleFirst

* Bump vanniktech maven publish

* Undo bump, seems to be gradle issue
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
@@ -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
@@ -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();
                 }
             }
         }

Original file line number	Diff line number	Diff line change
`@@ -105,9 +105,10 @@ public void onNext(T t) {`
`105`	`105`	`downstream.onNext(t);`
`106`	`106`	`BackpressureHelper.produced(this, 1);`
`107`	`107`	`} else {`
	`108`	`+ upstream.cancel();`
`108`	`109`	`done = true;`
`109`		`- cancel();`
`110`	`110`	`downstream.onError(MissingBackpressureException.createDefault());`
	`111`	`+ worker.dispose();`
`111`	`112`	`return;`
`112`	`113`	`}`
`113`	`114`
`@@ -122,10 +123,10 @@ public void onNext(T t) {`
`122`	`123`	`onDropped.accept(t);`
`123`	`124`	`} catch (Throwable ex) {`
`124`	`125`	`Exceptions.throwIfFatal(ex);`
`125`		`- downstream.onError(ex);`
`126`		`- worker.dispose();`
`127`	`126`	`upstream.cancel();`
`128`	`127`	`done = true;`
	`128`	`+ downstream.onError(ex);`
	`129`	`+ worker.dispose();`
`129`	`130`	`}`
`130`	`131`	`}`
`131`	`132`	`}`
Original file line number	Diff line number	Diff line change
`@@ -32,7 +32,7 @@ public final class ObservableThrottleFirstTimed<T> extends AbstractObservableWit`
`32`	`32`
`33`	`33`	`public ObservableThrottleFirstTimed(`
`34`	`34`	`ObservableSource<T> source,`
`35`		`- long timeout,`
	`35`	`+ long timeout,`
`36`	`36`	`TimeUnit unit,`
`37`	`37`	`Scheduler scheduler,`
`38`	`38`	`Consumer<? super T> onDropped) {`
`@@ -102,9 +102,9 @@ public void onNext(T t) {`
`102`	`102`	`onDropped.accept(t);`
`103`	`103`	`} catch (Throwable ex) {`
`104`	`104`	`Exceptions.throwIfFatal(ex);`
	`105`	`+ upstream.dispose();`
`105`	`106`	`downstream.onError(ex);`
`106`	`107`	`worker.dispose();`
`107`		`- upstream.dispose();`
`108`	`108`	`}`
`109`	`109`	`}`
`110`	`110`	`}`