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/07 15:55:09 UTC
[pulsar-client-reactive] branch main updated: Test sender maxInflight with multiple values (#93)
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 3725e23 Test sender maxInflight with multiple values (#93)
3725e23 is described below
commit 3725e23f2e4598e99284ff827bf5bfea1973e40f
Author: Christophe Bornet <cb...@hotmail.com>
AuthorDate: Wed Dec 7 16:55:04 2022 +0100
Test sender maxInflight with multiple values (#93)
---
.../adapter/AdaptedReactiveMessageSenderTest.java | 35 +++++++++++++---------
1 file changed, 21 insertions(+), 14 deletions(-)
diff --git a/pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageSenderTest.java b/pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageSenderTest.java
index 43f73dd..163d166 100644
--- a/pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageSenderTest.java
+++ b/pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageSenderTest.java
@@ -58,6 +58,7 @@ import org.apache.pulsar.reactive.client.internal.api.InternalMessageSpec;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.InOrder;
import org.mockito.Mockito;
@@ -328,20 +329,25 @@ class AdaptedReactiveMessageSenderTest {
assertThat(reconnectTimeout).isBetween(Duration.ofSeconds(4), Duration.ofSeconds(5));
}
- @Test
- void maxInFlightUsingSendOne() throws Exception {
- doTestMaxInFlight((reactiveSender, inputFlux) -> inputFlux
- .flatMap((i) -> reactiveSender.sendOne(MessageSpec.of(String.valueOf(i))), 100));
+ @ParameterizedTest
+ @CsvSource({ "7,100", "13,100", "37,500", "51,1000" })
+ void maxInFlightUsingSendOne(int maxInflight, int maxElements) throws Exception {
+ doTestMaxInFlight(
+ (reactiveSender, inputFlux) -> inputFlux
+ .flatMap((i) -> reactiveSender.sendOne(MessageSpec.of(String.valueOf(i))), 100),
+ maxInflight, maxElements);
}
- @Test
- void maxInFlightUsingSendMany() throws Exception {
+ @ParameterizedTest
+ @CsvSource({ "7,100", "13,100", "37,500", "51,1000" })
+ void maxInFlightUsingSendMany(int maxInflight, int maxElements) throws Exception {
doTestMaxInFlight((reactiveSender, inputFlux) -> inputFlux.window(3).flatMap(
- (subFlux) -> subFlux.map((i) -> MessageSpec.of(String.valueOf(i))).as(reactiveSender::sendMany), 100));
+ (subFlux) -> subFlux.map((i) -> MessageSpec.of(String.valueOf(i))).as(reactiveSender::sendMany), 100),
+ maxInflight, maxElements);
}
- void doTestMaxInFlight(BiFunction<ReactiveMessageSender<String>, Flux<Integer>, Flux<MessageId>> sendingFunction)
- throws Exception {
+ void doTestMaxInFlight(BiFunction<ReactiveMessageSender<String>, Flux<Integer>, Flux<MessageId>> sendingFunction,
+ int maxInflight, int maxElements) throws Exception {
ScheduledExecutorService executorService = null;
try {
executorService = Executors.newSingleThreadScheduledExecutor();
@@ -368,7 +374,7 @@ class AdaptedReactiveMessageSenderTest {
int encodedEntryId = Integer.parseInt(typedMessageBuilder.getMessage().getValue());
messageSender.complete(
DefaultImplementation.getDefaultImplementation().newMessageId(1, encodedEntryId, 1));
- }, 5, TimeUnit.MILLISECONDS);
+ }, 100, TimeUnit.MILLISECONDS);
return messageSender;
});
return typedMessageBuilder;
@@ -378,10 +384,11 @@ class AdaptedReactiveMessageSenderTest {
.willReturn(CompletableFuture.completedFuture(producer));
ReactiveMessageSender<String> reactiveSender = AdaptedReactivePulsarClientFactory.create(pulsarClient)
- .messageSender(Schema.STRING).maxInflight(7).cache(AdaptedReactivePulsarClientFactory.createCache())
- .maxConcurrentSenderSubscriptions(1024).topic("my-topic").build();
+ .messageSender(Schema.STRING).maxInflight(maxInflight)
+ .cache(AdaptedReactivePulsarClientFactory.createCache()).maxConcurrentSenderSubscriptions(1024)
+ .topic("my-topic").build();
- List<Integer> inputValues = IntStream.rangeClosed(1, 1000).boxed().collect(Collectors.toList());
+ List<Integer> inputValues = IntStream.rangeClosed(1, maxElements).boxed().collect(Collectors.toList());
Flux<Integer> inputFlux = Flux.fromIterable(inputValues);
Flux<MessageId> outputFlux = sendingFunction.apply(reactiveSender, inputFlux);
@@ -390,7 +397,7 @@ class AdaptedReactiveMessageSenderTest {
List<Integer> outputValues = outputFlux.map((m) -> (int) ((MessageIdImpl) m).getEntryId()).collectList()
.block();
assertThat(outputValues).containsExactlyInAnyOrderElementsOf(inputValues);
- assertThat(requestsMax.get()).isEqualTo(7);
+ assertThat(requestsMax.get()).isEqualTo(maxInflight);
}
finally {
if (executorService != null) {