Skip to content

1.x: Add an operator to throttle data via controlling the requests going upstream #3781

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

1.x: Add an operator to throttle data via controlling the requests going upstream #3781

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions src/main/java/rx/Observable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7664,6 +7664,26 @@ public final <U> Observable<T> sample(Observable<U> sampler) {
return lift(new OperatorSampleWithObservable<T, U>(sampler));
}

/**
* Allow the an external signal control the amount of data being set through this Observable chain.
* When the control Observable emits false (closes the valve) requests upstream are stopped and any
* requests from downstream for more data are buffered until the control Observable emits a true
* (opens the valve). Should the control Observable error or complete while closed (last control
* emition was a false) an error is sent down the data stream. The granularity breaks up large requests
* from downstream to limit the number of onNexts that are possible after the control valve has closed.
* The smaller the number the tighter the control on the flow but the more overhead there will be in
* managing the requests.
*
* @param control
* an Observable that dictates if request signals propagate upstream
* @param granularity
* the maximum number of outstanding requests.
* @returns an Observable that mostly stops emiting after the control Observable emits a false.
*/
public final Observable<T> pressureValve(Observable<Boolean> control, long granularity) {
return lift(new OperatorValve<T>(control, granularity));
}

/**
* Returns an Observable that applies a specified accumulator function to the first item emitted by a source
* Observable, then feeds the result of that function along with the second item emitted by the source
Expand Down
204 changes: 204 additions & 0 deletions src/main/java/rx/internal/operators/OperatorValve.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,204 @@
/**
* Copyright 2016 Netflix, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rx.internal.operators;

import java.util.concurrent.atomic.AtomicBoolean;

import rx.Observable;
import rx.Observable.Operator;
import rx.Producer;
import rx.Subscriber;

/**
* An {@code Observable} that emits the first {@code num} items emitted by the source {@code Observable}.
* <p>
* <img width="640" height="305" src="https://raw.githubusercontent.com/wiki/ReactiveX/RxJava/images/rx-operators/take.png" alt="" />
* <p>
* You can choose to pay attention only to the first {@code num} items emitted by an {@code Observable} by using
* the {@code take} operator. This operator returns an {@code Observable} that will invoke a subscriber's
* {@link Subscriber#onNext onNext} function a maximum of {@code num} times before invoking
* {@link Subscriber#onCompleted onCompleted}.
*/
public final class OperatorValve<T> implements Operator<T, T> {
private final Observable<Boolean> onByDefault;
private final long _granularity;

public OperatorValve(Observable<Boolean> onByDefault, long granularity) {
this.onByDefault = onByDefault;
this._granularity = granularity;
}

@Override
public Subscriber<? super T> call(final Subscriber<? super T> child) {
return new Subscriber<T>(child) {
private final long granularity = _granularity;
private Producer p;
private long backlog;// synchronized access on Producer p
private long outstanding;// synchronized access on Producer p
private boolean isOpen = true;// synchronized access on Producer p
private AtomicBoolean terminated = new AtomicBoolean();

@Override
public void onCompleted() {
if (terminated.compareAndSet(false, true))
child.onCompleted();
}

@Override
public void onError(Throwable e) {
if (terminated.compareAndSet(false, true))
child.onError(e);
}

@Override
public void onNext(T t) {
child.onNext(t);
final long requestUp;
synchronized (this) {
if (--outstanding == 0 && isOpen) {
// all out and still open; check to see if there is a backlog.
if (backlog > granularity) {
// don't request too much at once
requestUp = granularity;
} else if (backlog > 0) {
// the backlog isn't too big
requestUp = backlog;
} else {
// no backlog
requestUp = 0;
}
} else {
// expecting more or closed
requestUp = 0;
}
if (requestUp > 0) {
// do the last of the accounting inside the synchronized block
backlog -= requestUp;
outstanding += requestUp;
}
}
// do the request work outside the synchronized block
if (requestUp != 0)
p.request(requestUp);
}

@Override
public void setProducer(final Producer p) {
this.p = p;

onByDefault.unsafeSubscribe(new Subscriber<Boolean>() {
@Override
public void onCompleted() {
boolean _isOpen;
synchronized (this) {
// make sure to get the latest value of isOpen
_isOpen = isOpen;
}
if (!_isOpen) {
if (terminated.compareAndSet(false, true)) {
child.onError(new IllegalStateException("control signal terminated while valve was closed"));
}
}
unsubscribe();
}

@Override
public void onError(Throwable e) {
if (terminated.compareAndSet(false, true))
child.onError(e);
unsubscribe();
}

@Override
public void onNext(Boolean open) {
if (open) {
final long requestUp;
synchronized (this) {
if (!isOpen) {
// opening, check backlog.
if (backlog > granularity) {
// don't request too much at once
requestUp = granularity;
} else if (backlog > 0) {
// the backlog isn't too big
requestUp = backlog;
} else {
// no backlog
requestUp = 0;
}
isOpen = true;
} else {
// was already open
requestUp = 0;
}
if (requestUp > 0) {
// do the last of the accounting inside the synchronized block
backlog -= requestUp;
outstanding += requestUp;
}
}
// do the request work outside the synchronized block
if (requestUp > 0)
p.request(requestUp);
} else {
synchronized (this) {
// closing
isOpen = false;
}
}
}
});

super.setProducer(new Producer() {
@Override
public void request(long n) {
if (n < 0)
throw new IllegalArgumentException("n >= 0 required but it was " + n);
final long requestUp;
Copy link
Member

Choose a reason for hiding this comment

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

validate n > 0?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Will do.

synchronized (this) {
// increase backlog
backlog += n;
// now figure out if what is going to happen to it.
if (!isOpen) {
// closed; don't send
requestUp = 0;
} else {
if (backlog > granularity) {
// don't request too much at once
requestUp = granularity;
} else if (backlog > 0) {
// the backlog isn't too big
requestUp = backlog;
} else {
// no backlog
requestUp = 0;
}
}
if (requestUp > 0) {
// do the last of the accounting inside the synchronized block
backlog -= requestUp;
outstanding += requestUp;
}
}
// do the request work outside the synchronized block
if (requestUp != 0)
p.request(requestUp);
}
});
}
};
}
}
Loading