diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index c4838b2c..5d7541ff 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -51,6 +51,21 @@ template.newMessage(msg) ---- ==== +====== Producer customization +A `ProducerBuilderCustomizer` can be specified in order to configure the underlying Pulsar producer builder that ultimately constructs the producer used to send the outgoing message. + +WARNING: Use with caution as this gives full access to the producer builder and invoking some of its method's may have unintended side effects (eg. `create`). + +For example, the following code shows how to disable batching and enable chunking: +==== +[source, java] +---- +template.newMessage(msg) + .withProducerCustomizer((pb) -> pb.enableChunking(true).enableBatching(false)) + .send(); +---- +==== + ====== Custom routing You can use custom routing when publishing records to partitioned topics. Simple specify your custom `MessageRouter` implementation on the fluent builder such as: ==== diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java index 4c27ca2f..a34427c3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java @@ -92,28 +92,30 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor } @Override - public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, - List producerInterceptors) { + protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors, + List> producerBuilderCustomizers) { final String topicName = ProducerUtils.resolveTopicName(topic, this); ProducerCacheKey producerCacheKey = new ProducerCacheKey<>(schema, topicName, messageRouter, producerInterceptors); - return this.producerCache.get(producerCacheKey, (st) -> { - try { - return this.doCreateProducer(st.topic, st.schema, st.router, producerInterceptors); - } - catch (PulsarClientException ex) { - throw new RuntimeException(ex); - } - }); + return this.producerCache.get(producerCacheKey, (st) -> createCacheableProducer(st.topic, st.schema, st.router, + st.interceptors, producerBuilderCustomizers)); } - @Override - protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter, - List producerInterceptors) throws PulsarClientException { - Producer producer = super.doCreateProducer(topic, schema, messageRouter, producerInterceptors); - return wrapProducerWithCloseCallback(producer, - (p) -> this.logger.trace(() -> String.format("Client closed producer %s but will skip actual closing", - ProducerUtils.formatProducer(producer)))); + private Producer createCacheableProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors, + List> producerBuilderCustomizers) { + try { + Producer producer = super.doCreateProducer(topic, schema, messageRouter, producerInterceptors, + producerBuilderCustomizers); + return wrapProducerWithCloseCallback(producer, + (p) -> this.logger + .trace(() -> String.format("Client closed producer %s but will skip actual closing", + ProducerUtils.formatProducer(producer)))); + } + catch (PulsarClientException ex) { + throw new RuntimeException(ex); + } } @SuppressWarnings("unchecked") diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index f645879a..417e82d7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -57,23 +57,43 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Override public Producer createProducer(String topic, Schema schema) throws PulsarClientException { - return createProducer(topic, schema, null, null); + return doCreateProducer(topic, schema, null, null, null); } @Override public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException { - return createProducer(topic, schema, messageRouter, null); + return doCreateProducer(topic, schema, messageRouter, null, null); } @Override public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, List producerInterceptors) throws PulsarClientException { - return doCreateProducer(topic, schema, messageRouter, producerInterceptors); + return doCreateProducer(topic, schema, messageRouter, producerInterceptors, null); } + @Override + public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors, + List> producerBuilderCustomizers) throws PulsarClientException { + return doCreateProducer(topic, schema, messageRouter, producerInterceptors, producerBuilderCustomizers); + } + + /** + * Create the actual producer. + * @param topic the topic the producer will send messages to or {@code null} to use + * the default topic + * @param schema the schema of the messages to be sent + * @param messageRouter the optional message router to use + * @param producerInterceptors the optional producer interceptors to use + * @param producerBuilderCustomizers the optional list of customizers to apply to the + * producer builder + * @return the created producer + * @throws PulsarClientException if any error occurs + */ protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter, - List producerInterceptors) throws PulsarClientException { + List producerInterceptors, + List> producerBuilderCustomizers) throws PulsarClientException { final String resolvedTopic = ProducerUtils.resolveTopicName(topic, this); this.logger.trace(() -> String.format("Creating producer for '%s' topic", resolvedTopic)); final ProducerBuilder producerBuilder = this.pulsarClient.newProducer(schema); @@ -87,6 +107,9 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory if (!CollectionUtils.isEmpty(producerInterceptors)) { producerBuilder.intercept(producerInterceptors.toArray(new ProducerInterceptor[0])); } + if (!CollectionUtils.isEmpty(producerBuilderCustomizers)) { + producerBuilderCustomizers.forEach((c) -> c.customize(producerBuilder)); + } return producerBuilder.create(); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerBuilderCustomizer.java new file mode 100644 index 00000000..b869fbb2 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerBuilderCustomizer.java @@ -0,0 +1,36 @@ +/* + * Copyright 2022 the original author or authors. + * + * 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 + * + * https://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 org.springframework.pulsar.core; + +import org.apache.pulsar.client.api.ProducerBuilder; + +/** + * The interface to customize a {@link ProducerBuilder}. + * + * @param The message payload type + * @author Chris Bono + */ +@FunctionalInterface +public interface ProducerBuilderCustomizer { + + /** + * Customizes a {@link ProducerBuilder}. + * @param producerBuilder the builder to customize + */ + void customize(ProducerBuilder producerBuilder); + +} 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 64c4dcba..432eeb87 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 @@ -103,6 +103,13 @@ public interface PulsarOperations { */ SendMessageBuilder withCustomRouter(MessageRouter messageRouter); + /** + * Specifies the customizer to use to further configure the producer builder. + * @param producerCustomizer the producer builder customizer + * @return the current builder with the producer builder customizer specified + */ + SendMessageBuilder withProducerCustomizer(ProducerBuilderCustomizer producerCustomizer); + /** * Send the message in a blocking manner using the configured specification. * @return the id assigned by the broker to the published message diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index ecd4b5f3..4dcd2911 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -70,6 +70,22 @@ public interface PulsarProducerFactory { Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, List producerInterceptors) throws PulsarClientException; + /** + * Create a producer. + * @param topic the topic the producer will send messages to or {@code null} to use + * the default topic + * @param schema the schema of the messages to be sent + * @param messageRouter the optional message router to use + * @param producerInterceptors the optional producer interceptors to use + * @param producerBuilderCustomizers the optional list of customizers to apply to the + * producer builder + * @return the producer + * @throws PulsarClientException if any error occurs + */ + Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors, + List> producerBuilderCustomizers) throws PulsarClientException; + /** * Return a map of configuration options to use when creating producers. * @return the map of configuration options 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 f740d635..09abc98d 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 @@ -16,6 +16,7 @@ package org.springframework.pulsar.core; +import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -68,22 +69,22 @@ public class PulsarTemplate implements PulsarOperations { @Override public MessageId send(T message) throws PulsarClientException { - return doSend(null, message, null, null); + return doSend(null, message, null, null, null); } @Override public MessageId send(String topic, T message) throws PulsarClientException { - return doSend(topic, message, null, null); + return doSend(topic, message, null, null, null); } @Override public CompletableFuture sendAsync(T message) throws PulsarClientException { - return doSendAsync(null, message, null, null); + return doSendAsync(null, message, null, null, null); } @Override public CompletableFuture sendAsync(String topic, T message) throws PulsarClientException { - return doSendAsync(topic, message, null, null); + return doSendAsync(topic, message, null, null, null); } @Override @@ -100,9 +101,9 @@ public class PulsarTemplate implements PulsarOperations { } private MessageId doSend(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, - MessageRouter messageRouter) throws PulsarClientException { + MessageRouter messageRouter, ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { try { - return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter).get(); + return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter, producerCustomizer).get(); } catch (Exception ex) { throw PulsarClientException.unwrap(ex); @@ -110,11 +111,11 @@ public class PulsarTemplate implements PulsarOperations { } private CompletableFuture doSendAsync(String topic, T message, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) - throws PulsarClientException { + TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter, + ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory); this.logger.trace(() -> String.format("Sending msg to '%s' topic", topicName)); - final Producer producer = prepareProducerForSend(topic, message, messageRouter); + final Producer producer = prepareProducerForSend(topic, message, messageRouter, producerCustomizer); TypedMessageBuilder messageBuilder = producer.newMessage().value(message); if (typedMessageBuilderCustomizer != null) { typedMessageBuilderCustomizer.customize(messageBuilder); @@ -132,10 +133,11 @@ public class PulsarTemplate implements PulsarOperations { }); } - private Producer prepareProducerForSend(String topic, T message, MessageRouter messageRouter) - throws PulsarClientException { + private Producer prepareProducerForSend(String topic, T message, MessageRouter messageRouter, + ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); - return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors); + return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors, + producerCustomizer == null ? Collections.emptyList() : Collections.singletonList(producerCustomizer)); } public static class SendMessageBuilderImpl implements SendMessageBuilder { @@ -150,6 +152,8 @@ public class PulsarTemplate implements PulsarOperations { private MessageRouter messageRouter; + private ProducerBuilderCustomizer producerCustomizer; + SendMessageBuilderImpl(PulsarTemplate template, T message) { this.template = template; this.message = message; @@ -173,14 +177,22 @@ public class PulsarTemplate implements PulsarOperations { return this; } + @Override + public SendMessageBuilder withProducerCustomizer(ProducerBuilderCustomizer producerCustomizer) { + this.producerCustomizer = producerCustomizer; + return this; + } + @Override public MessageId send() throws PulsarClientException { - return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter); + return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter, + this.producerCustomizer); } @Override public CompletableFuture sendAsync() throws PulsarClientException { - return this.template.doSendAsync(this.topic, this.message, this.messageCustomizer, this.messageRouter); + return this.template.doSendAsync(this.topic, this.message, this.messageCustomizer, this.messageRouter, + this.producerCustomizer); } } 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 9f7fe698..a92ccbdf 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 @@ -18,14 +18,18 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; 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.PulsarClientException; import org.apache.pulsar.client.api.Schema; @@ -108,6 +112,30 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { } } + @Test + @SuppressWarnings("unchecked") + void createProducerWithSingleProducerCustomizer() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + ProducerBuilderCustomizer producerCustomizer = mock(ProducerBuilderCustomizer.class); + try (Producer producer = producerFactory.createProducer("topic0", schema, null, null, + Collections.singletonList(producerCustomizer))) { + verify(producerCustomizer).customize(any(ProducerBuilder.class)); + } + } + + @Test + @SuppressWarnings("unchecked") + void createProducerWithMultipleProducerCustomizers() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); + ProducerBuilderCustomizer producerCustomizer1 = mock(ProducerBuilderCustomizer.class); + ProducerBuilderCustomizer producerCustomizer2 = mock(ProducerBuilderCustomizer.class); + try (Producer producer = producerFactory.createProducer("topic0", schema, null, null, + Arrays.asList(producerCustomizer1, producerCustomizer2))) { + verify(producerCustomizer1).customize(any(ProducerBuilder.class)); + verify(producerCustomizer2).customize(any(ProducerBuilder.class)); + } + } + @Test void createProducerWithNoTopic() { 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 8b40f1ae..d5afaf0f 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 @@ -20,6 +20,7 @@ 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.ArgumentMatchers.argThat; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -37,10 +38,8 @@ 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.MessageRouter; import org.apache.pulsar.client.api.PulsarClient; -import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TopicMetadata; import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; @@ -51,6 +50,8 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.springframework.pulsar.core.PulsarOperations.SendMessageBuilder; + /** * Tests for {@code PulsarTemplate}. * @@ -60,52 +61,6 @@ import org.junit.jupiter.params.provider.MethodSource; */ class PulsarTemplateTests extends AbstractContainerBaseTests { - private static final String SAMPLE_MESSAGE_KEY = "sample-key"; - - private static final TypedMessageBuilderCustomizer sampleMessageKeyCustomizer = messageBuilder -> messageBuilder - .key(SAMPLE_MESSAGE_KEY); - - @ParameterizedTest(name = "{0}") - @MethodSource("sendMessageTestProvider") - void sendMessageTest(String topic, Map producerConfig, SendHandler handler, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, 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); - } - } - 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, typedMessageBuilderCustomizer, - 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); - if (typedMessageBuilderCustomizer != null) { - assertThat(msg.getKey()).isEqualTo(SAMPLE_MESSAGE_KEY); - } - 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()); - } - } - } - @ParameterizedTest(name = "{0}") @MethodSource("interceptorInvocationTestProvider") void interceptorInvocationTest(String topic, List interceptors) throws Exception { @@ -120,6 +75,14 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { } } + private static Stream interceptorInvocationTestProvider() { + return Stream.of( + arguments(Named.of("testSingleInterceptor", "iit-topic-1"), + Collections.singletonList(mock(ProducerInterceptor.class))), + arguments(Named.of("testMultipleInterceptors", "iit-topic-2"), + List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class)))); + } + @Test void sendMessageWithSpecificSchemaTest() throws Exception { String topic = "smt-specific-schema-topic"; @@ -138,123 +101,192 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { } } + @ParameterizedTest(name = "{0}") + @MethodSource("sendMessageTestProvider") + void sendMessageTest(String testName, SendTestArgs testArgs) throws Exception { + // Use the test args to construct the params to pass to send handler + String topic = testName; + String subscription = topic + "-sub"; + String msgPayload = topic + "-msg"; + MessageRouter router = null; + if (testArgs.useCustomRouter) { + router = mock(MessageRouter.class); + when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0); + } + TypedMessageBuilderCustomizer messageCustomizer = null; + if (testArgs.useMessageCustomizer) { + messageCustomizer = (mb) -> mb.key("foo-key"); + } + ProducerBuilderCustomizer producerCustomizer = null; + if (testArgs.useProducerCustomizer) { + producerCustomizer = (pb) -> pb.producerName("foo-producer"); + } + + if (router != null) { + try (PulsarAdmin admin = PulsarAdmin.builder().serviceHttpUrl(getHttpServiceUrl()).build()) { + admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1); + } + } + try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) { + try (Consumer consumer = client.newConsumer(Schema.STRING).topic(topic) + .subscriptionName(subscription).subscribe()) { + Map producerConfig = testArgs.useSpecificTopic ? Collections.emptyMap() + : Collections.singletonMap("topicName", topic); + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + producerConfig); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + Object sendResponse; + if (testArgs.useSimpleApi) { + if (testArgs.useAsyncSend) { + sendResponse = testArgs.useSpecificTopic ? pulsarTemplate.sendAsync(topic, msgPayload) + : pulsarTemplate.sendAsync(msgPayload); + } + else { + sendResponse = testArgs.useSpecificTopic ? pulsarTemplate.send(topic, msgPayload) + : pulsarTemplate.send(msgPayload); + } + } + else { + SendMessageBuilder messageBuilder = pulsarTemplate.newMessage(msgPayload); + if (testArgs.useSpecificTopic) { + messageBuilder = messageBuilder.withTopic(topic); + } + if (messageCustomizer != null) { + messageBuilder = messageBuilder.withMessageCustomizer(messageCustomizer); + } + if (router != null) { + messageBuilder = messageBuilder.withCustomRouter(router); + } + if (producerCustomizer != null) { + messageBuilder = messageBuilder.withProducerCustomizer(producerCustomizer); + } + sendResponse = testArgs.useAsyncSend ? messageBuilder.sendAsync() : messageBuilder.send(); + } + + 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); + if (messageCustomizer != null) { + assertThat(msg.getKey()).isEqualTo("foo-key"); + } + if (router != null) { + verify(router).choosePartition(argThat((Message m) -> m.getTopicName().equals(topic)), + any(TopicMetadata.class)); + } + if (producerCustomizer != null) { + assertThat(msg.getProducerName()).isEqualTo("foo-producer"); + } + // 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()); + } + } + } + private static Stream sendMessageTestProvider() { - return Stream.of( - arguments("sendMessageToDefaultTopic", - Collections.singletonMap("topicName", "sendMessageToDefaultTopic"), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .send(), - null, null), + return Stream.of(arguments("sendMessageToDefaultTopic", SendTestArgs.useSpecificTopic(false)), arguments("sendMessageToDefaultTopicWithSimpleApi", - Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithSimpleApi"), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(msg), null, - null), + SendTestArgs.useSpecificTopic(false).useSimpleApi(true)), arguments("sendMessageToDefaultTopicWithRouter", - Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithRouter"), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withCustomRouter(router).send(), - null, mockRouter()), - arguments("sendMessageToDefaultTopicWithCustomizer", - Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizer"), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withMessageCustomizer(customizer).send(), - sampleMessageKeyCustomizer, null), - arguments("sendMessageToDefaultTopicWithCustomizerAndRouter", - Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizerAndRouter"), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withMessageCustomizer(customizer).withCustomRouter(router).send(), - sampleMessageKeyCustomizer, mockRouter()), - arguments("sendMessageToSpecificTopic", Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withTopic(topic).send(), - null, null), - arguments("sendMessageToSpecificTopicWithSimpleApi", Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(topic, - msg), - null, null), - arguments("sendMessageToSpecificTopicWithRouter", Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withTopic(topic).withCustomRouter(router).send(), - null, mockRouter()), - arguments("sendMessageToSpecificTopicWithCustomizer", Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withMessageCustomizer(customizer).withTopic(topic).send(), - sampleMessageKeyCustomizer, null), - arguments("sendMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) - .withMessageCustomizer(customizer).withTopic(topic).withCustomRouter(router).send(), - sampleMessageKeyCustomizer, mockRouter()), - arguments("sendAsyncMessageToDefaultTopic", - Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopic"), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).sendAsync(), - null, null), + SendTestArgs.useSpecificTopic(false).useCustomRouter(true)), + arguments("sendMessageToDefaultTopicWithMessageCustomizer", + SendTestArgs.useSpecificTopic(false).useMessageCustomizer(true)), + arguments("sendMessageToDefaultTopicWithProducerCustomizer", + SendTestArgs.useSpecificTopic(false).useProducerCustomizer(true)), + arguments("sendMessageToDefaultTopicWithAllOptions", + SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useMessageCustomizer(true) + .useProducerCustomizer(true)), + arguments("sendMessageToSpecificTopic", SendTestArgs.useSpecificTopic(true)), + arguments("sendMessageToSpecificTopicWithSimpleApi", + SendTestArgs.useSpecificTopic(true).useSimpleApi(true)), + arguments("sendMessageToSpecificTopicWithRouter", + SendTestArgs.useSpecificTopic(true).useCustomRouter(true)), + arguments("sendMessageToSpecificTopicWithMessageCustomizer", + SendTestArgs.useSpecificTopic(true).useMessageCustomizer(true)), + arguments("sendMessageToSpecificTopicWithProducerCustomizer", + SendTestArgs.useSpecificTopic(true).useProducerCustomizer(true)), + arguments("sendMessageToSpecificTopicWithAllOptions", + SendTestArgs.useSpecificTopic(true).useCustomRouter(true).useMessageCustomizer(true) + .useProducerCustomizer(true)), + arguments("sendAsyncMessageToDefaultTopic", SendTestArgs.useSpecificTopic(false).useAsyncSend(true)), arguments("sendAsyncMessageToDefaultTopicWithSimpleApi", - Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithSimpleApi"), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(msg), - null, null), + SendTestArgs.useSpecificTopic(false).useAsyncSend(true).useSimpleApi(true)), arguments("sendAsyncMessageToDefaultTopicWithRouter", - Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithRouter"), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withCustomRouter(router).sendAsync(), - null, mockRouter()), - arguments("sendAsyncMessageToDefaultTopicWithCustomizer", - Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizer"), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withMessageCustomizer(customizer).sendAsync(), - sampleMessageKeyCustomizer, null), - arguments("sendAsyncMessageToDefaultTopicWithCustomizerAndRouter", - Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizerAndRouter"), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withMessageCustomizer(customizer) - .withCustomRouter(router).sendAsync(), - sampleMessageKeyCustomizer, mockRouter()), - arguments("sendAsyncMessageToSpecificTopic", Collections.emptyMap(), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withTopic(topic).sendAsync(), - null, null), - arguments("sendAsyncMessageToSpecificTopicWithSimpleApi", Collections.emptyMap(), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(topic, msg), - null, null), - arguments("sendAsyncMessageToSpecificTopicWithRouter", Collections.emptyMap(), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withTopic(topic).withCustomRouter(router) - .sendAsync(), - null, mockRouter()), - arguments("sendAsyncMessageToSpecificTopicWithCustomizer", Collections.emptyMap(), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic) - .sendAsync(), - sampleMessageKeyCustomizer, null), - arguments("sendAsyncMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(), - (SendHandler>) (template, topic, msg, customizer, - router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic) - .withCustomRouter(router).sendAsync(), - sampleMessageKeyCustomizer, mockRouter())); + SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useAsyncSend(true)), + arguments("sendAsyncMessageToDefaultTopicWithMessageCustomizer", + SendTestArgs.useSpecificTopic(false).useMessageCustomizer(true).useAsyncSend(true)), + arguments("sendAsyncMessageToDefaultTopicWithProducerCustomizer", + SendTestArgs.useSpecificTopic(false).useProducerCustomizer(true).useAsyncSend(true)), + arguments("sendAsyncMessageToDefaultTopicWithAllOptions", + SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useMessageCustomizer(true) + .useProducerCustomizer(true).useAsyncSend(true)), + arguments("sendAsyncMessageToSpecificTopic", SendTestArgs.useSpecificTopic(true).useAsyncSend(true)), + arguments("sendAsyncMessageToSpecificTopicWithSimpleApi", + SendTestArgs.useSpecificTopic(true).useAsyncSend(true).useSimpleApi(true)), + arguments("sendAsyncMessageToSpecificTopicWithRouter", + SendTestArgs.useSpecificTopic(true).useCustomRouter(true).useAsyncSend(true)), + arguments("sendAsyncMessageToSpecificTopicWithMessageCustomizer", + SendTestArgs.useSpecificTopic(true).useMessageCustomizer(true).useAsyncSend(true)), + arguments("sendAsyncMessageToSpecificTopicWithProducerCustomizer", + SendTestArgs.useSpecificTopic(true).useProducerCustomizer(true).useAsyncSend(true)), + arguments("sendAsyncMessageToSpecificTopicWithAllOptions", + SendTestArgs.useSpecificTopic(true).useCustomRouter(true).useMessageCustomizer(true) + .useProducerCustomizer(true).useAsyncSend(true))); } - private static Stream interceptorInvocationTestProvider() { - return Stream.of( - arguments(Named.of("testSingleInterceptor", "iit-topic-1"), - Collections.singletonList(mock(ProducerInterceptor.class))), - arguments(Named.of("testMultipleInterceptors", "iit-topic-2"), - List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class)))); - } + static final class SendTestArgs { - private static MessageRouter mockRouter() { - MessageRouter router = mock(MessageRouter.class); - when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0); - return router; - } + private boolean useSpecificTopic; - @FunctionalInterface - interface SendHandler { + private boolean useCustomRouter; - V doSend(PulsarTemplate template, String topic, String msg, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter router) - throws PulsarClientException; + private boolean useMessageCustomizer; + + private boolean useProducerCustomizer; + + private boolean useAsyncSend; + + private boolean useSimpleApi; + + private SendTestArgs(boolean useSpecificTopic) { + this.useSpecificTopic = useSpecificTopic; + } + + static SendTestArgs useSpecificTopic(boolean useSpecificTopic) { + return new SendTestArgs(useSpecificTopic); + } + + SendTestArgs useCustomRouter(boolean useCustomRouter) { + this.useCustomRouter = useCustomRouter; + return this; + } + + SendTestArgs useMessageCustomizer(boolean useMessageCustomizer) { + this.useMessageCustomizer = useMessageCustomizer; + return this; + } + + SendTestArgs useProducerCustomizer(boolean useProducerCustomizer) { + this.useProducerCustomizer = useProducerCustomizer; + return this; + } + + SendTestArgs useAsyncSend(boolean useAsyncSend) { + this.useAsyncSend = useAsyncSend; + return this; + } + + SendTestArgs useSimpleApi(boolean useSimpleApi) { + this.useSimpleApi = useSimpleApi; + return this; + } }