diff --git a/build.gradle b/build.gradle index 5005b504..8833a562 100644 --- a/build.gradle +++ b/build.gradle @@ -76,3 +76,4 @@ subprojects { subproject -> } } } + diff --git a/buildSrc/src/main/groovy/io/spring/gradle/convention/RepositoryConventionPlugin.groovy b/buildSrc/src/main/groovy/io/spring/gradle/convention/RepositoryConventionPlugin.groovy index 5ea5d2d3..67eb897b 100644 --- a/buildSrc/src/main/groovy/io/spring/gradle/convention/RepositoryConventionPlugin.groovy +++ b/buildSrc/src/main/groovy/io/spring/gradle/convention/RepositoryConventionPlugin.groovy @@ -47,6 +47,10 @@ class RepositoryConventionPlugin implements Plugin { } url = 'https://repo.spring.io/snapshot/' } + maven { + name = 'apache-snapshot' + url = 'https://repository.apache.org/content/repositories/snapshots' + } } if (isSnapshot || isMilestone) { maven { diff --git a/spring-pulsar-dependencies/build.gradle b/spring-pulsar-dependencies/build.gradle index 8c6e274a..a53aa188 100644 --- a/spring-pulsar-dependencies/build.gradle +++ b/spring-pulsar-dependencies/build.gradle @@ -13,6 +13,7 @@ ext { protobufJavaVersion = '3.21.5' testcontainersVersion = '1.17.3' pulsarVersion = '2.10.1' + pulsarClientReactiveVersion = '0.1.0-SNAPSHOT' springBootVersion = '3.0.0-SNAPSHOT' } @@ -26,5 +27,6 @@ dependencies { api "com.google.code.findbugs:jsr305:$googleJsr305Version" api "com.google.protobuf:protobuf-java:$protobufJavaVersion" api "org.apache.pulsar:pulsar-client-all:$pulsarVersion" + api "org.apache.pulsar:pulsar-client-reactive-adapter:$pulsarClientReactiveVersion" } } diff --git a/spring-pulsar/build.gradle b/spring-pulsar/build.gradle index 177c0938..fc6d839c 100644 --- a/spring-pulsar/build.gradle +++ b/spring-pulsar/build.gradle @@ -9,6 +9,7 @@ dependencies { api 'com.google.protobuf:protobuf-java' api 'io.micrometer:micrometer-observation' api 'org.apache.pulsar:pulsar-client-all' + api "org.apache.pulsar:pulsar-client-reactive-adapter" api 'org.springframework:spring-context' api 'org.springframework:spring-messaging' api 'org.springframework:spring-tx' diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java new file mode 100644 index 00000000..6a2d8ba3 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/DefaultReactivePulsarSenderFactory.java @@ -0,0 +1,98 @@ +/* + * 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.reactive; + +import java.util.List; + +import org.apache.pulsar.client.api.MessageRouter; +import org.apache.pulsar.client.api.PulsarClient; +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.ReactiveMessageSenderSpec; +import org.apache.pulsar.reactive.client.api.ReactivePulsarClient; + +import org.springframework.core.log.LogAccessor; +import org.springframework.util.CollectionUtils; + +/** + * Default implementation of {@link ReactivePulsarSenderFactory}. + * + * @param reactive sender type. + * @author Christophe Bornet + */ +public class DefaultReactivePulsarSenderFactory implements ReactivePulsarSenderFactory { + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final ReactivePulsarClient reactivePulsarClient; + + private final ReactiveMessageSenderSpec reactiveMessageSenderSpec; + + public DefaultReactivePulsarSenderFactory(PulsarClient pulsarClient, + ReactiveMessageSenderSpec reactiveMessageSenderSpec) { + this(AdaptedReactivePulsarClientFactory.create(pulsarClient), reactiveMessageSenderSpec); + } + + public DefaultReactivePulsarSenderFactory(ReactivePulsarClient reactivePulsarClient, + ReactiveMessageSenderSpec reactiveMessageSenderSpec) { + this.reactivePulsarClient = reactivePulsarClient; + this.reactiveMessageSenderSpec = reactiveMessageSenderSpec; + } + + @Override + public ReactiveMessageSender createReactiveMessageSender(String topic, Schema schema) { + return doCreateReactiveMessageSender(topic, schema, null, null); + } + + @Override + public ReactiveMessageSender createReactiveMessageSender(String topic, Schema schema, + MessageRouter messageRouter) { + return doCreateReactiveMessageSender(topic, schema, messageRouter, null); + } + + @Override + public ReactiveMessageSender createReactiveMessageSender(String topic, Schema schema, + MessageRouter messageRouter, List> customizers) { + return doCreateReactiveMessageSender(topic, schema, messageRouter, customizers); + } + + private ReactiveMessageSender doCreateReactiveMessageSender(String topic, Schema schema, + MessageRouter messageRouter, List> customizers) { + final String resolvedTopic = ReactiveMessageSenderUtils.resolveTopicName(topic, this); + this.logger.trace(() -> String.format("Creating reactive message sender for '%s' topic", resolvedTopic)); + final ReactiveMessageSenderBuilder sender = this.reactivePulsarClient.messageSender(schema); + if (this.reactiveMessageSenderSpec != null) { + sender.applySpec(this.reactiveMessageSenderSpec); + } + sender.topic(resolvedTopic); + if (messageRouter != null) { + sender.messageRouter(messageRouter); + } + if (!CollectionUtils.isEmpty(customizers)) { + customizers.forEach((c) -> c.customize(sender)); + } + return sender.build(); + } + + @Override + public ReactiveMessageSenderSpec getReactiveMessageSenderSpec() { + return this.reactiveMessageSenderSpec; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.java new file mode 100644 index 00000000..f62394c5 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/MessageSpecBuilderCustomizer.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.reactive; + +import org.apache.pulsar.reactive.client.api.MessageSpecBuilder; + +/** + * The interface to customize a {@link MessageSpecBuilder}. + * + * @param The message payload type + * @author Christophe Bornet + */ +@FunctionalInterface +public interface MessageSpecBuilderCustomizer { + + /** + * Customizes a {@link MessageSpecBuilder}. + * @param messageSpecBuilder the MessageSpecBuilder to customize + */ + void customize(MessageSpecBuilder messageSpecBuilder); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.java new file mode 100644 index 00000000..b95e9749 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderBuilderCustomizer.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.reactive; + +import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderBuilder; + +/** + * The interface to customize a {@link ReactiveMessageSenderBuilder}. + * + * @param The message payload type + * @author Christophe Bornet + */ +@FunctionalInterface +public interface ReactiveMessageSenderBuilderCustomizer { + + /** + * Customizes a {@link ReactiveMessageSenderBuilder}. + * @param reactiveMessageSenderBuilder the builder to customize + */ + void customize(ReactiveMessageSenderBuilder reactiveMessageSenderBuilder); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java new file mode 100644 index 00000000..be9857b7 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactiveMessageSenderUtils.java @@ -0,0 +1,46 @@ +/* + * 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.reactive; + +import java.util.Optional; + +import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec; + +import org.springframework.util.StringUtils; + +/** + * Common utilities used by reactive sender components. + * + * @author Christophe Bornet + */ +final class ReactiveMessageSenderUtils { + + private ReactiveMessageSenderUtils() { + } + + static String resolveTopicName(String userSpecifiedTopic, + ReactivePulsarSenderFactory reactiveMessageSenderFactory) { + ReactiveMessageSenderSpec reactiveMessageSenderSpec = reactiveMessageSenderFactory + .getReactiveMessageSenderSpec(); + if (StringUtils.hasText(userSpecifiedTopic)) { + return userSpecifiedTopic; + } + return Optional.ofNullable(reactiveMessageSenderSpec).map(ReactiveMessageSenderSpec::getTopicName).orElseThrow( + () -> new IllegalArgumentException("Topic must be specified when no default topic is configured")); + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java new file mode 100644 index 00000000..d9c801e8 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderFactory.java @@ -0,0 +1,72 @@ +/* + * 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.reactive; + +import java.util.List; + +import org.apache.pulsar.client.api.MessageRouter; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.reactive.client.api.ReactiveMessageSender; +import org.apache.pulsar.reactive.client.api.ReactiveMessageSenderSpec; + +/** + * The strategy to create a {@link ReactiveMessageSender} instance(s). + * + * @param reactive message sender payload type + * @author Christophe Bornet + */ +public interface ReactivePulsarSenderFactory { + + /** + * Create a reactive message sender. + * @param topic the topic the reactive message sender will send messages to or + * {@code null} to use the default topic + * @param schema the schema of the messages to be sent + * @return the reactive message sender + */ + ReactiveMessageSender createReactiveMessageSender(String topic, Schema schema); + + /** + * Create a reactive message sender. + * @param topic the topic the reactive message sender 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 + * @return the reactive message sender + */ + ReactiveMessageSender createReactiveMessageSender(String topic, Schema schema, MessageRouter messageRouter); + + /** + * Create a reactive message sender. + * @param topic the topic the reactive message sender 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 customizers the optional list of customizers to apply to the reactive + * message sender builder + * @return the reactive message sender + */ + ReactiveMessageSender createReactiveMessageSender(String topic, Schema schema, MessageRouter messageRouter, + List> customizers); + + /** + * Return the ReactiveMessageSenderSpec to use when creating reactive senders. + * @return the ReactiveMessageSenderSpec + */ + ReactiveMessageSenderSpec getReactiveMessageSenderSpec(); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderOperations.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderOperations.java new file mode 100644 index 00000000..1d0d5352 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderOperations.java @@ -0,0 +1,102 @@ +/* + * 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.reactive; + +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.MessageRouter; + +import reactor.core.publisher.Mono; + +/** + * The Pulsar reactive send operations contract. + * + * @param the message payload type + * @author Christophe Bornet + */ +public interface ReactivePulsarSenderOperations { + + /** + * Sends a message to the default topic in a reactive manner. + * @param message the message to send + * @return the id assigned by the broker to the published message + */ + Mono send(T message); + + /** + * Sends a message to the specified topic in a reactive 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 assigned by the broker to the published message + */ + Mono send(String topic, T message); + + /** + * Create a {@link SendMessageBuilder builder} for configuring and sending a message + * reactively. + * @param message the payload of the message + * @return the builder to configure and send the message + */ + SendMessageBuilder newMessage(T message); + + /** + * Builder that can be used to configure and send a message. Provides more options + * than the send methods provided by {@link ReactivePulsarSenderOperations}. + * + * @param the message payload type + */ + interface SendMessageBuilder { + + /** + * 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); + + /** + * Specifies the message customizer to use to further configure the message. + * @param customizer the message customizer + * @return the current builder with the message customizer specified + */ + SendMessageBuilder withMessageCustomizer(MessageSpecBuilderCustomizer customizer); + + /** + * 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); + + /** + * Specifies the customizer to use to further configure the reactive sender + * builder. + * @param customizer the reactive sender builder customizer + * @return the current builder with the reactive sender builder customizer + * specified + */ + SendMessageBuilder withSenderCustomizer(ReactiveMessageSenderBuilderCustomizer customizer); + + /** + * Send the message in a reactive manner using the configured specification. + * @return the id assigned by the broker to the published message + */ + Mono send(); + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderTemplate.java new file mode 100644 index 00000000..c76646f5 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/reactive/ReactivePulsarSenderTemplate.java @@ -0,0 +1,159 @@ +/* + * 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.reactive; + +import java.util.Collections; + +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.MessageRouter; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.reactive.client.api.MessageSpec; +import org.apache.pulsar.reactive.client.api.MessageSpecBuilder; +import org.apache.pulsar.reactive.client.api.ReactiveMessageSender; + +import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.core.SchemaUtils; + +import reactor.core.publisher.Mono; + +/** + * A thread-safe template for executing high-level reactive Pulsar operations. + * + * @param the message payload type + * @author Christophe Bornet + */ +public class ReactivePulsarSenderTemplate implements ReactivePulsarSenderOperations { + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final ReactivePulsarSenderFactory reactiveMessageSenderFactory; + + private Schema schema; + + /** + * Construct a template instance with observation configuration. + * @param reactiveMessageSenderFactory the factory used to create the backing Pulsar + * reactive senders + */ + public ReactivePulsarSenderTemplate(ReactivePulsarSenderFactory reactiveMessageSenderFactory) { + this.reactiveMessageSenderFactory = reactiveMessageSenderFactory; + } + + @Override + public Mono send(T message) { + return doSend(null, message, null, null, null); + } + + @Override + public Mono send(String topic, T message) { + return doSend(topic, message, null, null, null); + } + + @Override + public SendMessageBuilderImpl newMessage(T message) { + return new SendMessageBuilderImpl<>(this, message); + } + + /** + * Set the schema to use on this template. + * @param schema provides the {@link Schema} used on this template + */ + public void setSchema(Schema schema) { + this.schema = schema; + } + + private Mono doSend(String topic, T message, + MessageSpecBuilderCustomizer messageSpecBuilderCustomizer, MessageRouter messageRouter, + ReactiveMessageSenderBuilderCustomizer customizer) { + final String topicName = ReactiveMessageSenderUtils.resolveTopicName(topic, this.reactiveMessageSenderFactory); + this.logger.trace(() -> String.format("Sending reative msg to '%s' topic", topicName)); + + final ReactiveMessageSender sender = createMessageSender(topic, message, messageRouter, customizer); + + MessageSpecBuilder messageSpecBuilder = MessageSpec.builder(message); + + if (messageSpecBuilderCustomizer != null) { + messageSpecBuilderCustomizer.customize(messageSpecBuilder); + } + + MessageSpec messageSpec = messageSpecBuilder.build(); + return sender.sendMessage(Mono.just(messageSpec)) + .doOnError( + ex -> this.logger.error(ex, () -> String.format("Failed to send msg to '%s' topic", topicName))) + .doOnSuccess(msgId -> this.logger.trace(() -> String.format("Sent msg to '%s' topic", topicName))); + } + + private ReactiveMessageSender createMessageSender(String topic, T message, MessageRouter messageRouter, + ReactiveMessageSenderBuilderCustomizer customizer) { + Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); + return this.reactiveMessageSenderFactory.createReactiveMessageSender(topic, schema, messageRouter, + customizer == null ? Collections.emptyList() : Collections.singletonList(customizer)); + } + + public static class SendMessageBuilderImpl implements SendMessageBuilder { + + private final ReactivePulsarSenderTemplate template; + + private final T message; + + private String topic; + + private MessageSpecBuilderCustomizer messageCustomizer; + + private MessageRouter messageRouter; + + private ReactiveMessageSenderBuilderCustomizer senderCustomizer; + + SendMessageBuilderImpl(ReactivePulsarSenderTemplate template, T message) { + this.template = template; + this.message = message; + } + + @Override + public SendMessageBuilderImpl withTopic(String topic) { + this.topic = topic; + return this; + } + + @Override + public SendMessageBuilderImpl withMessageCustomizer(MessageSpecBuilderCustomizer messageCustomizer) { + this.messageCustomizer = messageCustomizer; + return this; + } + + @Override + public SendMessageBuilderImpl withCustomRouter(MessageRouter messageRouter) { + this.messageRouter = messageRouter; + return this; + } + + @Override + public SendMessageBuilderImpl withSenderCustomizer( + ReactiveMessageSenderBuilderCustomizer senderCustomizer) { + this.senderCustomizer = senderCustomizer; + return this; + } + + @Override + public Mono send() { + return this.template.doSend(this.topic, this.message, this.messageCustomizer, this.messageRouter, + this.senderCustomizer); + } + + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java new file mode 100644 index 00000000..78623e7d --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/DefaultReactiveMessageSenderFactoryTests.java @@ -0,0 +1,158 @@ +/* + * 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.reactive; + +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 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.api.MutableReactiveMessageSenderSpec; +import org.apache.pulsar.reactive.client.api.ReactiveMessageSender; +import org.junit.jupiter.api.BeforeEach; +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 schema = Schema.STRING; + + private ProducerBuilder producerBuilder; + + private PulsarClient pulsarClient; + + @BeforeEach + void createPulsarClient() { + pulsarClient = mock(PulsarClient.class); + producerBuilder = mock(ProducerBuilder.class); + Producer producer = mock(Producer.class); + TypedMessageBuilder 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 createProducerWithSpecificTopic() { + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + null); + ReactiveMessageSender sender = senderFactory.createReactiveMessageSender("topic1", schema); + sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5)); + assertSenderHasTopicAndRouter("topic1", null); + } + + @Test + void createProducerWithSpecificTopicAndMessageRouter() { + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + null); + MessageRouter router = mock(MessageRouter.class); + ReactiveMessageSender sender = senderFactory.createReactiveMessageSender("topic1", schema, router); + sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5)); + assertSenderHasTopicAndRouter("topic1", router); + } + + @Test + void createProducerWithDefaultTopic() { + MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); + senderSpec.setTopicName("topic0"); + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + senderSpec); + ReactiveMessageSender sender = senderFactory.createReactiveMessageSender(null, schema); + sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5)); + assertSenderHasTopicAndRouter("topic0", null); + } + + @Test + void createProducerWithDefaultTopicAndMessageRouter() { + MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); + senderSpec.setTopicName("topic0"); + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + senderSpec); + MessageRouter router = mock(MessageRouter.class); + ReactiveMessageSender sender = senderFactory.createReactiveMessageSender(null, schema, router); + sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5)); + assertSenderHasTopicAndRouter("topic0", router); + } + + @Test + void createProducerWithSingleProducerCustomizer() { + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + null); + ReactiveMessageSenderBuilderCustomizer customizer = builder -> builder.topic("topic1"); + ReactiveMessageSender sender = senderFactory.createReactiveMessageSender("topic0", schema, null, + Collections.singletonList(customizer)); + sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5)); + assertSenderHasTopicAndRouter("topic1", null); + } + + @Test + void createProducerWithMultipleProducerCustomizer() { + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + null); + ReactiveMessageSenderBuilderCustomizer customizer1 = builder -> builder.topic("topic1"); + MessageRouter router = mock(MessageRouter.class); + ReactiveMessageSenderBuilderCustomizer customizer2 = builder -> builder.messageRouter(router); + ReactiveMessageSender sender = senderFactory.createReactiveMessageSender("topic0", schema, null, + Arrays.asList(customizer1, customizer2)); + sender.sendMessage(Mono.just(MessageSpec.of("test"))).block(Duration.ofSeconds(5)); + assertSenderHasTopicAndRouter("topic1", router); + } + + @Test + void createProducerWithNoTopic() { + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(pulsarClient, + 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()); + } + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java new file mode 100644 index 00000000..9e691da2 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/reactive/ReactivePulsarTemplateTests.java @@ -0,0 +1,235 @@ +/* + * 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.reactive; + +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.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.time.Duration; +import java.util.UUID; +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.MessageRouter; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TopicMetadata; +import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +import org.springframework.pulsar.core.PulsarTestContainerSupport; + +import reactor.core.publisher.Mono; + +/** + * Tests for {@link ReactivePulsarSenderTemplate}. + * + * @author Christophe Bornet + */ +class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { + + @Test + void sendMessageWithSpecificSchemaTest() throws Exception { + String topic = "smt-specific-schema-topic-reactive"; + try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) + .subscriptionName("test-specific-schema-subscription").subscribe()) { + MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); + senderSpec.setTopicName(topic); + ReactivePulsarSenderFactory producerFactory = new DefaultReactivePulsarSenderFactory<>(client, + senderSpec); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(producerFactory); + pulsarTemplate.setSchema(Schema.JSON(Foo.class)); + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + pulsarTemplate.send(foo).subscribe(); + assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) + .isEqualTo(foo); + } + } + } + + @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); + } + MessageSpecBuilderCustomizer messageCustomizer = null; + if (testArgs.useMessageCustomizer) { + messageCustomizer = (mb) -> mb.key("foo-key"); + } + ReactiveMessageSenderBuilderCustomizer senderCustomizer = null; + if (testArgs.useSenderCustomizer) { + senderCustomizer = (sb) -> sb.producerName("foo-producer"); + } + + if (router != null) { + try (PulsarAdmin admin = PulsarAdmin.builder() + .serviceHttpUrl(PulsarTestContainerSupport.getHttpServiceUrl()).build()) { + admin.topics().createPartitionedTopic("persistent://public/default/" + topic, 1); + } + } + try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.STRING).topic(topic) + .subscriptionName(subscription).subscribe()) { + MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); + if (!testArgs.useSpecificTopic) { + senderSpec.setTopicName(topic); + } + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, + senderSpec); + ReactivePulsarSenderTemplate pulsarTemplate = new ReactivePulsarSenderTemplate<>(senderFactory); + Mono sendResponse; + if (testArgs.useSimpleApi) { + sendResponse = testArgs.useSpecificTopic ? pulsarTemplate.send(topic, msgPayload) + : pulsarTemplate.send(msgPayload); + } + else { + ReactivePulsarSenderTemplate.SendMessageBuilderImpl 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 (senderCustomizer != null) { + messageBuilder = messageBuilder.withSenderCustomizer(senderCustomizer); + } + sendResponse = messageBuilder.send(); + } + sendResponse.subscribe(); + + Message msg = consumer.receive(3, TimeUnit.SECONDS); + + assertThat(msg).isNotNull(); + 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 (senderCustomizer != 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("sendReactiveMessageToDefaultTopic", SendTestArgs.useSpecificTopic(false)), + arguments("sendReactiveMessageToDefaultTopicWithSimpleApi", + SendTestArgs.useSpecificTopic(false).useSimpleApi(true)), + arguments("sendReactiveMessageToDefaultTopicWithRouter", + SendTestArgs.useSpecificTopic(false).useCustomRouter(true)), + arguments("sendReactiveMessageToDefaultTopicWithMessageCustomizer", + SendTestArgs.useSpecificTopic(false).useMessageCustomizer(true)), + arguments("sendReactiveMessageToDefaultTopicWithProducerCustomizer", + SendTestArgs.useSpecificTopic(false).useSenderCustomizer(true)), + arguments("sendReactiveMessageToDefaultTopicWithAllOptions", + SendTestArgs.useSpecificTopic(false).useCustomRouter(true).useMessageCustomizer(true) + .useSenderCustomizer(true)), + arguments("sendReactiveMessageToSpecificTopic", SendTestArgs.useSpecificTopic(true)), + arguments("sendReactiveMessageToSpecificTopicWithSimpleApi", + SendTestArgs.useSpecificTopic(true).useSimpleApi(true)), + arguments("sendReactiveMessageToSpecificTopicWithRouter", + SendTestArgs.useSpecificTopic(true).useCustomRouter(true)), + arguments("sendReactiveMessageToSpecificTopicWithMessageCustomizer", + SendTestArgs.useSpecificTopic(true).useMessageCustomizer(true)), + arguments("sendReactiveMessageToSpecificTopicWithProducerCustomizer", + SendTestArgs.useSpecificTopic(true).useSenderCustomizer(true)), + arguments("sendReactiveMessageToSpecificTopicWithAllOptions", SendTestArgs.useSpecificTopic(true) + .useCustomRouter(true).useMessageCustomizer(true).useSenderCustomizer(true))); + } + + static final class SendTestArgs { + + private boolean useSpecificTopic; + + private boolean useCustomRouter; + + private boolean useMessageCustomizer; + + private boolean useSenderCustomizer; + + 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 useSenderCustomizer(boolean useSenderCustomizer) { + this.useSenderCustomizer = useSenderCustomizer; + return this; + } + + SendTestArgs useSimpleApi(boolean useSimpleApi) { + this.useSimpleApi = useSimpleApi; + return this; + } + + } + + record Foo(String foo, String bar) { + } + +}