1.x: OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST #3487

srvaroa · 2015-11-02T20:07:28Z

Introduce a new interface BackpressureOverflowStrategy that allows implementing different handlers for an overflow situation. This patch adds three implementations, reachable via OverflowStrategy:

static class OverflowStrategy {
    static final BackpressureOverflowStrategy DEFAULT = Error.INSTANCE;
    static final Error ERROR = Error.INSTANCE;
    static final DropOldest DROP_OLDEST = DropOldest.INSTANCE;
    static final DropLatest DROP_LATEST = DropLatest.INSTANCE;
}

The behavior for each is the following:

ERROR remains the default as the existing implementation.
DROP_LATEST will drop newly produced items after the buffer fills up.
DROP_OLDEST will drop the oldest elements in the buffer, making room for
newer ones.

In all cases, a drop will result in a notification to the producer by invoking the onOverflow callback.

None of the two new behaviours (DROP_*) will unsubscribe from the source nor onError.

akarnokd · 2015-11-02T20:31:31Z

src/main/java/rx/Observable.java

@@ -20,6 +20,7 @@
 import rx.functions.*;
 import rx.internal.operators.*;
 import rx.internal.producers.SingleProducer;
+import rx.internal.operators.OperatorOnBackpressureBuffer.OnOverflow;


This enumeration should be part of a more public part of the library instead of the internal part.

Makes sense.

Any preference rather than in Observable.java? This seems to be the first case of an enum that's part of the public API (other than Notification.Kind, which is in Notification itself).

could be moved under rx and I'd rename it to BackpressureOverflowStrategy

akarnokd · 2015-11-02T20:56:32Z

Otherwise, it looks okay. 👍

srvaroa · 2015-11-03T07:57:58Z

Thanks for the review @akarnokd

akarnokd · 2015-11-03T09:01:01Z

src/main/java/rx/BackpressureOverflowStrategy.java

@@ -0,0 +1,5 @@
+package rx;


Please add the copyright header.

srvaroa · 2015-11-10T19:13:27Z

Rebased - @akarnokd let me know if there are any fixes pending.

akarnokd · 2015-11-10T20:59:40Z

Excellent. 👍. I don't think there are conflicting PRs right now.

artem-zinnatullin · 2015-11-11T06:17:31Z

src/main/java/rx/Observable.java

