You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2022/12/01 07:46:45 UTC
[pulsar-client-reactive] branch main updated: Add javadoc to InflightLimiter (#55)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pulsar-client-reactive.git
The following commit(s) were added to refs/heads/main by this push:
new 574d2e0 Add javadoc to InflightLimiter (#55)
574d2e0 is described below
commit 574d2e0382c04244b3cf234eae82fa477025cadc
Author: Christophe Bornet <cb...@hotmail.com>
AuthorDate: Thu Dec 1 08:46:40 2022 +0100
Add javadoc to InflightLimiter (#55)
---
.../client/internal/api/InflightLimiter.java | 38 ++++++++++++++++++----
.../client/internal/api/PublisherTransformer.java | 16 +++++++++
2 files changed, 48 insertions(+), 6 deletions(-)
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InflightLimiter.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InflightLimiter.java
index 0f1f95c..2411ba4 100644
--- a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InflightLimiter.java
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/InflightLimiter.java
@@ -23,8 +23,6 @@ import java.util.concurrent.atomic.AtomicReference;
import org.jctools.queues.MpmcArrayQueue;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import reactor.core.CoreSubscriber;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
@@ -35,13 +33,17 @@ import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.util.context.Context;
+/**
+ * Transformer class that limits the number of reactive streams subscription requests to
+ * keep the number of pending messages under a defined limit.
+ *
+ * @author Lari Hotari
+ */
public class InflightLimiter implements PublisherTransformer {
/** Default limit for pending Reactive Stream subscriptions. */
public static final int DEFAULT_MAX_PENDING_SUBSCRIPTIONS = 1024;
- private static final Logger LOG = LoggerFactory.getLogger(InflightLimiter.class);
-
private final MpmcArrayQueue<InflightLimiterSubscriber<?>> pendingSubscriptions;
private final AtomicInteger inflight = new AtomicInteger();
@@ -54,10 +56,22 @@ public class InflightLimiter implements PublisherTransformer {
private final Scheduler.Worker triggerNextWorker;
+ /**
+ * Constructs an InflightLimiter with a maximum number of in-flight messages.
+ * @param maxInflight the maximum number of in-flight messages
+ */
public InflightLimiter(int maxInflight) {
this(maxInflight, maxInflight, Schedulers.single(), DEFAULT_MAX_PENDING_SUBSCRIPTIONS);
}
+ /**
+ * Constructs an InflightLimiter.
+ * @param maxInflight the maximum number of in-flight messages
+ * @param expectedSubscriptionsInflight the expected number of in-flight subscriptions
+ * @param triggerNextScheduler the scheduler on which it will be checked if the
+ * subscriber can request more
+ * @param maxPendingSubscriptions the maximum number of pending subscriptions
+ */
public InflightLimiter(int maxInflight, int expectedSubscriptionsInflight, Scheduler triggerNextScheduler,
int maxPendingSubscriptions) {
this.maxInflight = maxInflight;
@@ -72,13 +86,19 @@ public class InflightLimiter implements PublisherTransformer {
@Override
public <T> Publisher<T> transform(Publisher<T> publisher) {
if (publisher instanceof Mono<?>) {
- return createOperator((Mono) publisher);
+ return createOperator((Mono<T>) publisher);
}
else {
return createOperator(Flux.from(publisher));
}
}
+ /**
+ * Limits in-flight messages for a {@link Flux}.
+ * @param source the Flux to limit
+ * @param <I> the type of the FLux
+ * @return the limited Flux
+ */
public <I> Flux<I> createOperator(Flux<I> source) {
return new FluxOperator<I, I>(source) {
@Override
@@ -88,6 +108,12 @@ public class InflightLimiter implements PublisherTransformer {
};
}
+ /**
+ * Limits in-flight messages for a {@link Mono}.
+ * @param source the Mono to limit
+ * @param <I> the type of the Mono
+ * @return the limited Mono
+ */
public <I> Mono<I> createOperator(Mono<I> source) {
return new MonoOperator<I, I>(source) {
@Override
@@ -99,7 +125,7 @@ public class InflightLimiter implements PublisherTransformer {
<I> void handleSubscribe(Publisher<I> source, CoreSubscriber<? super I> actual) {
this.activeSubscriptions.incrementAndGet();
- InflightLimiterSubscriber<I> subscriber = new InflightLimiterSubscriber<I>(actual, source);
+ InflightLimiterSubscriber<I> subscriber = new InflightLimiterSubscriber<>(actual, source);
actual.onSubscribe(subscriber.getSubscription());
}
diff --git a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/PublisherTransformer.java b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/PublisherTransformer.java
index 98f4415..33e9a27 100644
--- a/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/PublisherTransformer.java
+++ b/pulsar-client-reactive-api/src/main/java/org/apache/pulsar/reactive/client/internal/api/PublisherTransformer.java
@@ -19,8 +19,18 @@ package org.apache.pulsar.reactive.client.internal.api;
import org.reactivestreams.Publisher;
import reactor.core.Disposable;
+/**
+ * Class to transform a {@link Publisher} to another {@link Publisher}.
+ *
+ * @author Lari Hotari
+ */
public interface PublisherTransformer extends Disposable {
+ /**
+ * Gets a {@link PublisherTransformer} that returns the input {@link Publisher}
+ * without modification.
+ * @return the input {@link Publisher} not modified
+ */
static PublisherTransformer identity() {
return new PublisherTransformer() {
@Override
@@ -34,6 +44,12 @@ public interface PublisherTransformer extends Disposable {
};
}
+ /**
+ * Transforms a {@link Publisher} to another {@link Publisher}.
+ * @param publisher the publisher to transform
+ * @param <T> the type of the publisher
+ * @return the transformed publisher
+ */
<T> Publisher<T> transform(Publisher<T> publisher);
}