diff --git a/src/main/java/io/reactivex/rxjava3/core/Flowable.java b/src/main/java/io/reactivex/rxjava3/core/Flowable.java index 55465fa042..4d277ab08b 100644 --- a/src/main/java/io/reactivex/rxjava3/core/Flowable.java +++ b/src/main/java/io/reactivex/rxjava3/core/Flowable.java @@ -12546,7 +12546,7 @@ public final Flowable onBackpressureBuffer(int capacity, boolean delayError) @NonNull public final Flowable onBackpressureBuffer(int capacity, boolean delayError, boolean unbounded) { ObjectHelper.verifyPositive(capacity, "capacity"); - return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBuffer<>(this, capacity, unbounded, delayError, Functions.EMPTY_ACTION)); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBuffer<>(this, capacity, unbounded, delayError, Functions.EMPTY_ACTION, Functions.emptyConsumer())); } /** @@ -12577,6 +12577,7 @@ public final Flowable onBackpressureBuffer(int capacity, boolean delayError, * @throws NullPointerException if {@code onOverflow} is {@code null} * @throws IllegalArgumentException if {@code capacity} is non-positive * @see ReactiveX operators documentation: backpressure operators + * @see #onBackpressureBuffer(int, boolean, boolean, Action, Consumer) * @since 1.1.0 */ @CheckReturnValue @@ -12587,7 +12588,51 @@ public final Flowable onBackpressureBuffer(int capacity, boolean delayError, @NonNull Action onOverflow) { Objects.requireNonNull(onOverflow, "onOverflow is null"); ObjectHelper.verifyPositive(capacity, "capacity"); - return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBuffer<>(this, capacity, unbounded, delayError, onOverflow)); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBuffer<>(this, capacity, unbounded, delayError, onOverflow, Functions.emptyConsumer())); + } + + /** + * Buffers an optionally unlimited number of items from the current {@code Flowable} and allows it to emit as fast it can while allowing the + * downstream to consume the items at its own place. + * If {@code unbounded} is {@code true}, the resulting {@code Flowable} will signal a + * {@link MissingBackpressureException} via {@code onError} as soon as the buffer's capacity is exceeded, dropping all undelivered + * items, canceling the flow and calling the {@code onOverflow} action. + *

+ * + *

+ *
Backpressure:
+ *
The operator honors backpressure from downstream and consumes the current {@code Flowable} in an unbounded + * manner (i.e., not applying backpressure to it).
+ *
Scheduler:
+ *
{@code onBackpressureBuffer} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param capacity number of slots available in the buffer. + * @param delayError + * if {@code true}, an exception from the current {@code Flowable} is delayed until all buffered elements have been + * consumed by the downstream; if {@code false}, an exception is immediately signaled to the downstream, skipping + * any buffered element + * @param unbounded + * if {@code true}, the capacity value is interpreted as the internal "island" size of the unbounded buffer + * @param onOverflow action to execute if an item needs to be buffered, but there are no available slots. + * @param onDropped the {@link Consumer} to be called with the item that could not be buffered due to capacity constraints. + * @return the new {@code Flowable} instance + * @throws NullPointerException if {@code onOverflow} or {@code onDropped} is {@code null} + * @throws IllegalArgumentException if {@code capacity} is non-positive + * @see ReactiveX operators documentation: backpressure operators + * @since 3.1.7 + */ + @CheckReturnValue + @NonNull + @BackpressureSupport(BackpressureKind.SPECIAL) + @SchedulerSupport(SchedulerSupport.NONE) + @Experimental + public final Flowable onBackpressureBuffer(int capacity, boolean delayError, boolean unbounded, + @NonNull Action onOverflow, @NonNull Consumer onDropped) { + Objects.requireNonNull(onOverflow, "onOverflow is null"); + Objects.requireNonNull(onDropped, "onDropped is null"); + ObjectHelper.verifyPositive(capacity, "capacity"); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBuffer<>(this, capacity, unbounded, delayError, onOverflow, onDropped)); } /** @@ -12653,6 +12698,7 @@ public final Flowable onBackpressureBuffer(int capacity, @NonNull Action onOv * @throws NullPointerException if {@code onOverflow} or {@code overflowStrategy} is {@code null} * @throws IllegalArgumentException if {@code capacity} is non-positive * @see ReactiveX operators documentation: backpressure operators + * @see #onBackpressureBuffer(long, Action, BackpressureOverflowStrategy) * @since 2.0 */ @CheckReturnValue @@ -12662,9 +12708,55 @@ public final Flowable onBackpressureBuffer(int capacity, @NonNull Action onOv public final Flowable onBackpressureBuffer(long capacity, @Nullable Action onOverflow, @NonNull BackpressureOverflowStrategy overflowStrategy) { Objects.requireNonNull(overflowStrategy, "overflowStrategy is null"); ObjectHelper.verifyPositive(capacity, "capacity"); - return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBufferStrategy<>(this, capacity, onOverflow, overflowStrategy)); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBufferStrategy<>(this, capacity, onOverflow, overflowStrategy, null)); } + /** + * Buffers an optionally unlimited number of items from the current {@code Flowable} and allows it to emit as fast it can while allowing the + * downstream to consume the items at its own place. + * The resulting {@code Flowable} will behave as determined by {@code overflowStrategy} if the buffer capacity is exceeded: + *
    + *
  • {@link BackpressureOverflowStrategy#ERROR} (default) will call {@code onError} dropping all undelivered items, + * canceling the source, and notifying the producer with {@code onOverflow}.
    • + *
  • {@link BackpressureOverflowStrategy#DROP_LATEST} will drop any new items emitted by the producer while + * the buffer is full, without generating any {@code onError}. Each drop will, however, invoke {@code onOverflow} + * to signal the overflow to the producer.
    • + *
  • {@link BackpressureOverflowStrategy#DROP_OLDEST} will drop the oldest items in the buffer in order to make + * room for newly emitted ones. Overflow will not generate an {@code onError}, but each drop will invoke + * {@code onOverflow} to signal the overflow to the producer.
    • + *
+ * + *

+ * + *

+ *
Backpressure:
+ *
The operator honors backpressure from downstream and consumes the current {@code Flowable} in an unbounded + * manner (i.e., not applying backpressure to it).
+ *
Scheduler:
+ *
{@code onBackpressureBuffer} does not operate by default on a particular {@link Scheduler}.
+ *
+ * + * @param capacity number of slots available in the buffer. + * @param onOverflow action to execute if an item needs to be buffered, but there are no available slots, {@code null} is allowed. + * @param overflowStrategy how should the resulting {@code Flowable} react to buffer overflows, {@code null} is not allowed. + * @param onDropped the {@link Consumer} to be called with the item that could not be buffered due to capacity constraints. + * @return the new {@code Flowable} instance + * @throws NullPointerException if {@code onOverflow}, {@code overflowStrategy} or {@code onDropped} is {@code null} + * @throws IllegalArgumentException if {@code capacity} is non-positive + * @see ReactiveX operators documentation: backpressure operators + * @since 3.1.7 + */ + @CheckReturnValue + @NonNull + @BackpressureSupport(BackpressureKind.SPECIAL) + @SchedulerSupport(SchedulerSupport.NONE) + @Experimental + public final Flowable onBackpressureBuffer(long capacity, @Nullable Action onOverflow, @NonNull BackpressureOverflowStrategy overflowStrategy, @NonNull Consumer onDropped) { + Objects.requireNonNull(overflowStrategy, "overflowStrategy is null"); + Objects.requireNonNull(onDropped, "onDropped is null"); + ObjectHelper.verifyPositive(capacity, "capacity"); + return RxJavaPlugins.onAssembly(new FlowableOnBackpressureBufferStrategy<>(this, capacity, onOverflow, overflowStrategy, onDropped)); + } /** * Drops items from the current {@code Flowable} if the downstream is not ready to receive new items (indicated * by a lack of {@link Subscription#request(long)} calls from it). diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java index 8261df29dc..db58b68a10 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBuffer.java @@ -20,7 +20,7 @@ import io.reactivex.rxjava3.annotations.Nullable; import io.reactivex.rxjava3.core.*; import io.reactivex.rxjava3.exceptions.*; -import io.reactivex.rxjava3.functions.Action; +import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.subscriptions.*; import io.reactivex.rxjava3.internal.util.BackpressureHelper; import io.reactivex.rxjava3.operators.*; @@ -30,19 +30,21 @@ public final class FlowableOnBackpressureBuffer extends AbstractFlowableWithU final boolean unbounded; final boolean delayError; final Action onOverflow; + final Consumer onDropped; public FlowableOnBackpressureBuffer(Flowable source, int bufferSize, boolean unbounded, - boolean delayError, Action onOverflow) { + boolean delayError, Action onOverflow, Consumer onDropped) { super(source); this.bufferSize = bufferSize; this.unbounded = unbounded; this.delayError = delayError; this.onOverflow = onOverflow; + this.onDropped = onDropped; } @Override protected void subscribeActual(Subscriber s) { - source.subscribe(new BackpressureBufferSubscriber<>(s, bufferSize, unbounded, delayError, onOverflow)); + source.subscribe(new BackpressureBufferSubscriber<>(s, bufferSize, unbounded, delayError, onOverflow, onDropped)); } static final class BackpressureBufferSubscriber extends BasicIntQueueSubscription implements FlowableSubscriber { @@ -53,6 +55,7 @@ static final class BackpressureBufferSubscriber extends BasicIntQueueSubscrip final SimplePlainQueue queue; final boolean delayError; final Action onOverflow; + final Consumer onDropped; Subscription upstream; @@ -66,10 +69,11 @@ static final class BackpressureBufferSubscriber extends BasicIntQueueSubscrip boolean outputFused; BackpressureBufferSubscriber(Subscriber actual, int bufferSize, - boolean unbounded, boolean delayError, Action onOverflow) { + boolean unbounded, boolean delayError, Action onOverflow, Consumer onDropped) { this.downstream = actual; this.onOverflow = onOverflow; this.delayError = delayError; + this.onDropped = onDropped; SimplePlainQueue q; @@ -98,6 +102,7 @@ public void onNext(T t) { MissingBackpressureException ex = new MissingBackpressureException("Buffer is full"); try { onOverflow.run(); + onDropped.accept(t); } catch (Throwable e) { Exceptions.throwIfFatal(e); ex.initCause(e); diff --git a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java index 29a207fba5..7963fb7d40 100644 --- a/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java +++ b/src/main/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategy.java @@ -20,7 +20,7 @@ import io.reactivex.rxjava3.core.*; import io.reactivex.rxjava3.exceptions.*; -import io.reactivex.rxjava3.functions.Action; +import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper; import io.reactivex.rxjava3.internal.util.BackpressureHelper; import io.reactivex.rxjava3.plugins.RxJavaPlugins; @@ -38,17 +38,21 @@ public final class FlowableOnBackpressureBufferStrategy extends AbstractFlowa final BackpressureOverflowStrategy strategy; + final Consumer onDropped; + public FlowableOnBackpressureBufferStrategy(Flowable source, - long bufferSize, Action onOverflow, BackpressureOverflowStrategy strategy) { + long bufferSize, Action onOverflow, BackpressureOverflowStrategy strategy, + Consumer onDropped) { super(source); this.bufferSize = bufferSize; this.onOverflow = onOverflow; this.strategy = strategy; + this.onDropped = onDropped; } @Override protected void subscribeActual(Subscriber s) { - source.subscribe(new OnBackpressureBufferStrategySubscriber<>(s, onOverflow, strategy, bufferSize)); + source.subscribe(new OnBackpressureBufferStrategySubscriber<>(s, onOverflow, strategy, bufferSize, onDropped)); } static final class OnBackpressureBufferStrategySubscriber @@ -61,6 +65,8 @@ static final class OnBackpressureBufferStrategySubscriber final Action onOverflow; + final Consumer onDropped; + final BackpressureOverflowStrategy strategy; final long bufferSize; @@ -77,13 +83,15 @@ static final class OnBackpressureBufferStrategySubscriber Throwable error; OnBackpressureBufferStrategySubscriber(Subscriber actual, Action onOverflow, - BackpressureOverflowStrategy strategy, long bufferSize) { + BackpressureOverflowStrategy strategy, long bufferSize, + Consumer onDropped) { this.downstream = actual; this.onOverflow = onOverflow; this.strategy = strategy; this.bufferSize = bufferSize; this.requested = new AtomicLong(); this.deque = new ArrayDeque<>(); + this.onDropped = onDropped; } @Override @@ -104,44 +112,60 @@ public void onNext(T t) { } boolean callOnOverflow = false; boolean callError = false; + boolean callDrain = false; Deque dq = deque; + T toDrop = null; synchronized (dq) { if (dq.size() == bufferSize) { switch (strategy) { case DROP_LATEST: - dq.pollLast(); + toDrop = dq.pollLast(); dq.offer(t); callOnOverflow = true; break; case DROP_OLDEST: - dq.poll(); + toDrop = dq.poll(); dq.offer(t); callOnOverflow = true; break; default: // signal error + toDrop = t; callError = true; break; } } else { dq.offer(t); + callDrain = true; } } - if (callOnOverflow) { - if (onOverflow != null) { - try { - onOverflow.run(); - } catch (Throwable ex) { - Exceptions.throwIfFatal(ex); - upstream.cancel(); - onError(ex); - } + if (callOnOverflow && onOverflow != null) { + try { + onOverflow.run(); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + upstream.cancel(); + onError(ex); + } + } + + if (onDropped != null && toDrop != null) { + try { + onDropped.accept(toDrop); + } catch (Throwable ex) { + Exceptions.throwIfFatal(ex); + upstream.cancel(); + onError(ex); } - } else if (callError) { + } + + if (callError) { upstream.cancel(); onError(MissingBackpressureException.createDefault()); - } else { + } + + if (callDrain) { drain(); } } diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategyTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategyTest.java index 4fc8ba3c25..217162c206 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategyTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferStrategyTest.java @@ -15,7 +15,9 @@ import static io.reactivex.rxjava3.core.BackpressureOverflowStrategy.*; import static io.reactivex.rxjava3.internal.functions.Functions.EMPTY_ACTION; -import static org.junit.Assert.assertEquals; +import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -28,8 +30,9 @@ import io.reactivex.rxjava3.functions.*; import io.reactivex.rxjava3.internal.functions.Functions; import io.reactivex.rxjava3.internal.subscriptions.BooleanSubscription; +import io.reactivex.rxjava3.processors.PublishProcessor; import io.reactivex.rxjava3.subscribers.*; -import io.reactivex.rxjava3.testsupport.TestHelper; +import io.reactivex.rxjava3.testsupport.*; public class FlowableOnBackpressureBufferStrategyTest extends RxJavaTest { @@ -225,4 +228,98 @@ public void cancelOnDrain() { .requestMore(10) .assertResult(1); } + + @Test + public void onDroppedNormalDropOldest() throws Throwable { + PublishProcessor pp = PublishProcessor.create(); + + @SuppressWarnings("unchecked") + Consumer onDropped = mock(Consumer.class); + + TestSubscriber ts = pp.onBackpressureBuffer(1, null, BackpressureOverflowStrategy.DROP_OLDEST, onDropped) + .test(0L); + + ts.assertEmpty(); + + pp.onNext(1); + + ts.assertEmpty(); + verify(onDropped, never()).accept(any()); + + pp.onNext(2); + + ts.assertEmpty(); + + verify(onDropped).accept(1); + } + + @Test + public void onDroppedNormalDropLatest() throws Throwable { + PublishProcessor pp = PublishProcessor.create(); + + @SuppressWarnings("unchecked") + Consumer onDropped = mock(Consumer.class); + + TestSubscriber ts = pp.onBackpressureBuffer(2, null, BackpressureOverflowStrategy.DROP_LATEST, onDropped) + .test(0L); + + ts.assertEmpty(); + + pp.onNext(1); + + pp.onNext(2); + + ts.assertEmpty(); + verify(onDropped, never()).accept(any()); + + pp.onNext(3); + + ts.assertEmpty(); + + verify(onDropped).accept(2); + } + + @Test + public void onDroppedNormalError() throws Throwable { + PublishProcessor pp = PublishProcessor.create(); + + @SuppressWarnings("unchecked") + Consumer onDropped = mock(Consumer.class); + + TestSubscriber ts = pp.onBackpressureBuffer(1, null, BackpressureOverflowStrategy.ERROR, onDropped) + .test(0L); + + ts.assertEmpty(); + + pp.onNext(1); + + ts.assertEmpty(); + verify(onDropped, never()).accept(any()); + + pp.onNext(2); + + ts.assertFailure(MissingBackpressureException.class); + + verify(onDropped).accept(2); + } + + @Test + public void onDroppedCrash() throws Throwable { + PublishProcessor pp = PublishProcessor.create(); + + Consumer onDropped = v -> { throw new TestException(); }; + + TestSubscriberEx ts = pp.onBackpressureBuffer(1, null, BackpressureOverflowStrategy.DROP_OLDEST, onDropped) + .subscribeWith(new TestSubscriberEx(0L)); + + ts.assertEmpty(); + + pp.onNext(1); + + ts.assertEmpty(); + + pp.onNext(2); + + ts.assertFailure(TestException.class); + } } diff --git a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java index 5a02218423..657ce36f1b 100644 --- a/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java +++ b/src/test/java/io/reactivex/rxjava3/internal/operators/flowable/FlowableOnBackpressureBufferTest.java @@ -14,6 +14,8 @@ package io.reactivex.rxjava3.internal.operators.flowable; import static org.junit.Assert.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; import java.util.List; import java.util.concurrent.*; @@ -350,4 +352,50 @@ public void doubleOnSubscribe() { public void badRequest() { TestHelper.assertBadRequestReported(Flowable.never().onBackpressureBuffer()); } + + @Test + public void onDroppedNormal() throws Throwable { + PublishProcessor pp = PublishProcessor.create(); + + @SuppressWarnings("unchecked") + Consumer onDropped = mock(Consumer.class); + + TestSubscriber ts = pp.onBackpressureBuffer(1, false, false, () -> { }, onDropped) + .test(0L); + + ts.assertEmpty(); + + pp.onNext(1); + + ts.assertEmpty(); + verify(onDropped, never()).accept(any()); + + pp.onNext(2); + + ts.assertFailure(MissingBackpressureException.class); + + verify(onDropped).accept(2); + } + + @Test + public void onDroppedCrash() throws Throwable { + PublishProcessor pp = PublishProcessor.create(); + + Consumer onDropped = v -> { throw new TestException(); }; + + TestSubscriberEx ts = pp.onBackpressureBuffer(1, false, false, () -> { }, onDropped) + .subscribeWith(new TestSubscriberEx(0L)); + + ts.assertEmpty(); + + pp.onNext(1); + + ts.assertEmpty(); + + pp.onNext(2); + + ts.assertFailure(MissingBackpressureException.class); + + assertTrue(ts.errors().get(0).getCause() instanceof TestException); + } } diff --git a/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java b/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java index 8dd240119b..c2ca87d4d2 100644 --- a/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java +++ b/src/test/java/io/reactivex/rxjava3/validators/ParamValidationCheckerTest.java @@ -153,6 +153,7 @@ public void checkParallelFlowable() { // null Action allowed addOverride(new ParamOverride(Flowable.class, 1, ParamMode.ANY, "onBackpressureBuffer", Long.TYPE, Action.class, BackpressureOverflowStrategy.class)); + addOverride(new ParamOverride(Flowable.class, 1, ParamMode.ANY, "onBackpressureBuffer", Long.TYPE, Action.class, BackpressureOverflowStrategy.class, Consumer.class)); // zero repeat is allowed addOverride(new ParamOverride(Flowable.class, 0, ParamMode.NON_NEGATIVE, "repeat", Long.TYPE));