+     */
+    @Experimental
+    public final Observable<T> onBackpressureBuffer(long capacity, Action0 onOverflow, BackpressureOverflowStrategy overflowStrategy) {
+        return lift(new OperatorOnBackpressureBuffer<T>(capacity, onOverflow, overflowStrategy));


Knowing that some RxJava users like to pass null in places where RxJava developers do not expect that I'd add null checks for onOverflow and overflowStrategy or at least mention it in javadoc.

artem-zinnatullin · 2015-11-11T06:25:50Z

Great 👍, just a few nits.

srvaroa · 2015-11-11T07:47:22Z

Thanks @artem-zinnatullin & @akarnokd. Addressed Artem's comments.

akarnokd · 2015-11-11T10:59:33Z

👍

stealthcode · 2015-11-11T19:17:43Z

@abersnaze can you take a look and verify that it solves the issue raised in #3233?

srvaroa · 2015-11-22T16:30:48Z

Rebased on 1.x

stealthcode · 2015-12-16T01:36:09Z

The only concern that @abersnaze and I have had with this is that this uses enums to determine the strategy of operator functionality. This pattern hasn't been used much in the public API as of yet. @akarnokd @zsxwing do you guys have any opinion? We could just convert the strategies to instance methods (for consistency) or is this an emerging case that should be encouraged?

akarnokd · 2015-12-16T08:03:40Z

I wouldn't turn a the whole set of onBackpressureXXX into enums of Drop|Buffer|Latest because the pattern doesn't allow parametrizing just the Buffer variant. In this case, however, having 3-9 new overload feels unnecessary expansion of the public API. I'm in for the use of enums here, but we should store them in a reasonable subpackage.

srvaroa · 2015-12-16T08:10:43Z

@akarnokd @stealthcode I don't love using enums in this way, but I agree with @akarnokd that this can lead to excessive API expansion. But happy to move to methods if you guys prefer. Ditto for moving the enums package from rx to rx.strategies or some similar alternative.

zsxwing · 2015-12-16T18:33:39Z

Agree with @akarnokd

I prefer to add a new interface BackpressureOverflowStrategy instead of enum.

stealthcode · 2015-12-16T18:49:05Z

I agree with @zsxwing on the use of an empty interface and impls to compare instanceof instead of using an enum. Eventually they might be usable to generalize the strategies.

srvaroa · 2015-12-16T19:03:53Z

Sounds good, I'll update the patch soon. Any suggestion on the package for the strategy?

srvaroa · 2016-02-01T23:13:52Z

Rebased

akarnokd · 2016-02-02T09:02:50Z

src/main/java/rx/Observable.java

+     * @since (if this graduates from Experimental/Beta to supported, replace this parenthetical with the release number)
+     */
+    @Experimental
+    public final Observable<T> onBackpressureBuffer(long capacity, Action0 onOverflow, BackpressureOverflowStrategy overflowStrategy) {


This points to an internal interface from the public API, plus the allowed values are also internal classes and the BufferSubscriber is a private class.

Is this still a problem? I'm not sure that I understand this comment.

BackpressureOverflowStrategy is still in the internals package, it should be moved up to the rx package.

Okay, BackpressureOverflowStrategy is now in rx package.

srvaroa · 2016-02-11T18:57:50Z

@stealthcode I've been busy these days, just returned to this. I had the same doubt re. BufferSubscribe reference, @akarnokd mind to clarify where it happens?

I've fixed the visibility of OverflowStrategy members (new commit on the way)

akarnokd · 2016-02-12T08:06:33Z

src/main/java/rx/internal/operators/OperatorOnBackpressureBuffer.java

@@ -71,6 +175,7 @@ public OperatorOnBackpressureBuffer(long capacity, Action0 onOverflow) {

        return parent;
    }
+
    private static final class BufferSubscriber<T> extends Subscriber<T> implements BackpressureDrainManager.BackpressureQueueCallback {


BufferSubscriber is private here but the public BackpressureOverflowStrategy exposes it, which makes it not extendable outside this package, which is not better than having a BackpressureOverflowStrategy enum in the first place. Plus, leaving BufferSubscriber private generates accessibility bridges.

Thanks. I replaced with the parent interface implemented by BufferSubscriber, which is public.

stevegury · 2016-02-15T19:10:58Z

Thanks for the addition.
Could you also update the PR description?

srvaroa · 2016-02-15T19:27:27Z

@stevegury thanks for the review. PR fixed.

akarnokd · 2016-02-15T20:18:46Z

src/main/java/rx/internal/operators/OperatorOnBackpressureBuffer.java

+     * A convenience class to provide singleton instances of available {@code BackpressureOverflowStrategy}.
+     */
+    @SuppressWarnings("unused")
+    public static class OverflowStrategy {


Once BackpressureOverflowStrategy has been moved to rx package, the values of this class can become its constants, but all of them should have the type BackpressureOverflowStrategy. The type after the assignment can stay the same as now and can point to the current internal location.

stevegury · 2016-03-14T16:56:02Z

If I'm right the only missing thing is the header in BackpressureOverflowStrategy.java.
@srvaroa could you add that?
👍

akarnokd · 2016-03-14T17:04:08Z

Wait, it has still the problem of exposing internal classes and types into the public API:

boolean triggerOn(rx.internal.util.BackpressureDrainManager.BackpressureQueueCallback buffer) throws MissingBackpressureException;

In addition, for a proper instance of BackpressureOverflowStrategy, you have to reach into

rx.internal.operators.OperatorOnBackpressureBuffer.ON_OVERFLOW_DEFAULT

for example.

srvaroa · 2016-03-14T20:24:36Z

Thanks @akarnokd @stevegury

I'm submitting a patch that eliminates the internal API leak.

While doing it I'm starting to realize that without exposing the buffer or callback the motivation for a strategy vs a plain enum practically dissapears. The point of using a strategy was to give users ability to implement their own strategies, this led me to expose the internal buffer (which is bad, obviously). However, without it, we're basically doing the same as an enum as the real overflow behaviour is not really in the Strategy. It actually feels redundant with the onOverflow callback.

To make the strategy worth the extra code we'd need to make BufferSubscriber implement some public API interface defined by BackpressureOverflow through which we allow the user to actually interact with the buffer without leaking the internal API. Something like:

BPBuffer {
  boolean dropOldest()
  ...
}

And then:

BPStrategy {
  boolean triggerOn(BPBuffer buffer)  
}

This would become similar to my current patch (https://github.com/srvaroa/RxJava/commit/b09061048f34f539bbc022e846e7361ae6e38606) and enable meaningful strategies without leaking APIs. That said, I think that would be overengineering this patch.

Options I see:

a) revert to enums (problem: we're introducing an enum-based public API).
b) leave the patch I'm submitting next (problem: we're committing to a public API based on a Strategy that doesn't really define any behaviour).
c) define BPBuffer and BPStrategy for the sake of introducing a meaningful strategy-based API. I ca call triggerOn() with a null for now (so it effectively behaves as -b-) and proceed with making BufferSubscriber extend BPBuffer in a later patch.

I'm leaning for (a), but let me know what you prefer.

akarnokd · 2016-03-16T14:31:11Z

src/main/java/rx/BackpressureOverflow.java

+    /**
+     * Drop oldest items from the buffer making room for newer ones.
+     */
+    public static class DropOldest implements BackpressureOverflow.Strategy {


Let's have these package private at least; users should use the ON_OVERFLOW_* constants.

akarnokd · 2016-03-16T14:33:56Z

The current API looks good enough to me, only some small visibility and annotation issues.

Introduce a new interface BackpressureOverflow.Strategy that allows implementing different handlers for an overflow situation. This patch adds three implementations: - ON_OVERFLOW_ERROR remains the default as the existing implementation. - ON_OVERFLOW_DROP_LATEST will drop newly produced items after the buffer fills up. - ON_OVERFLOW_DROP_OLDEST will drop the oldest elements in the buffer, making room for newer ones. The default strategy remains ON_OVERFLOW_ERROR. In all cases, a drop will result in a notification to the producer by invoking the onOverflow callback. None of the two new behaviours (ON_OVERFLOW_DROP_*) will unsubscribe from the source nor onError. Fixes: #3233

srvaroa · 2016-03-16T20:47:23Z

Thanks for the reviews, I updated the patch.

akarnokd · 2016-03-16T22:08:39Z

👍

stevegury · 2016-03-17T17:39:24Z

Thanks @srvaroa for taking the time of writing a good PR!
👍

1.x: OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST

stealthcode · 2016-03-17T18:40:17Z

👍 thanks!

akarnokd reviewed Nov 2, 2015
View reviewed changes

akarnokd added Enhancement Feature-Request labels Nov 2, 2015

akarnokd added this to the 1.0.x milestone Nov 2, 2015

akarnokd reviewed Nov 3, 2015
View reviewed changes

src/main/java/rx/BackpressureOverflowStrategy.java

@@ -0,0 +1,5 @@

package rx;

Copy link

Member

akarnokd Nov 3, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please add the copyright header.

artem-zinnatullin reviewed Nov 11, 2015
View reviewed changes

akarnokd changed the title ~~OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST~~ 1.x: OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST Nov 12, 2015

DavidMGross mentioned this pull request Dec 1, 2015

Document new OnBackpressureBuffer drop options in RxJava ReactiveX/reactivex.github.io#154

Open

akarnokd reviewed Feb 2, 2016
View reviewed changes

akarnokd reviewed Feb 12, 2016
View reviewed changes

akarnokd reviewed Feb 15, 2016
View reviewed changes

This was referenced Mar 13, 2016

Confusion when using Backpressure operators #3751

Closed

onBackpressureBuffer is not chainable with onBackpressureDrop #2768

Closed

akarnokd reviewed Mar 16, 2016
View reviewed changes

akarnokd added a commit that referenced this pull request Mar 17, 2016

Merge pull request #3487 from srvaroa/1.x

df963fa

1.x: OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST

akarnokd merged commit df963fa into ReactiveX:1.x Mar 17, 2016

akarnokd mentioned this pull request Mar 17, 2016

Release 1.1.2 preparations #3773

Closed

1.x: OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST #3487

1.x: OnBackpressureBuffer: DROP_LATEST and DROP_OLDEST #3487

Conversation

srvaroa commented Nov 2, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

akarnokd commented Nov 2, 2015

srvaroa commented Nov 3, 2015

Choose a reason for hiding this comment

srvaroa commented Nov 10, 2015

akarnokd commented Nov 10, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

artem-zinnatullin commented Nov 11, 2015

srvaroa commented Nov 11, 2015

akarnokd commented Nov 11, 2015

stealthcode commented Nov 11, 2015

srvaroa commented Nov 22, 2015

stealthcode commented Dec 16, 2015

akarnokd commented Dec 16, 2015

srvaroa commented Dec 16, 2015

zsxwing commented Dec 16, 2015

stealthcode commented Dec 16, 2015

srvaroa commented Dec 16, 2015

srvaroa commented Feb 1, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

srvaroa commented Feb 11, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stevegury commented Feb 15, 2016

srvaroa commented Feb 15, 2016

Choose a reason for hiding this comment

stevegury commented Mar 14, 2016

akarnokd commented Mar 14, 2016

srvaroa commented Mar 14, 2016

Choose a reason for hiding this comment

akarnokd commented Mar 16, 2016

srvaroa commented Mar 16, 2016

akarnokd commented Mar 16, 2016

stevegury commented Mar 17, 2016

stealthcode commented Mar 17, 2016