From c1d155e50747c6e1f21f432901528ff7d171d6d5 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Mon, 29 Aug 2022 14:42:18 -0500 Subject: [PATCH] Add fluent API to PulsarTemplate Fixes #68 --- .../src/main/asciidoc/pulsar.adoc | 43 ++-- .../main/java/app2/FailoverConsumerApp.java | 9 +- .../pulsar/core/PulsarOperations.java | 213 +++++------------- .../pulsar/core/PulsarTemplate.java | 102 +++++++-- .../pulsar/core/FailoverConsumerTests.java | 7 +- .../pulsar/core/PulsarTemplateTests.java | 108 +++++---- 6 files changed, 240 insertions(+), 242 deletions(-) diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index f8797d0e..5dfc6299 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -23,7 +23,7 @@ include::application-properties/pulsar-client.adoc[lines=3..-1] [[pulsar-producer]] ==== Pulsar Producer -On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarOperations.html[several variants of 'send' methods] to publish records through its contract. +On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides methods to publish records through its contract. There are two categories of these send API methods - `send` and `sendAsync`. The `send` methods are blocking calls using the synchronous sending capabilities on the Pulsar producer. @@ -31,20 +31,40 @@ They return the `MessageId` of the message that was published once the message i The `sendAsync` method calls are asynchronous calls that are non-blocking. They return a `CompletableFuture` using which you can asynchronously receive the message id once the messages are published. -Both `send` and `sendAsync` methods have a variant that allows to publish simply with the message. -When you do that, the application must provide the topic name using the property `spring.pulsar.producer.topicName`. -There is another variant for both flavors that takes the topic name as an argument. +===== Simple API +The template provides a handful of methods ({javadocs}/org/springframework/pulsar/core/PulsarOperations.html[prefixed with _'send'_]) for simple send requests that contain only a message and/or destination topic. For more complicated send requests there is a fluent API that allows the user to configure more options (see below). -The methods using a `TypedMessageBuilderCustomizer` allow modifying the basic `TypedMessageBuilder`. -This can be used to send a keyed message for example: +TIP: Both `send` and `sendAsync` methods have a variant that allows to publish simply with the message. +When you do that, the application must provide the topic name using the property `spring.pulsar.producer.topic-name`. +===== Fluent API +The template provides a {javadocs}/org/springframework/pulsar/core/PulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests. + +====== Message customization +A `TypedMessageBuilderCustomizer` can be specified in order to configure the outgoing message. For example, the following code shows how to send a keyed message: ==== [source, java] ---- -template.send(msg, (messageBuilder -> messageBuilder.key(myMessageKey))); +template.newMessage(msg) + .withMessageCustomizer((mb) -> mb.key("foo-msg-key")) + .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: +==== +[source, java] +---- +template.newMessage(msg) + .withCustomRouter(myCustomRouter) + .send(); +---- +==== + +TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`. + +===== Schema If you are using simple Java primitive types, then the framework auto-detects the schema for you, and you do not need to specify any schema types for publishing the data. However, if you are using any complex types such as `JSON`, `AVRO`, `PROTOBUF`, etc. then you need to set the proper schema type on the `PulsarTemplate` before invoking any send operations as shown below. @@ -55,11 +75,6 @@ pulsarTemplate.setSchema(Schema.JSON(Foo.class)); ---- ==== - -When using partitioned topics, then you can use custom partitioning routing when publishing records. -For this purpose, you can provide an implementation of a `MessageRouter` and pass it along with both `send` and `sendAsync` methods. -Note that, when using a `MessageRouter`, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`. - [[producer-application-properties]] .[.underline]#Click ##here## to view the available **Pulsar Producer Properties**# [%collapsible] @@ -366,7 +381,7 @@ It is a topic that is partitioned and for this sample we assume that the topic i public class PulsarBootPartitioned { public static void main(String[] args) { - SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition"); + SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.message-routing-mode=CustomPartition"); } @Bean @@ -421,7 +436,7 @@ In order to do that, we are providing a message router object with the send meth Look at the three message routers implemented. `FooRouter` always sends data to partition `0`, `BarRouter` to partition `1` and `BuzzRouter` to partition `2`. Also note that, we are now using the `sendAsync` method of `PulsarTemplate` that returns a `CompletableFuture`. -When running the application, we also need to set the `messageRoutingMode` on the producer to `CustomPartition` (`spring.pulsar.producer.messageRoutingMode`). +When running the application, we also need to set the `messageRoutingMode` on the producer to `CustomPartition` (`spring.pulsar.producer.message-routing-mode`). On the consumer side, we are using a `PulsarListener` with the exclusive subscription type. This means that data from all the partitions will end up in the same consumer and there is no ordering guarantee. diff --git a/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java b/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java index df1f71b0..e82b7243 100644 --- a/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java @@ -45,9 +45,12 @@ public class FailoverConsumerApp { String topic = "failover-demo-topic"; return args -> { for (int i = 0; i < 10; i++) { - pulsarTemplate.sendAsync(topic, "hello john doe 0 ", new FooRouter()); - pulsarTemplate.sendAsync(topic, "hello alice doe 1", new BarRouter()); - pulsarTemplate.sendAsync(topic, "hello buzz doe 2", new BuzzRouter()); + pulsarTemplate.newMessage("hello john doe 0 ").withTopic(topic).withCustomRouter(new FooRouter()) + .sendAsync(); + pulsarTemplate.newMessage("hello alice doe 1").withTopic(topic).withCustomRouter(new BarRouter()) + .sendAsync(); + pulsarTemplate.newMessage("hello buzz doe 2").withTopic(topic).withCustomRouter(new BuzzRouter()) + .sendAsync(); Thread.sleep(1_000); } }; 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 5d54b567..64c4dcba 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 @@ -34,201 +34,90 @@ public interface PulsarOperations { /** * Sends a message to the default topic in a blocking manner. * @param message the message to send - * @return the id of the sent message + * @return the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - default MessageId send(T message) throws PulsarClientException { - return send(null, message); - } + MessageId send(T message) throws PulsarClientException; /** * 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 - * @return the id of the sent message + * @return the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - default MessageId send(String topic, T message) throws PulsarClientException { - return send(topic, message, (MessageRouter) null); - } - - /** - * Sends a message to the default topic in a blocking manner. - * @param message the message to send - * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer - * @return the id of the sent message - * @throws PulsarClientException if an error occurs - */ - default MessageId send(T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) - throws PulsarClientException { - return send(message, typedMessageBuilderCustomizer, 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 - */ - default MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException { - return send(topic, message, null, messageRouter); - } - - /** - * Sends a message to the default topic in a blocking manner. - * @param message the message to send - * @param typedMessageBuilderCustomizer the TypeMessageBuilder customizer - * @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, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, - MessageRouter messageRouter) throws PulsarClientException { - return send(null, message, typedMessageBuilderCustomizer, 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 typedMessageBuilderCustomizer the TypeMessageBuilder customizer - * @return the id of the sent message - * @throws PulsarClientException if an error occurs - */ - default MessageId send(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) - throws PulsarClientException { - return send(topic, message, typedMessageBuilderCustomizer, null); - } - - /** - * 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 typedMessageBuilderCustomizer the optional TypedMessageBuilder customizer - * @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, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, - MessageRouter messageRouter) throws PulsarClientException; + MessageId send(String topic, T message) throws PulsarClientException; /** * Sends a message to the default topic in a non-blocking manner. * @param message the message to send - * @return a future that holds the id of the sent message + * @return a future that holds the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - default CompletableFuture sendAsync(T message) throws PulsarClientException { - return sendAsync(null, message); - } + CompletableFuture sendAsync(T message) throws PulsarClientException; /** * Sends a message to the specified topic in a non-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 - * @return a future that holds the id of the sent message + * @return a future that holds the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - default CompletableFuture sendAsync(String topic, T message) throws PulsarClientException { - return sendAsync(topic, message, (MessageRouter) null); - } + CompletableFuture sendAsync(String topic, T message) throws PulsarClientException; /** - * Sends a message to the default topic in a non-blocking manner. - * @param message the message to send - * @param messageRouter the optional message router to use - * @return a future that holds the id of the sent message - * @throws PulsarClientException if an error occurs + * Create a {@link SendMessageBuilder builder} for configuring and sending a message. + * @param message the payload of the message + * @return the builder to configure and send the message */ - default CompletableFuture sendAsync(T message, MessageRouter messageRouter) - throws PulsarClientException { - return sendAsync(null, message, messageRouter); - } + SendMessageBuilder newMessage(T message); /** - * Sends a message to the default topic in a non-blocking manner. - * @param message the message to send - * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer - * @return a future that holds the id of the sent message - * @throws PulsarClientException if an error occurs + * Builder that can be used to configure and send a message. Provides more options + * than the basic send/sendAsync methods provided by {@link PulsarOperations}. + * + * @param the message payload type */ - default CompletableFuture sendAsync(T message, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) throws PulsarClientException { - return sendAsync(null, message, typedMessageBuilderCustomizer); - } + interface SendMessageBuilder { - /** - * Sends a message to the specified topic in a non-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 typedMessageBuilderCustomizer the TypedMessageBuilder customizer - * @return a future that holds the id of the sent message - * @throws PulsarClientException if an error occurs - */ - default CompletableFuture sendAsync(String topic, T message, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) throws PulsarClientException { - return sendAsync(topic, message, typedMessageBuilderCustomizer, null); - } + /** + * Specify the topic to send the message to. + * @param topic the destination topic + * @return the current builder with the destination topic specified + */ + SendMessageBuilder withTopic(String topic); - /** - * Sends a message to the default in a non-blocking manner. - * @param message the message to send - * @param typedMessageBuilderCustomizer the TypedMessageBuilder customizer - * @param messageRouter the optional message router to use - * @return a future that holds the id of the sent message - * @throws PulsarClientException if an error occurs - */ - default CompletableFuture sendAsync(T message, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) - throws PulsarClientException { - return sendAsync(null, message, typedMessageBuilderCustomizer, messageRouter); - } + /** + * Specifies the message customizer to use to further configure the message. + * @param messageCustomizer the message customizer + * @return the current builder with the message customizer specified + */ + SendMessageBuilder withMessageCustomizer(TypedMessageBuilderCustomizer messageCustomizer); - /** - * Sends a message to the specified topic in a non-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 a future that holds the id of the sent message - * @throws PulsarClientException if an error occurs - */ - default CompletableFuture sendAsync(String topic, T message, MessageRouter messageRouter) - throws PulsarClientException { - return sendAsync(topic, message, null, messageRouter); - } + /** + * Specifies the custom message router to use when sending the message. + * @param messageRouter the custom message router + * @return the current builder with the custom message router specified + */ + SendMessageBuilder withCustomRouter(MessageRouter messageRouter); - /** - * Sends a message to the specified topic in a non-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 typedMessageBuilderCustomizer the optional TypedMessageBuilder customizer - * @param messageRouter the optional message router to use - * @return a future that holds the id of the sent message - * @throws PulsarClientException if an error occurs - */ - CompletableFuture sendAsync(String topic, T message, - TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) - throws PulsarClientException; + /** + * Send the message in a blocking manner using the configured specification. + * @return the id assigned by the broker to the published message + * @throws PulsarClientException if an error occurs + */ + MessageId send() throws PulsarClientException; + + /** + * Uses the configured specification to send the message in a non-blocking manner. + * @return a future that holds the id assigned by the broker to the published + * message + * @throws PulsarClientException if an error occurs + */ + CompletableFuture sendAsync() throws PulsarClientException; + + } } 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 5b563818..f740d635 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 @@ -49,19 +49,17 @@ public class PulsarTemplate implements PulsarOperations { private Schema schema; /** - * Constructs a template instance. - * @param producerFactory the producer factory used to create the backing Pulsar - * producers. + * Construct a template instance. + * @param producerFactory the factory used to create the backing Pulsar producers. */ public PulsarTemplate(PulsarProducerFactory producerFactory) { this(producerFactory, null); } /** - * Constructs a template instance. - * @param producerFactory the producer factory used to create the backing Pulsar - * producers. - * @param interceptors the {@link ProducerInterceptor}s to add to the producer. + * Construct a template instance. + * @param producerFactory the factory used to create the backing Pulsar producers. + * @param interceptors the interceptors to add to the producer. */ public PulsarTemplate(PulsarProducerFactory producerFactory, List interceptors) { this.producerFactory = producerFactory; @@ -69,18 +67,49 @@ public class PulsarTemplate implements PulsarOperations { } @Override - public MessageId send(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, + public MessageId send(T message) throws PulsarClientException { + return doSend(null, message, null, null); + } + + @Override + public MessageId send(String topic, T message) throws PulsarClientException { + return doSend(topic, message, null, null); + } + + @Override + public CompletableFuture sendAsync(T message) throws PulsarClientException { + return doSendAsync(null, message, null, null); + } + + @Override + public CompletableFuture sendAsync(String topic, T message) throws PulsarClientException { + return doSendAsync(topic, message, null, null); + } + + @Override + public SendMessageBuilder newMessage(T message) { + return new SendMessageBuilderImpl<>(this, message); + } + + /** + * Setter for schema. + * @param schema provides the {@link Schema} used on this template + */ + public void setSchema(Schema schema) { + this.schema = schema; + } + + private MessageId doSend(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException { try { - return this.sendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter).get(); + return doSendAsync(topic, message, typedMessageBuilderCustomizer, messageRouter).get(); } catch (Exception ex) { throw PulsarClientException.unwrap(ex); } } - @Override - public CompletableFuture sendAsync(String topic, T message, + private CompletableFuture doSendAsync(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException { final String topicName = ProducerUtils.resolveTopicName(topic, this.producerFactory); @@ -109,12 +138,51 @@ public class PulsarTemplate implements PulsarOperations { return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors); } - /** - * Setter for schema. - * @param schema provides the {@link Schema} used on this template - */ - public void setSchema(Schema schema) { - this.schema = schema; + public static class SendMessageBuilderImpl implements SendMessageBuilder { + + private final PulsarTemplate template; + + private final T message; + + private String topic; + + private TypedMessageBuilderCustomizer messageCustomizer; + + private MessageRouter messageRouter; + + SendMessageBuilderImpl(PulsarTemplate template, T message) { + this.template = template; + this.message = message; + } + + @Override + public SendMessageBuilder withTopic(String topic) { + this.topic = topic; + return this; + } + + @Override + public SendMessageBuilder withMessageCustomizer(TypedMessageBuilderCustomizer messageCustomizer) { + this.messageCustomizer = messageCustomizer; + return this; + } + + @Override + public SendMessageBuilder withCustomRouter(MessageRouter messageRouter) { + this.messageRouter = messageRouter; + return this; + } + + @Override + public MessageId send() throws PulsarClientException { + return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter); + } + + @Override + public CompletableFuture sendAsync() throws PulsarClientException { + return this.template.doSendAsync(this.topic, this.message, this.messageCustomizer, this.messageRouter); + } + } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java index 686ecf58..0bbbc547 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/FailoverConsumerTests.java @@ -82,9 +82,10 @@ class FailoverConsumerTests extends AbstractContainerBaseTests { pulsarClient, prodConfig); final PulsarTemplate pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory); - pulsarTemplate.sendAsync("hello john doe", new FooRouter()); - pulsarTemplate.sendAsync("hello alice doe", new BarRouter()); - pulsarTemplate.sendAsync("hello buzz doe", new BuzzRouter()); + pulsarTemplate.newMessage("hello john doe").withCustomRouter(new FooRouter()).sendAsync(); + pulsarTemplate.newMessage("hello alice doe").withCustomRouter(new BarRouter()).sendAsync(); + pulsarTemplate.newMessage("hello buzz doe").withCustomRouter(new BuzzRouter()).sendAsync(); + final boolean await = latch.await(10, TimeUnit.SECONDS); assertThat(await).isTrue(); } 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 fb5073a1..8b40f1ae 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 @@ -140,76 +140,98 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { private static Stream sendMessageTestProvider() { return Stream.of( - - arguments(Named.of("sendMessageToDefaultTopic", "smt-topic-1"), - Collections.singletonMap("topicName", "smt-topic-1"), + arguments("sendMessageToDefaultTopic", + Collections.singletonMap("topicName", "sendMessageToDefaultTopic"), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .send(), + null, null), + arguments("sendMessageToDefaultTopicWithSimpleApi", + Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithSimpleApi"), (SendHandler) (template, topic, msg, customizer, router) -> template.send(msg), null, null), - arguments(Named.of("sendMessageToDefaultTopicWithRouter", "smt-topic-2"), - Collections.singletonMap("topicName", "smt-topic-2"), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(msg, - router), + arguments("sendMessageToDefaultTopicWithRouter", + Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithRouter"), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .withCustomRouter(router).send(), null, mockRouter()), - arguments(Named.of("sendMessageToDefaultTopicWithCustomizer", "smt-topic-3"), - Collections.singletonMap("topicName", "smt-topic-3"), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(msg, - customizer), + arguments("sendMessageToDefaultTopicWithCustomizer", + Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizer"), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .withMessageCustomizer(customizer).send(), sampleMessageKeyCustomizer, null), - arguments(Named.of("sendMessageToDefaultTopicWithCustomizerAndRouter", "smt-topic-4"), - Collections.singletonMap("topicName", "smt-topic-4"), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(msg, - customizer, router), + arguments("sendMessageToDefaultTopicWithCustomizerAndRouter", + Collections.singletonMap("topicName", "sendMessageToDefaultTopicWithCustomizerAndRouter"), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .withMessageCustomizer(customizer).withCustomRouter(router).send(), sampleMessageKeyCustomizer, mockRouter()), - arguments(Named.of("sendMessageToSpecificTopic", "smt-topic-5"), Collections.emptyMap(), + 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(Named.of("sendMessageToSpecificTopicWithRouter", "smt-topic-6"), Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(topic, msg, - null, router), + arguments("sendMessageToSpecificTopicWithRouter", Collections.emptyMap(), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .withTopic(topic).withCustomRouter(router).send(), null, mockRouter()), - arguments(Named.of("sendMessageToSpecificTopicWithCustomizer", "smt-topic-7"), Collections.emptyMap(), - (SendHandler) (template, topic, msg, customizer, router) -> template.send(topic, msg, - customizer), + arguments("sendMessageToSpecificTopicWithCustomizer", Collections.emptyMap(), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .withMessageCustomizer(customizer).withTopic(topic).send(), sampleMessageKeyCustomizer, null), - arguments(Named.of("sendMessageToSpecificTopicWithCustomizerAndRouter", "smt-topic-8"), - Collections.emptyMap(), (SendHandler) PulsarTemplate::send, + arguments("sendMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(), + (SendHandler) (template, topic, msg, customizer, router) -> template.newMessage(msg) + .withMessageCustomizer(customizer).withTopic(topic).withCustomRouter(router).send(), sampleMessageKeyCustomizer, mockRouter()), - arguments(Named.of("sendAsyncMessageToDefaultTopic", "smt-topic-9"), - Collections.singletonMap("topicName", "smt-topic-9"), + arguments("sendAsyncMessageToDefaultTopic", + Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopic"), + (SendHandler>) (template, topic, msg, customizer, + router) -> template.newMessage(msg).sendAsync(), + null, null), + arguments("sendAsyncMessageToDefaultTopicWithSimpleApi", + Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithSimpleApi"), (SendHandler>) (template, topic, msg, customizer, router) -> template.sendAsync(msg), null, null), - arguments(Named.of("sendAsyncMessageToDefaultTopicWithRouter", "smt-topic-10"), - Collections.singletonMap("topicName", "smt-topic-10"), + arguments("sendAsyncMessageToDefaultTopicWithRouter", + Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithRouter"), (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(msg, router), + router) -> template.newMessage(msg).withCustomRouter(router).sendAsync(), null, mockRouter()), - arguments(Named.of("sendAsyncMessageToDefaultTopicWithCustomizer", "smt-topic-11"), - Collections.singletonMap("topicName", "smt-topic-11"), + arguments("sendAsyncMessageToDefaultTopicWithCustomizer", + Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizer"), (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(msg, customizer), + router) -> template.newMessage(msg).withMessageCustomizer(customizer).sendAsync(), sampleMessageKeyCustomizer, null), - arguments(Named.of("sendAsyncMessageToDefaultTopicWithCustomizerAndRouter", "smt-topic-12"), - Collections.singletonMap("topicName", "smt-topic-12"), + arguments("sendAsyncMessageToDefaultTopicWithCustomizerAndRouter", + Collections.singletonMap("topicName", "sendAsyncMessageToDefaultTopicWithCustomizerAndRouter"), (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(msg, customizer, router), + router) -> template.newMessage(msg).withMessageCustomizer(customizer) + .withCustomRouter(router).sendAsync(), sampleMessageKeyCustomizer, mockRouter()), - arguments(Named.of("sendAsyncMessageToSpecificTopic", "smt-topic-13"), Collections.emptyMap(), + 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(Named.of("sendAsyncMessageToSpecificTopicWithRouter", "smt-topic-14"), Collections.emptyMap(), + arguments("sendAsyncMessageToSpecificTopicWithRouter", Collections.emptyMap(), (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(topic, msg, null, router), + router) -> template.newMessage(msg).withTopic(topic).withCustomRouter(router) + .sendAsync(), null, mockRouter()), - arguments(Named.of("sendAsyncMessageToSpecificTopicWithCustomizer", "smt-topic-15"), - Collections.emptyMap(), + arguments("sendAsyncMessageToSpecificTopicWithCustomizer", Collections.emptyMap(), (SendHandler>) (template, topic, msg, customizer, - router) -> template.sendAsync(topic, msg, customizer), + router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic) + .sendAsync(), sampleMessageKeyCustomizer, null), - arguments(Named.of("sendAsyncMessageToSpecificTopicWithCustomizerAndRouter", "smt-topic-16"), - Collections.emptyMap(), (SendHandler>) PulsarTemplate::sendAsync, + arguments("sendAsyncMessageToSpecificTopicWithCustomizerAndRouter", Collections.emptyMap(), + (SendHandler>) (template, topic, msg, customizer, + router) -> template.newMessage(msg).withMessageCustomizer(customizer).withTopic(topic) + .withCustomRouter(router).sendAsync(), sampleMessageKeyCustomizer, mockRouter())); }