3.x: onReduceBackpressure internals cleanup (#7151)

* 3.x: onReduceBackpressure internals cleanup

* Update FlowableOnBackpressureReduceTest.java

* Update FlowableOnBackpressureReduceWithTest.java

Author: akarnokd

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureReduce.java

@@ -49,19 +49,20 @@ static final class BackpressureReduceSubscriber<T> extends AbstractBackpressureT
         @Override
         public void onNext(T t) {
             T v = current.get();
+            if (v != null) {
+                v = current.getAndSet(null);
+            }
             if (v == null) {
                 current.lazySet(t);
-            } else if ((v = current.getAndSet(null)) != null) {
+            } else {
                 try {
                     current.lazySet(Objects.requireNonNull(reducer.apply(v, t), "The reducer returned a null value"));
                 } catch (Throwable ex) {
                     Exceptions.throwIfFatal(ex);
+                    upstream.cancel();
                     onError(ex);
-                    cancel();
                     return;
                 }
-            } else {
-                current.lazySet(t);
             }
             drain();
         }

src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureReduceWith.java

@@ -58,24 +58,23 @@ static final class BackpressureReduceWithSubscriber<T, R> extends AbstractBackpr
         @Override
         public void onNext(T t) {
             R v = current.get();
+            if (v != null) {
+                v = current.getAndSet(null);
+            }
             try {
                 if (v == null) {
                     current.lazySet(Objects.requireNonNull(
                             reducer.apply(Objects.requireNonNull(supplier.get(), "The supplier returned a null value"), t),
                             "The reducer returned a null value"
                     ));
-                } else if ((v = current.getAndSet(null)) != null) {
-                    current.lazySet(Objects.requireNonNull(reducer.apply(v, t), "The reducer returned a null value"));
                 } else {
-                    current.lazySet(Objects.requireNonNull(
-                            reducer.apply(Objects.requireNonNull(supplier.get(), "The supplier returned a null value"), t),
-                            "The reducer returned a null value"
-                    ));
+                    current.lazySet(Objects.requireNonNull(reducer.apply(v, t), "The reducer returned a null value"));
                 }
             } catch (Throwable ex) {
                 Exceptions.throwIfFatal(ex);
+                upstream.cancel();
                 onError(ex);
-                cancel();
+                return;
             }
             drain();
         }

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureReduceTest.java

@@ -110,6 +110,30 @@ public void synchronousDrop() {
         ts.assertTerminated();
     }
 
+    @Test
+    public void reduceBackpressuredSync() {
+        PublishProcessor<Integer> source = PublishProcessor.create();
+        TestSubscriberEx<Integer> ts = new TestSubscriberEx<>(0L);
+
+        source.onBackpressureReduce(Integer::sum).subscribe(ts);
+
+        source.onNext(1);
+        source.onNext(2);
+        source.onNext(3);
+
+        ts.request(1);
+
+        ts.assertValuesOnly(6);
+
+        source.onNext(4);
+        source.onComplete();
+
+        ts.assertValuesOnly(6);
+
+        ts.request(1);
+        ts.assertResult(6, 4);
+    }
+
     private <T> TestSubscriberEx<T> createDelayedSubscriber() {
         return new TestSubscriberEx<T>(1L) {
             final Random rnd = new Random();

src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureReduceWithTest.java

@@ -90,6 +90,30 @@ public void simpleBackpressure() {
         ts.assertNotComplete();
     }
 
+    @Test
+    public void reduceBackpressuredSync() {
+        PublishProcessor<Integer> source = PublishProcessor.create();
+        TestSubscriberEx<Integer> ts = new TestSubscriberEx<>(0L);
+
+        source.onBackpressureReduce(() -> 0, Integer::sum).subscribe(ts);
+
+        source.onNext(1);
+        source.onNext(2);
+        source.onNext(3);
+
+        ts.request(1);
+
+        ts.assertValuesOnly(6);
+
+        source.onNext(4);
+        source.onComplete();
+
+        ts.assertValuesOnly(6);
+
+        ts.request(1);
+        ts.assertResult(6, 4);
+    }
+
     @Test
     public void synchronousDrop() {
         PublishProcessor<Integer> source = PublishProcessor.create();