Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add cache to DefaultReactivePulsarSenderFactory (#169)

Browse Source
This commit is contained in:
Christophe Bornet
2022-10-21 01:59:31 +02:00
committed by Chris Bono
parent c9a459482b
commit 0447b19aa5
3 changed files with 61 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java
View File

@@ -24,6 +24,7 @@ import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
@@ -44,15 +45,21 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
private final ReactiveMessageSenderSpec reactiveMessageSenderSpec;
private final ReactiveMessageSenderCache reactiveMessageSenderCache;
public DefaultReactivePulsarSenderFactory(PulsarClient pulsarClient,
ReactiveMessageSenderSpec reactiveMessageSenderSpec) {
this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec);
ReactiveMessageSenderSpec reactiveMessageSenderSpec,
ReactiveMessageSenderCache reactiveMessageSenderCache) {
this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec,
reactiveMessageSenderCache);
}
public DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient,
ReactiveMessageSenderSpec reactiveMessageSenderSpec) {
ReactiveMessageSenderSpec reactiveMessageSenderSpec,
ReactiveMessageSenderCache reactiveMessageSenderCache) {
this.reactivePulsarClient = reactivePulsarClient;
this.reactiveMessageSenderSpec = reactiveMessageSenderSpec;
this.reactiveMessageSenderCache = reactiveMessageSenderCache;
}
@Override
@@ -81,6 +88,9 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
sender.applySpec(this.reactiveMessageSenderSpec);
}
sender.topic(resolvedTopic);
if (this.reactiveMessageSenderCache != null) {
sender.cache(this.reactiveMessageSenderCache);
}
if (messageRouter != null) {
sender.messageRouter(messageRouter);
}

130
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java
View File

