Skip to content

Commit e0122a4

Browse files
ansipansip
authored
3.x: Add onBackpressureReduce operator (#7129)
* onBackpressureReduce operator added * onBackpressureReduce operator added * onBackpressureReduce operator added * 3.x: onBackpressureReduce operator added * 3.x: onBackpressureReduce operator added * 3.x: onBackpressureReduce operator added * 3.x: onBackpressureReduce operator added * 3.x: onBackpressureReduce operator added * 3.x: onBackpressureReduce operator * 3.x: onBackpressureReduce operator * 3.x: onBackpressureReduce operator * 3.x: onBackpressureReduce operator * 3.x: add onBackpressureReduce operator * 3.x: add onBackpressureReduce operator * 3.x: add onBackpressureReduce operator
1 parent 7741c59 commit e0122a4

File tree

7 files changed

+452
-18
lines changed

7 files changed

+452
-18
lines changed

‎src/main/java/io/reactivex/rxjava3/core/Flowable.java

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12707,9 +12707,52 @@ public final Flowable<T> onBackpressureLatest() {
1270712707
@SchedulerSupport(SchedulerSupport.NONE)
1270812708
@NonNull
1270912709
public final Flowable<T> onBackpressureReduce(@NonNull BiFunction<T, T, T> reducer) {
12710+
Objects.requireNonNull(reducer, "reducer is null");
1271012711
return RxJavaPlugins.onAssembly(new FlowableOnBackpressureReduce<>(this, reducer));
1271112712
}
1271212713

12714+
/**
12715+
* Reduces upstream values into an aggregate value, provided by a supplier and combined via a reducer function,
12716+
* while the downstream is not ready to receive items, then emits this aggregate value when the downstream becomes ready.
12717+
* <p>
12718+
* <img width="640" height="315" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Flowable.onBackpressureReduce.ff.png" alt="">
12719+
* <p>
12720+
* Note that even if the downstream is ready to receive an item, the upstream item will always be aggregated into the output type,
12721+
* calling both the supplier and the reducer to produce the output value.
12722+
* <p>
12723+
* Note that if the current {@code Flowable} does support backpressure, this operator ignores that capability
12724+
* and doesn't propagate any backpressure requests from downstream.
12725+
* <p>
12726+
* Note that due to the nature of how backpressure requests are propagated through subscribeOn/observeOn,
12727+
* requesting more than 1 from downstream doesn't guarantee a continuous delivery of {@code onNext} events.
12728+
* <dl>
12729+
* <dt><b>Backpressure:</b></dt>
12730+
* <dd>The operator honors backpressure from downstream and consumes the current {@code Flowable} in an unbounded
12731+
* manner (i.e., not applying backpressure to it).</dd>
12732+
* <dt><b>Scheduler:</b></dt>
12733+
* <dd>{@code onBackpressureReduce} does not operate by default on a particular {@link Scheduler}.</dd>
12734+
* </dl>
12735+
* @param supplier the factory to call to create new item of type R to pass it as the first argument to {@code reducer}.
12736+
* It is called when previous returned value by {@code reducer} already sent to downstream or the very first update from upstream received.
12737+
* @param reducer the bi-function to call to reduce excessive updates which downstream is not ready to receive.
12738+
* The first argument of type R is the object returned by {@code supplier} or result of previous {@code reducer} invocation.
12739+
* The second argument of type T is the current update from upstream.
12740+
* @return the new {@code Flowable} instance
12741+
* @throws NullPointerException if {@code supplier} or {@code reducer} is {@code null}
12742+
* @see #onBackpressureReduce(BiFunction)
12743+
* @since 3.0.9 - experimental
12744+
*/
12745+
@Experimental
12746+
@CheckReturnValue
12747+
@BackpressureSupport(BackpressureKind.UNBOUNDED_IN)
12748+
@SchedulerSupport(SchedulerSupport.NONE)
12749+
@NonNull
12750+
public final <R> Flowable<R> onBackpressureReduce(@NonNull Supplier<R> supplier, @NonNull BiFunction<R, ? super T, R> reducer) {
12751+
Objects.requireNonNull(supplier, "supplier is null");
12752+
Objects.requireNonNull(reducer, "reducer is null");
12753+
return RxJavaPlugins.onAssembly(new FlowableOnBackpressureReduceWith<>(this, supplier, reducer));
12754+
}
12755+
1271312756
/**
1271412757
* Returns a {@code Flowable} instance that if the current {@code Flowable} emits an error, it will emit an {@code onComplete}
1271512758
* and swallow the throwable.

‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/AbstractBackpressureThrottlingSubscriber.java

Lines changed: 10 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -28,11 +28,12 @@
2828
* Abstract base class for operators that throttle excessive updates from upstream in case if
2929
* downstream {@link Subscriber} is not ready to receive updates
3030
*
31-
* @param <T> the upstream and downstream value type
31+
* @param <T> the upstream value type
32+
* @param <R> the downstream value type
3233
*/
33-
abstract class AbstractBackpressureThrottlingSubscriber<T> extends AtomicInteger implements FlowableSubscriber<T>, Subscription {
34+
abstract class AbstractBackpressureThrottlingSubscriber<T, R> extends AtomicInteger implements FlowableSubscriber<T>, Subscription {
3435

35-
final Subscriber<? super T> downstream;
36+
final Subscriber<? super R> downstream;
3637

3738
Subscription upstream;
3839

@@ -43,9 +44,9 @@ abstract class AbstractBackpressureThrottlingSubscriber<T> extends AtomicInteger
4344

4445
final AtomicLong requested = new AtomicLong();
4546

46-
final AtomicReference<T> current = new AtomicReference<>();
47+
final AtomicReference<R> current = new AtomicReference<>();
4748

48-
AbstractBackpressureThrottlingSubscriber(Subscriber<? super T> downstream) {
49+
AbstractBackpressureThrottlingSubscriber(Subscriber<? super R> downstream) {
4950
this.downstream = downstream;
5051
}
5152

@@ -98,17 +99,17 @@ void drain() {
9899
if (getAndIncrement() != 0) {
99100
return;
100101
}
101-
final Subscriber<? super T> a = downstream;
102+
final Subscriber<? super R> a = downstream;
102103
int missed = 1;
103104
final AtomicLong r = requested;
104-
final AtomicReference<T> q = current;
105+
final AtomicReference<R> q = current;
105106

106107
for (;;) {
107108
long e = 0L;
108109

109110
while (e != r.get()) {
110111
boolean d = done;
111-
T v = q.getAndSet(null);
112+
R v = q.getAndSet(null);
112113
boolean empty = v == null;
113114

114115
if (checkTerminated(d, empty, a, q)) {
@@ -139,7 +140,7 @@ void drain() {
139140
}
140141
}
141142

142-
boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, AtomicReference<T> q) {
143+
boolean checkTerminated(boolean d, boolean empty, Subscriber<?> a, AtomicReference<R> q) {
143144
if (cancelled) {
144145
q.lazySet(null);
145146
return true;

‎src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureLatest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ protected void subscribeActual(Subscriber<? super T> s) {
2727
source.subscribe(new BackpressureLatestSubscriber<>(s));
2828
}
2929

30-
static final class BackpressureLatestSubscriber<T> extends AbstractBackpressureThrottlingSubscriber<T> {
30+
static final class BackpressureLatestSubscriber<T> extends AbstractBackpressureThrottlingSubscriber<T, T> {
3131

3232
private static final long serialVersionUID = 163080509307634843L;
3333

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

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313

1414
package io.reactivex.rxjava3.internal.operators.flowable;
1515

16+
import io.reactivex.rxjava3.annotations.NonNull;
1617
import io.reactivex.rxjava3.core.Flowable;
1718
import io.reactivex.rxjava3.exceptions.Exceptions;
1819
import io.reactivex.rxjava3.functions.BiFunction;
@@ -24,23 +25,23 @@ public final class FlowableOnBackpressureReduce<T> extends AbstractFlowableWithU
2425

2526
final BiFunction<T, T, T> reducer;
2627

27-
public FlowableOnBackpressureReduce(Flowable<T> source, BiFunction<T, T, T> reducer) {
28+
public FlowableOnBackpressureReduce(@NonNull Flowable<T> source, @NonNull BiFunction<T, T, T> reducer) {
2829
super(source);
29-
this.reducer = Objects.requireNonNull(reducer, "reducer is null");
30+
this.reducer = reducer;
3031
}
3132

3233
@Override
33-
protected void subscribeActual(Subscriber<? super T> s) {
34+
protected void subscribeActual(@NonNull Subscriber<? super T> s) {
3435
source.subscribe(new BackpressureReduceSubscriber<>(s, reducer));
3536
}
3637

37-
static final class BackpressureReduceSubscriber<T> extends AbstractBackpressureThrottlingSubscriber<T> {
38+
static final class BackpressureReduceSubscriber<T> extends AbstractBackpressureThrottlingSubscriber<T, T> {
3839

3940
private static final long serialVersionUID = 821363947659780367L;
4041

4142
final BiFunction<T, T, T> reducer;
4243

43-
BackpressureReduceSubscriber(Subscriber<? super T> downstream, BiFunction<T, T, T> reducer) {
44+
BackpressureReduceSubscriber(@NonNull Subscriber<? super T> downstream, @NonNull BiFunction<T, T, T> reducer) {
4445
super(downstream);
4546
this.reducer = reducer;
4647
}

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

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
/**
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
* <p>
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
* <p>
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
* <p>
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava3.internal.operators.flowable;
15+
16+
import io.reactivex.rxjava3.annotations.NonNull;
17+
import io.reactivex.rxjava3.core.Flowable;
18+
import io.reactivex.rxjava3.exceptions.Exceptions;
19+
import io.reactivex.rxjava3.functions.BiFunction;
20+
import io.reactivex.rxjava3.functions.Supplier;
21+
import org.reactivestreams.Subscriber;
22+
23+
import java.util.Objects;
24+
25+
public final class FlowableOnBackpressureReduceWith<T, R> extends AbstractFlowableWithUpstream<T, R> {
26+
27+
final BiFunction<R, ? super T, R> reducer;
28+
final Supplier<R> supplier;
29+
30+
public FlowableOnBackpressureReduceWith(@NonNull Flowable<T> source,
31+
@NonNull Supplier<R> supplier,
32+
@NonNull BiFunction<R, ? super T, R> reducer) {
33+
super(source);
34+
this.reducer = reducer;
35+
this.supplier = supplier;
36+
}
37+
38+
@Override
39+
protected void subscribeActual(@NonNull Subscriber<? super R> s) {
40+
source.subscribe(new BackpressureReduceWithSubscriber<>(s, supplier, reducer));
41+
}
42+
43+
static final class BackpressureReduceWithSubscriber<T, R> extends AbstractBackpressureThrottlingSubscriber<T, R> {
44+
45+
private static final long serialVersionUID = 8255923705960622424L;
46+
47+
final BiFunction<R, ? super T, R> reducer;
48+
final Supplier<R> supplier;
49+
50+
BackpressureReduceWithSubscriber(@NonNull Subscriber<? super R> downstream,
51+
@NonNull Supplier<R> supplier,
52+
@NonNull BiFunction<R, ? super T, R> reducer) {
53+
super(downstream);
54+
this.reducer = reducer;
55+
this.supplier = supplier;
56+
}
57+
58+
@Override
59+
public void onNext(T t) {
60+
R v = current.get();
61+
try {
62+
if (v == null) {
63+
current.lazySet(Objects.requireNonNull(
64+
reducer.apply(Objects.requireNonNull(supplier.get(), "The supplier returned a null value"), t),
65+
"The reducer returned a null value"
66+
));
67+
} else if ((v = current.getAndSet(null)) != null) {
68+
current.lazySet(Objects.requireNonNull(reducer.apply(v, t), "The reducer returned a null value"));
69+
} else {
70+
current.lazySet(Objects.requireNonNull(
71+
reducer.apply(Objects.requireNonNull(supplier.get(), "The supplier returned a null value"), t),
72+
"The reducer returned a null value"
73+
));
74+
}
75+
} catch (Throwable ex) {
76+
Exceptions.throwIfFatal(ex);
77+
onError(ex);
78+
cancel();
79+
}
80+
drain();
81+
}
82+
}
83+
}

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

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import org.junit.Assert;
2626
import org.junit.Test;
2727

28-
import java.io.IOException;
2928
import java.util.Random;
3029
import java.util.concurrent.TimeUnit;
3130

@@ -203,13 +202,13 @@ public void exceptionFromReducer() {
203202
PublishProcessor<Integer> source = PublishProcessor.create();
204203
TestSubscriberEx<Integer> ts = new TestSubscriberEx<>(0);
205204
source.onBackpressureReduce((l, r) -> {
206-
throw new IOException("Test exception");
205+
throw new TestException("Test exception");
207206
}).subscribe(ts);
208207

209208
source.onNext(1);
210209
source.onNext(2);
211210

212-
TestHelper.assertError(ts.errors(), 0, IOException.class, "Test exception");
211+
TestHelper.assertError(ts.errors(), 0, TestException.class, "Test exception");
213212
}
214213

215214
@Test

0 commit comments

Comments
 (0)