diff --git a/build.gradle b/build.gradle index 1cebea66..e21dc0cc 100644 --- a/build.gradle +++ b/build.gradle @@ -147,6 +147,7 @@ subprojects { subproject -> dependencies { implementation "com.google.code.findbugs:jsr305:$googleJsr305Version" testImplementation 'org.junit.jupiter:junit-jupiter-api' + testImplementation 'org.junit.jupiter:junit-jupiter-params' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' testRuntimeOnly 'org.junit.platform:junit-platform-launcher' diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java index 6b58fa38..d4425371 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java @@ -38,7 +38,7 @@ public interface PulsarOperations { * @throws PulsarClientException if an error occurs */ default MessageId send(T message) throws PulsarClientException { - return send(null, message); + return send(null, message, null); } /** @@ -48,7 +48,30 @@ public interface PulsarOperations { * @return the id of the sent message * @throws PulsarClientException if an error occurs */ - MessageId send(String topic, T message) throws PulsarClientException; + default MessageId send(String topic, T message) throws PulsarClientException { + return send(topic, message, null); + } + + /** + * Sends a message to the default topic in a blocking manner. + * @param message the message to send + * @param messageRouter the optional message router to use + * @return the id of the sent message + * @throws PulsarClientException if an error occurs + */ + default MessageId send(T message, MessageRouter messageRouter) throws PulsarClientException { + return send(null, message, messageRouter); + } + + /** + * Sends a message to the specified topic in a blocking manner. + * @param topic the topic to send the message to or {@code null} to send to the default topic + * @param message the message to send + * @param messageRouter the optional message router to use + * @return the id of the sent message + * @throws PulsarClientException if an error occurs + */ + MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException; /** * Sends a message to the default topic in a blocking manner. @@ -57,7 +80,7 @@ public interface PulsarOperations { * @throws PulsarClientException if an error occurs */ default CompletableFuture sendAsync(T message) throws PulsarClientException { - return sendAsync(null, message); + return sendAsync(null, message, null); } /** diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index 231220f7..139aa837 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -50,9 +50,9 @@ public class PulsarTemplate implements PulsarOperations { } @Override - public MessageId send(String topic, T message) throws PulsarClientException { + public MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException { try { - return this.sendAsync(topic, message).get(); + return this.sendAsync(topic, message, messageRouter).get(); } catch (Exception ex) { throw PulsarClientException.unwrap(ex); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java index 847bd752..07b29268 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java @@ -30,6 +30,7 @@ import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -45,12 +46,19 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { protected PulsarClient pulsarClient; @BeforeEach - void setup() throws PulsarClientException { + void createPulsarClient() throws PulsarClientException { pulsarClient = PulsarClient.builder() .serviceUrl(getPulsarBrokerUrl()) .build(); } + @AfterEach + void closePulsarClient() throws PulsarClientException { + if (pulsarClient != null && !pulsarClient.isClosed()) { + pulsarClient.close(); + } + } + @Test void createProducerWithSpecificTopic() throws PulsarClientException { PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index d3ab6d6e..3b103ee7 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -17,110 +17,111 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.params.provider.Arguments.arguments; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; -import java.util.HashMap; +import java.time.Duration; +import java.util.Collections; import java.util.Map; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.stream.Stream; +import org.apache.pulsar.client.admin.PulsarAdmin; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; -import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.MessageRouter; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; -import org.junit.jupiter.api.Test; +import org.apache.pulsar.client.api.TopicMetadata; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Named; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; /** + * Tests for {@code PulsarTemplate}. + * * @author Soby Chacko + * @author Chris Bono */ class PulsarTemplateTests extends AbstractContainerBaseTests { - public static final String TEST_TOPIC = "test_topic"; - - @Test - void testUsage() throws Exception { - testPulsarFunctionality(getPulsarBrokerUrl()); - } - - @Test - void testSendAsync() throws Exception { - Map config = new HashMap<>(); - config.put("topicName", "foo-bar-123"); - Map clientConfig = new HashMap<>(); - clientConfig.put("serviceUrl", getPulsarBrokerUrl()); - try ( - PulsarClient client = PulsarClient.builder() - .loadConf(clientConfig) - .build(); - Consumer consumer = client.newConsumer(Schema.STRING) - .topic("foo-bar-123") - .subscriptionName("xyz-test-subs-123") - .subscribe() - ) { - final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(client, config); - final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); - final CompletableFuture future = pulsarTemplate.sendAsync("hello john doe"); - future.thenAccept(m -> { }); - try { - Thread.sleep(2000); - future.get(); + @ParameterizedTest(name = "{0}") + @MethodSource("sendMessageTestProvider") + void sendMessageTest(String topic, Map producerConfig, SendHandler handler, MessageRouter router) throws Exception { + String subscription = topic + "-sub"; + String msgPayload = topic + "-msg"; + if (router != null) { + try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build()) { + admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1); } - catch (InterruptedException | ExecutionException e) { - e.printStackTrace(); + } + try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) { + try (Consumer consumer = client.newConsumer(Schema.STRING).topic(topic).subscriptionName(subscription).subscribe()) { + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, producerConfig); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + + Object sendResponse = handler.doSend(pulsarTemplate, topic, msgPayload, router); + if (sendResponse instanceof CompletableFuture) { + sendResponse = ((CompletableFuture) sendResponse).get(3, TimeUnit.SECONDS); + } + assertThat(sendResponse).isNotNull(); + + CompletableFuture> receiveMsgFuture = consumer.receiveAsync(); + Message msg = receiveMsgFuture.get(3, TimeUnit.SECONDS); + assertThat(msg.getData()).asString().isEqualTo(msgPayload); + + // Make sure the producer was closed by the template (albeit indirectly as client removes closed producers) + await().atMost(Duration.ofSeconds(3)).untilAsserted(() -> + assertThat(client).extracting("producers").asInstanceOf(InstanceOfAssertFactories.COLLECTION).isEmpty()); } - CompletableFuture> future0 = consumer.receiveAsync(); - Message message = future0.get(5, TimeUnit.SECONDS); - assertThat(new String(message.getData())) - .isEqualTo("hello john doe"); } } - @Test - void testSendSync() throws Exception { - Map config = new HashMap<>(); - config.put("topicName", "foo-bar-123"); - Map clientConfig = new HashMap<>(); - clientConfig.put("serviceUrl", getPulsarBrokerUrl()); - try ( - PulsarClient client = PulsarClient.builder() - .loadConf(clientConfig) - .build(); - Consumer consumer = client.newConsumer(Schema.STRING) - .topic("foo-bar-123") - .subscriptionName("xyz-test-subs-123") - .subscribe(); - ) { - final DefaultPulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(client, config); - final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); - final MessageId messageId = pulsarTemplate.send("hello john doe"); - CompletableFuture> future0 = consumer.receiveAsync(); - Message message = future0.get(5, TimeUnit.SECONDS); - assertThat(new String(message.getData())) - .isEqualTo("hello john doe"); - } + static Stream sendMessageTestProvider() { + return Stream.of( + arguments(Named.of("sendMessageToDefaultTopicNoRouter", "smt-topic-1"), Collections.singletonMap("topicName", "smt-topic-1"), + (SendHandler) (template, topic, msg, router) -> template.send(msg), + null), + arguments(Named.of("sendMessageToDefaultTopicWithRouter", "smt-topic-2"), Collections.singletonMap("topicName", "smt-topic-2"), + (SendHandler) (template, topic, msg, router) -> template.send(msg, router), + mockRouter()), + arguments(Named.of("sendMessageToSpecificTopicNoRouter", "smt-topic-3"), Collections.emptyMap(), + (SendHandler) (template, topic, msg, router) -> template.send(topic, msg), + null), + arguments(Named.of("sendMessageToSpecificTopicWithRouter", "smt-topic-4"), Collections.emptyMap(), + (SendHandler) PulsarTemplate::send, + mockRouter()), + arguments(Named.of("sendAsyncMessageToDefaultTopicNoRouter", "smt-topic-5"), Collections.singletonMap("topicName", "smt-topic-5"), + (SendHandler>) (template, topic, msg, router) -> template.sendAsync(msg), + null), + arguments(Named.of("sendAsyncMessageToDefaultTopicWithRouter", "smt-topic-6"), Collections.singletonMap("topicName", "smt-topic-6"), + (SendHandler>) (template, topic, msg, router) -> template.sendAsync(msg, router), + mockRouter()), + arguments(Named.of("sendAsyncMessageToSpecificTopicNoRouter", "smt-topic-7"), Collections.emptyMap(), + (SendHandler>) (template, topic, msg, router) -> template.sendAsync(topic, msg), + null), + arguments(Named.of("sendAsyncMessageToSpecificTopicWithRouter", "smt-topic-8"), Collections.emptyMap(), + (SendHandler>) PulsarTemplate::sendAsync, + mockRouter()) + ); } - private void testPulsarFunctionality(String pulsarBrokerUrl) throws Exception { - try ( - PulsarClient client = PulsarClient.builder() - .serviceUrl(pulsarBrokerUrl) - .build(); - Consumer consumer = client.newConsumer() - .topic(TEST_TOPIC) - .subscriptionName("test-subs") - .subscribe(); - Producer producer = client.newProducer() - .topic(TEST_TOPIC) - .create() - ) { - producer.send("test containers".getBytes()); - CompletableFuture> future = consumer.receiveAsync(); - Message message = future.get(5, TimeUnit.SECONDS); + private static MessageRouter mockRouter() { + MessageRouter router = mock(MessageRouter.class); + when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0); + return router; + } - assertThat(new String(message.getData())) - .isEqualTo("test containers"); - } + @FunctionalInterface + interface SendHandler { + V doSend(PulsarTemplate template, String topic, String msg, MessageRouter router) throws PulsarClientException; } } diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index 742b02de..da1141d8 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -79,6 +79,7 @@ value="org.assertj.core.api.Assertions.*, org.awaitility.Awaitility.*, org.junit.jupiter.api.Assertions.*, + org.junit.jupiter.params.provider.Arguments.*, org.junit.Assert.*, org.junit.Assume.*, org.junit.internal.matchers.ThrowableMessageMatcher.*,