-
Notifications
You must be signed in to change notification settings - Fork 1.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
`limitRate` is good at splitting large batches (like flatMap prefetches) into smaller batches of request. However, it still is ultimately pass-through in the total amount requested, and that can be wasteful for sources of data that is costly to create/emit. This commit introduces `limitRequest`, a slightly different backpressure control operator that caps the total requested amount to a hard ceiling N. If smaller requests come in from downstream, they will be passed as is until the total requested amount reaches N, at which point no more request will be sent to the upstream (which will also be cancelled). In addition, a variant of `limitRate` is also introduced, which takes a lowTide parameter setting the amount requested by replenishing prefetch optimization. These two operators work well in tandem when one wants to ensure no more data than absolutely necessary is generated in the source, but that prefetching requests are still made (to avoid waiting for the emitted data to be fully processed before requesting more). `limitRequest` also serves as a `take(n)` alternative that is stricter with request handling rather than relying on a fast cancel.
- Loading branch information
1 parent
d9cbcac
commit 98be36f
Showing
9 changed files
with
642 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
148 changes: 148 additions & 0 deletions
148
reactor-core/src/main/java/reactor/core/publisher/FluxLimitRequest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| /* | ||
| * Copyright (c) 2011-2017 Pivotal Software Inc, All Rights Reserved. | ||
| * | ||
| * 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 reactor.core.publisher; | ||
|
|
||
| import java.util.concurrent.atomic.AtomicLongFieldUpdater; | ||
|
|
||
| import org.reactivestreams.Subscription; | ||
| import reactor.core.CoreSubscriber; | ||
|
|
||
| /** | ||
| * @author Simon Baslé | ||
| * @author David Karnok | ||
| */ | ||
| final class FluxLimitRequest<T> extends FluxOperator<T, T> { | ||
|
|
||
| final long cap; | ||
|
|
||
| FluxLimitRequest(Flux<T> flux, long cap) { | ||
| super(flux); | ||
| this.cap = cap; | ||
| } | ||
|
|
||
| @Override | ||
| public void subscribe(CoreSubscriber<? super T> actual) { | ||
| source.subscribe(new FluxLimitRequestSubscriber<>(actual, this.cap)); | ||
| } | ||
|
|
||
| @Override | ||
| public int getPrefetch() { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public Object scanUnsafe(Attr key) { | ||
| if (key == Attr.REQUESTED_FROM_DOWNSTREAM) return cap; | ||
|
|
||
| //FluxOperator defines PREFETCH and PARENT | ||
| return super.scanUnsafe(key); | ||
| } | ||
|
|
||
| static class FluxLimitRequestSubscriber<T> implements InnerOperator<T, T> { | ||
|
|
||
| final CoreSubscriber<? super T> actual; | ||
|
|
||
| Subscription parent; | ||
| long toProduce; | ||
|
|
||
| volatile long requestRemaining; | ||
| static final AtomicLongFieldUpdater<FluxLimitRequestSubscriber> REQUEST_REMAINING = | ||
| AtomicLongFieldUpdater.newUpdater(FluxLimitRequestSubscriber.class, "requestRemaining"); | ||
|
|
||
|
|
||
| FluxLimitRequestSubscriber(CoreSubscriber<? super T> actual, long cap) { | ||
| this.actual = actual; | ||
| this.toProduce = cap; | ||
| this.requestRemaining = cap; | ||
| } | ||
|
|
||
| @Override | ||
| public CoreSubscriber<? super T> actual() { | ||
| return this.actual; | ||
| } | ||
|
|
||
| @Override | ||
| public void onNext(T t) { | ||
| long r = toProduce; | ||
| if (r > 0L) { | ||
| toProduce = --r; | ||
| actual.onNext(t); | ||
|
|
||
| if (r == 0) { | ||
| parent.cancel(); | ||
| actual.onComplete(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onError(Throwable throwable) { | ||
| if (toProduce != 0L) { | ||
| toProduce = 0L; | ||
| actual.onError(throwable); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onComplete() { | ||
| if (toProduce != 0L) { | ||
| toProduce = 0L; | ||
| actual.onComplete(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void onSubscribe(Subscription s) { | ||
| parent = s; | ||
| actual.onSubscribe(this); | ||
| } | ||
|
|
||
| @Override | ||
| public void request(long l) { | ||
| for (;;) { | ||
| long r = requestRemaining; | ||
| long newRequest; | ||
| if (r <= l) { | ||
| newRequest = r; | ||
| } else { | ||
| newRequest = l; | ||
| } | ||
| long u = r - newRequest; | ||
| if (REQUEST_REMAINING.compareAndSet(this, r, u)) { | ||
| if (newRequest != 0) { | ||
| parent.request(newRequest); | ||
| } | ||
| break; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void cancel() { | ||
| parent.cancel(); | ||
| } | ||
|
|
||
| @Override | ||
| public Object scanUnsafe(Attr key) { | ||
| if (key == Attr.PARENT) return parent; | ||
| if (key == Attr.TERMINATED) return toProduce == 0L; | ||
|
|
||
| //InnerOperator defines ACTUAL | ||
| return InnerOperator.super.scanUnsafe(key); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.