@@ -16,143 +16,105 @@
package org.springframework.pulsar.core.reactive;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.List;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.apache.pulsar.reactive.client.api.MessageSpec;
import org.apache.pulsar.reactive.client.adapter.AdaptedReactivePulsarClientFactory;
import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSender;
import org.junit.jupiter.api.BeforeEach;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderCache;
import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec;
import org.assertj.core.api.InstanceOfAssertFactories;
import org.assertj.core.api.ObjectAssert;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Mono;
/**
* Common tests for {@link DefaultReactivePulsarSenderFactory}
*
* @author Christophe Bornet
*/
@SuppressWarnings("unchecked")
class DefaultReactiveMessageSenderFactoryTests {
protected final Schema<String> schema = Schema.STRING;
private ProducerBuilder<String> producerBuilder;
private PulsarClient pulsarClient;
@BeforeEach
void createPulsarClient() {
pulsarClient = mock(PulsarClient.class);
producerBuilder = mock(ProducerBuilder.class);
Producer<String> producer = mock(Producer.class);
TypedMessageBuilder<String> mockMessage = mock(TypedMessageBuilder.class);
when(mockMessage.sendAsync()).thenReturn(CompletableFuture.completedFuture(MessageId.latest));
when(producer.newMessage()).thenReturn(mockMessage);
when(producer.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));
when(producerBuilder.createAsync()).thenReturn(CompletableFuture.completedFuture(producer));
when(pulsarClient.newProducer(schema)).thenReturn(producerBuilder);
@Test
void createSenderWithSpecificTopic() {
testCreateSender(null, null, "topic1", null, null, "topic1", null);
}
@Test
void createProducerWithSpecificTopic() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
null);
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic1", schema);
sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
assertSenderHasTopicAndRouter("topic1", null);
}
@Test
void createProducerWithSpecificTopicAndMessageRouter() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
null);
void createSenderWithSpecificTopicAndMessageRouter() {
MessageRouter router = mock(MessageRouter.class);
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic1", schema, router);
sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
assertSenderHasTopicAndRouter("topic1", router);
testCreateSender(null, null, "topic1", router, null, "topic1", router);
}
@Test
void createProducerWithDefaultTopic() {
void createSenderWithDefaultTopic() {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
senderSpec.setTopicName("topic0");
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
senderSpec);
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender(null, schema);
sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
assertSenderHasTopicAndRouter("topic0", null);
testCreateSender(senderSpec, null, null, null, null, "topic0", null);
}
@Test
void createProducerWithDefaultTopicAndMessageRouter() {
void createSenderWithDefaultTopicAndMessageRouter() {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
senderSpec.setTopicName("topic0");
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
senderSpec);
MessageRouter router = mock(MessageRouter.class);
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender(null, schema, router);
sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
assertSenderHasTopicAndRouter("topic0", router);
testCreateSender(senderSpec, null, null, router, null, "topic0", router);
}
@Test
void createProducerWithSingleProducerCustomizer() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
null);
ReactiveMessageSenderBuilderCustomizer<String> customizer = builder -> builder.topic("topic1");
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic0", schema, null,
Collections.singletonList(customizer));
sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
assertSenderHasTopicAndRouter("topic1", null);
void createSenderWithSingleSenderCustomizer() {
testCreateSender(null, null, "topic1", null, Collections.singletonList(builder -> builder.topic("topic1")),
"topic1", null);
}
@Test
void createProducerWithMultipleProducerCustomizer() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
null);
void createSenderWithMultipleSenderCustomizer() {
ReactiveMessageSenderBuilderCustomizer<String> customizer1 = builder -> builder.topic("topic1");
MessageRouter router = mock(MessageRouter.class);
ReactiveMessageSenderBuilderCustomizer<String> customizer2 = builder -> builder.messageRouter(router);
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender("topic0", schema, null,
Arrays.asList(customizer1, customizer2));
sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5));
assertSenderHasTopicAndRouter("topic1", router);
testCreateSender(null, null, "topic0", null, Arrays.asList(customizer1, customizer2), "topic1", router);
}
@Test
void createProducerWithNoTopic() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient,
null);
void createSenderWithNoTopic() {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(
(PulsarClient) null, null, null);
assertThatIllegalArgumentException().isThrownBy(() -> senderFactory.createReactiveMessageSender(null, schema))
.withMessageContaining("Topic must be specified when no default topic is configured");
}
protected void assertSenderHasTopicAndRouter(String topic, MessageRouter router) {
verify(producerBuilder).topic(topic);
if (router != null) {
verify(producerBuilder).messageRouter(router);
}
else {
verify(producerBuilder, never()).messageRouter(any());
}
@Test
void createSenderWithCache() {
testCreateSender(null, AdaptedReactivePulsarClientFactory.createCache(), "topic1", null, null, "topic1", null);
}
private void testCreateSender(ReactiveMessageSenderSpec spec, ReactiveMessageSenderCache cache, String topic,
MessageRouter router, List<ReactiveMessageSenderBuilderCustomizer<String>> customizers,
String expectedTopic, MessageRouter expectedRouter) {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(
(PulsarClient) null, spec, cache);
ReactiveMessageSender<String> sender = senderFactory.createReactiveMessageSender(topic, schema, router,
customizers);
ObjectAssert<ReactiveMessageSenderSpec> objectAssert = assertThat(sender).extracting("senderSpec")
.asInstanceOf(InstanceOfAssertFactories.type(ReactiveMessageSenderSpec.class));
objectAssert.extracting(ReactiveMessageSenderSpec::getTopicName).isEqualTo(expectedTopic);
objectAssert.extracting(ReactiveMessageSenderSpec::getMessageRouter).isSameAs(expectedRouter);
assertThat(sender).extracting("producerCache").isSameAs(cache);
}
}

4
spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java
View File

@@ -66,7 +66,7 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
senderSpec.setTopicName(topic);
ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client,
senderSpec);
senderSpec, null);
ReactivePulsarSenderTemplate<Foo> pulsarTemplate = new ReactivePulsarSenderTemplate<>(producerFactory);
pulsarTemplate.setSchema(Schema.JSON(Foo.class));
Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
@@ -113,7 +113,7 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
senderSpec.setTopicName(topic);
}
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
senderSpec);
senderSpec, null);
ReactivePulsarSenderTemplate<String> pulsarTemplate = new ReactivePulsarSenderTemplate<>(senderFactory);
Mono<MessageId> sendResponse;
if (testArgs.useSimpleApi) {