From 33712c6beff58e493018b9bf4428ce2c9f8a03a4 Mon Sep 17 00:00:00 2001 From: Christophe Bornet Date: Mon, 6 Feb 2023 19:28:41 +0100 Subject: [PATCH] Add support for tombstone/null values Fixes #311 --- .../src/main/asciidoc/topic-resolution.adoc | 2 + .../MethodReactivePulsarListenerEndpoint.java | 9 +- .../DefaultReactivePulsarSenderFactory.java | 4 +- .../core/ReactivePulsarOperations.java | 10 +- .../reactive/core/ReactivePulsarTemplate.java | 44 +- .../core/ReactivePulsarTemplateTests.java | 437 +++++++---------- .../binder/PulsarMessageChannelBinder.java | 2 +- ...essageChannelBinderResolveSchemaTests.java | 23 +- .../config/MethodPulsarListenerEndpoint.java | 9 +- .../core/DefaultPulsarProducerFactory.java | 3 +- .../pulsar/core/DefaultSchemaResolver.java | 98 ++-- .../pulsar/core/DefaultTopicResolver.java | 29 +- .../pulsar/core/PulsarOperations.java | 20 +- .../pulsar/core/PulsarTemplate.java | 62 +-- .../springframework/pulsar/core/Resolved.java | 72 +++ .../pulsar/core/SchemaResolver.java | 16 +- .../pulsar/core/TopicResolver.java | 7 +- .../core/DefaultSchemaResolverTests.java | 52 +- .../core/DefaultTopicResolverTests.java | 20 +- .../pulsar/core/PulsarTemplateTests.java | 445 ++++++++---------- 20 files changed, 669 insertions(+), 695 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java diff --git a/spring-pulsar-docs/src/main/asciidoc/topic-resolution.adoc b/spring-pulsar-docs/src/main/asciidoc/topic-resolution.adoc index 90b0eb86..2b59695c 100644 --- a/spring-pulsar-docs/src/main/asciidoc/topic-resolution.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/topic-resolution.adoc @@ -36,6 +36,8 @@ spring: NOTE: The `message-type` is the fully-qualified name of the message class. +WARNING: If the message (or the first message of a `Publisher` input) is `null`, the framework won't be able to determine the topic from it. Another method shall be used to specify the topic if your application is likely to send `null` messages. + === Custom topic resolver The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can replace the default resolver by proving your own implementation, for example: diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java index 0764f3f2..02fbf5bc 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java @@ -138,10 +138,9 @@ public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePul SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver(); SchemaType schemaType = pulsarContainerProperties.getSchemaType(); ResolvableType messageType = resolvableType(messageParameter); - Schema schema = schemaResolver.getSchema(schemaType, messageType); - if (schema != null) { - pulsarContainerProperties.setSchema((Schema) schema); - } + schemaResolver.resolveSchema(schemaType, messageType) + .ifResolved(schema -> pulsarContainerProperties.setSchema((Schema) schema)); + // Make sure the schemaType is updated to match the current schema if (pulsarContainerProperties.getSchema() != null) { SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType(); @@ -154,7 +153,7 @@ public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePul || !ObjectUtils.isEmpty(pulsarContainerProperties.getTopics()); if (!hasTopicInfo) { topicResolver.resolveTopic(null, messageType.getRawClass(), () -> null) - .ifPresent((topic) -> pulsarContainerProperties.setTopics(Collections.singleton(topic))); + .ifResolved((topic) -> pulsarContainerProperties.setTopics(Collections.singleton(topic))); } ReactiveMessageConsumerBuilderCustomizer customizer1 = b -> b.deadLetterPolicy(this.deadLetterPolicy); diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java index 37355d80..865a0c6f 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java @@ -96,9 +96,7 @@ public class DefaultReactivePulsarSenderFactory implements ReactivePulsarSend @Nullable List> customizers) { Objects.requireNonNull(schema, "Schema must be specified"); String resolvedTopic = this.topicResolver - .resolveTopic(topic, () -> getReactiveMessageSenderSpec().getTopicName()) - .orElseThrow(() -> new IllegalArgumentException( - "Topic must be specified when no default topic is configured")); + .resolveTopic(topic, () -> getReactiveMessageSenderSpec().getTopicName()).orElseThrow(); this.logger.trace(() -> "Creating reactive message sender for '%s' topic".formatted(resolvedTopic)); ReactiveMessageSenderBuilder sender = this.reactivePulsarClient.messageSender(schema); diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java index 1bd5aaf9..2a8b1f9e 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java @@ -41,7 +41,7 @@ public interface ReactivePulsarOperations { * @param message the message to send * @return the id assigned by the broker to the published message */ - Mono send(T message); + Mono send(@Nullable T message); /** * Sends a message to the specified topic in a reactive manner. default topic @@ -50,7 +50,7 @@ public interface ReactivePulsarOperations { * resolution * @return the id assigned by the broker to the published message */ - Mono send(T message, @Nullable Schema schema); + Mono send(@Nullable T message, @Nullable Schema schema); /** * Sends a message to the specified topic in a reactive manner. @@ -59,7 +59,7 @@ public interface ReactivePulsarOperations { * @param message the message to send * @return the id assigned by the broker to the published message */ - Mono send(@Nullable String topic, T message); + Mono send(@Nullable String topic, @Nullable T message); /** * Sends a message to the specified topic in a reactive manner. @@ -70,7 +70,7 @@ public interface ReactivePulsarOperations { * resolution * @return the id assigned by the broker to the published message */ - Mono send(@Nullable String topic, T message, @Nullable Schema schema); + Mono send(@Nullable String topic, @Nullable T message, @Nullable Schema schema); /** * Sends multiple messages to the default topic in a reactive manner. @@ -119,7 +119,7 @@ public interface ReactivePulsarOperations { * @param message the payload of the message * @return the builder to configure and send the message */ - SendOneMessageBuilder newMessage(T message); + SendOneMessageBuilder newMessage(@Nullable T message); /** * Create a {@link SendManyMessageBuilder builder} for configuring and sending diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java index 0c1b0ee9..c212e733 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java @@ -16,8 +16,6 @@ package org.springframework.pulsar.reactive.core; -import java.util.Optional; - import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.reactive.client.api.MessageSendResult; @@ -77,22 +75,22 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { } @Override - public Mono send(T message) { + public Mono send(@Nullable T message) { return send(null, message); } @Override - public Mono send(T message, @Nullable Schema schema) { + public Mono send(@Nullable T message, @Nullable Schema schema) { return doSend(null, message, schema, null, null); } @Override - public Mono send(@Nullable String topic, T message) { + public Mono send(@Nullable String topic, @Nullable T message) { return doSend(topic, message, null, null, null); } @Override - public Mono send(@Nullable String topic, T message, @Nullable Schema schema) { + public Mono send(@Nullable String topic, @Nullable T message, @Nullable Schema schema) { return doSend(topic, message, schema, null, null); } @@ -118,7 +116,7 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { } @Override - public SendOneMessageBuilder newMessage(T message) { + public SendOneMessageBuilder newMessage(@Nullable T message) { return new SendOneMessageBuilderImpl<>(this, message); } @@ -127,10 +125,10 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { return new SendManyMessageBuilderImpl<>(this, messages); } - private Mono doSend(@Nullable String topic, T message, @Nullable Schema schema, + private Mono doSend(@Nullable String topic, @Nullable T message, @Nullable Schema schema, @Nullable MessageSpecBuilderCustomizer messageSpecBuilderCustomizer, @Nullable ReactiveMessageSenderBuilderCustomizer customizer) { - String topicName = resolveTopic(topic, message.getClass()); + String topicName = resolveTopic(topic, message); this.logger.trace(() -> "Sending reactive msg to '%s' topic".formatted(topicName)); ReactiveMessageSender sender = createMessageSender(topicName, message, schema, customizer); // @formatter:off @@ -145,7 +143,7 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { return messages.switchOnFirst((firstSignal, messageFlux) -> { MessageSpec firstMessage = firstSignal.get(); if (firstMessage != null && firstSignal.isOnNext()) { - String topicName = resolveTopic(topic, firstMessage.getValue().getClass()); + String topicName = resolveTopic(topic, firstMessage.getValue()); ReactiveMessageSender sender = createMessageSender(topicName, firstMessage.getValue(), schema, customizer); return messageFlux.as(sender::sendMany).doOnError( @@ -157,21 +155,13 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { }); } - private String resolveTopic(@Nullable String topic, @Nullable Class messageType) { + private String resolveTopic(@Nullable String topic, @Nullable Object message) { String defaultTopic = this.reactiveMessageSenderFactory.getReactiveMessageSenderSpec().getTopicName(); - Optional resolvedTopic; - if (messageType == null) { - resolvedTopic = this.topicResolver.resolveTopic(topic, () -> defaultTopic); - } - else { - resolvedTopic = this.topicResolver.resolveTopic(topic, messageType, () -> defaultTopic); - } - return resolvedTopic.orElseThrow( - () -> new IllegalArgumentException("Topic must be specified when no default topic is configured")); + return this.topicResolver.resolveTopic(topic, message, () -> defaultTopic).orElseThrow(); } private static MessageSpec getMessageSpec( - @Nullable MessageSpecBuilderCustomizer messageSpecBuilderCustomizer, T message) { + @Nullable MessageSpecBuilderCustomizer messageSpecBuilderCustomizer, @Nullable T message) { MessageSpecBuilder messageSpecBuilder = MessageSpec.builder(message); if (messageSpecBuilderCustomizer != null) { messageSpecBuilderCustomizer.customize(messageSpecBuilder); @@ -179,12 +169,9 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { return messageSpecBuilder.build(); } - private ReactiveMessageSender createMessageSender(@Nullable String topic, T message, @Nullable Schema schema, - @Nullable ReactiveMessageSenderBuilderCustomizer customizer) { - Schema resolvedSchema = schema == null ? this.schemaResolver.getSchema(message) : schema; - if (resolvedSchema == null) { - throw new IllegalArgumentException("Couldn't resolve a schema for the message"); - } + private ReactiveMessageSender createMessageSender(@Nullable String topic, @Nullable T message, + @Nullable Schema schema, @Nullable ReactiveMessageSenderBuilderCustomizer customizer) { + Schema resolvedSchema = schema == null ? this.schemaResolver.resolveSchema(message).orElseThrow() : schema; return this.reactiveMessageSenderFactory.createSender(resolvedSchema, topic, customizer); } @@ -228,12 +215,13 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { private static final class SendOneMessageBuilderImpl extends SendMessageBuilderImpl, T> implements SendOneMessageBuilder { + @Nullable private final T message; @Nullable private MessageSpecBuilderCustomizer messageCustomizer; - SendOneMessageBuilderImpl(ReactivePulsarTemplate template, T message) { + SendOneMessageBuilderImpl(ReactivePulsarTemplate template, @Nullable T message) { super(template); this.message = message; } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java index 3cc15a50..a2a0c382 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java @@ -17,38 +17,36 @@ package org.springframework.pulsar.reactive.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.params.provider.Arguments.arguments; import java.time.Duration; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; import java.util.UUID; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; -import java.util.function.BiConsumer; +import java.util.function.Consumer; import java.util.stream.Stream; -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.PulsarClient; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.reactive.client.api.MessageSpec; import org.apache.pulsar.reactive.client.api.MutableReactiveMessageSenderSpec; import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +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.junit.jupiter.params.provider.ValueSource; import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.DefaultTopicResolver; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.util.function.ThrowingConsumer; import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; /** * Tests for {@link ReactivePulsarTemplate}. @@ -58,275 +56,198 @@ import reactor.core.publisher.Mono; */ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { - @ParameterizedTest - @ValueSource(booleans = { true, false }) - void sendManyWithSpecificSchema(boolean useSimpleApi) throws Exception { - String topic = "rptt-sendMessagesWithSpecificSchema-" + useSimpleApi + "-topic"; - String sub = "rptt-sendMessagesWithSpecificSchema-" + useSimpleApi + "-sub"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic).subscriptionName(sub) - .subscribe()) { - MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); - senderSpec.setTopicName(topic); - ReactivePulsarSenderFactory producerFactory = new DefaultReactivePulsarSenderFactory<>(client, - senderSpec, null); - ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory); + private PulsarClient client; - List foos = new ArrayList<>(); - for (int i = 0; i < 10; i++) { - foos.add(new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID())); - } - - if (useSimpleApi) { - pulsarTemplate.send(Flux.fromIterable(foos).map(MessageSpec::of), Schema.JSON(Foo.class)) - .subscribe(); - } - else { - pulsarTemplate.newMessages(Flux.fromIterable(foos).map(MessageSpec::of)) - .withSchema(Schema.JSON(Foo.class)).send().subscribe(); - } - - for (int i = 0; i < 10; i++) { - assertThat(consumer.receiveAsync().thenApply(Message::getValue)) - .succeedsWithin(Duration.ofSeconds(3)).isEqualTo(foos.get(i)); - } - } - } + @BeforeEach + void setup() throws PulsarClientException { + client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); } - @ParameterizedTest - @ValueSource(booleans = { true, false }) - void sendManyWithInferredSchema(boolean useSimpleApi) throws Exception { - String topic = "rptt-sendMessagesWithInferredSchema-" + useSimpleApi + "-topic"; - String sub = "rptt-sendMessagesWithInferredSchema-" + useSimpleApi + "-sub"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic).subscriptionName(sub) - .subscribe()) { - MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); - senderSpec.setTopicName(topic); - ReactivePulsarSenderFactory producerFactory = new DefaultReactivePulsarSenderFactory<>(client, - senderSpec, null); - // Custom schema resolver allows not specifying the schema when sending - DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); - schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); - ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory, - schemaResolver, new DefaultTopicResolver()); - - List foos = new ArrayList<>(); - for (int i = 0; i < 10; i++) { - foos.add(new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID())); - } - - if (useSimpleApi) { - pulsarTemplate.send(Flux.fromIterable(foos).map(MessageSpec::of)).subscribe(); - } - else { - pulsarTemplate.newMessages(Flux.fromIterable(foos).map(MessageSpec::of)).send().subscribe(); - } - - // TODO figure out if expected to not be ordered when schema not set on - // template - List foos2 = new ArrayList<>(); - for (int i = 0; i < 10; i++) { - CompletableFuture> receiveFuture = consumer.receiveAsync(); - assertThat(receiveFuture).succeedsWithin(Duration.ofSeconds(3)); - foos2.add(receiveFuture.get().getValue()); - } - assertThat(foos).containsExactlyInAnyOrderElementsOf(foos2); - } - } - } - - @ParameterizedTest(name = "{0}") - @MethodSource("sendManyWithInferredTopicProvider") - void sendManyWithInferredTopic(String testName, - BiConsumer, ReactivePulsarTemplate> sendHandler) throws Exception { - String topic = "rptt-" + testName + "-topic"; - String sub = "rptt-" + testName + "-sub"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.STRING).topic(topic).subscriptionName(sub) - .subscribe()) { - MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); - senderSpec.setTopicName(topic + "-fake"); - ReactivePulsarSenderFactory producerFactory = new DefaultReactivePulsarSenderFactory<>(client, - senderSpec, null); - - // Topic mappings allows not specifying the topic when sending (nor having - // default on sender) - DefaultTopicResolver topicResolver = new DefaultTopicResolver(); - topicResolver.addCustomTopicMapping(String.class, topic); - ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory, - new DefaultSchemaResolver(), topicResolver); - - String theSingleFoo = "Foo-" + UUID.randomUUID(); - - sendHandler.accept(Collections.singletonList(theSingleFoo), pulsarTemplate); - - assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) - .isEqualTo(theSingleFoo); - - } - } - } - - static Stream sendManyWithInferredTopicProvider() { - return Stream.of( - arguments("simpleApiNoSchema", - (BiConsumer, ReactivePulsarTemplate>) (data, template) -> template - .send(Flux.fromIterable(data).map(MessageSpec::of)).subscribe()), - arguments("simpleApiWithSchema", - (BiConsumer, ReactivePulsarTemplate>) (data, template) -> template - .send(Flux.fromIterable(data).map(MessageSpec::of), Schema.STRING).subscribe()), - arguments("fluentApiNoSchema", - (BiConsumer, ReactivePulsarTemplate>) (data, template) -> template - .newMessages(Flux.fromIterable(data).map(MessageSpec::of)).send().subscribe()), - arguments("fluentApiWithSchema", - (BiConsumer, ReactivePulsarTemplate>) (data, template) -> template - .newMessages(Flux.fromIterable(data).map(MessageSpec::of)).withSchema(Schema.STRING) - .send().subscribe())); + @AfterEach + void tearDown() throws PulsarClientException { + // 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()); + client.close(); } @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"; - MessageSpecBuilderCustomizer messageCustomizer = null; - if (testArgs.messageCustomizer) { - messageCustomizer = (mb) -> mb.key("foo-key"); - } - ReactiveMessageSenderBuilderCustomizer senderCustomizer = null; - if (testArgs.senderCustomizer) { - senderCustomizer = (sb) -> sb.producerName("foo-sender"); - } - 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.explicitTopic) { - senderSpec.setTopicName(topic); - } - ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, - senderSpec, null); - ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory); - Mono sendResponse; - if (testArgs.simpleApi) { - if (testArgs.explicitSchema && testArgs.explicitTopic) { - sendResponse = pulsarTemplate.send(topic, msgPayload, Schema.STRING); - } - else if (testArgs.explicitSchema) { - sendResponse = pulsarTemplate.send(msgPayload, Schema.STRING); - } - else if (testArgs.explicitTopic) { - sendResponse = pulsarTemplate.send(topic, msgPayload); - } - else { - sendResponse = pulsarTemplate.send(msgPayload); - } - } - else { - ReactivePulsarTemplate.SendOneMessageBuilder messageBuilder = pulsarTemplate - .newMessage(msgPayload); - if (testArgs.explicitTopic) { - messageBuilder = messageBuilder.withTopic(topic); - } - if (testArgs.explicitSchema) { - messageBuilder = messageBuilder.withSchema(Schema.STRING); - } - if (messageCustomizer != null) { - messageBuilder = messageBuilder.withMessageCustomizer(messageCustomizer); - } - 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 (senderCustomizer != null) { - assertThat(msg.getProducerName()).isEqualTo("foo-sender"); - } - // 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()); - } - } + void sendMessageTest(String testName, Consumer> sendFunction, + Boolean withDefaultTopic, String expectedValue) throws Exception { + sendAndConsume(sendFunction, testName, Schema.STRING, expectedValue, withDefaultTopic); } - private static Stream sendMessageTestProvider() { - return Stream.of(arguments("simpleReactiveSend", SendTestArgs.simple()), - arguments("simpleReactiveSendWithTopic", SendTestArgs.simple().topic()), - arguments("simpleReactiveSendWithSchema", SendTestArgs.simple().schema()), - arguments("simpleReactiveSendWithTopicAndSchema", SendTestArgs.simple().topic().schema()), - arguments("fluentReactiveSend", SendTestArgs.fluent()), - arguments("fluentReactiveSendWithSchema", SendTestArgs.fluent().schema()), - arguments("fluentReactiveSendWithTopic", SendTestArgs.fluent().topic()), - arguments("fluentReactiveSendWithMessageCustomizer", SendTestArgs.fluent().messageCustomizer()), - arguments("fluentReactiveSendWithSenderCustomizer", SendTestArgs.fluent().senderCustomizer()), - arguments("fluentReactiveSendWithTopicAndSchema", SendTestArgs.fluent().topic().schema()), - arguments("fluentReactiveSendWithTopicAndSchemaAndCustomizers", - SendTestArgs.fluent().topic().schema().messageCustomizer().senderCustomizer())); + static Stream sendMessageTestProvider() { + String message = "test-message"; + Flux> messagePublisher = Flux.just(MessageSpec.of(message)); + return Stream.of( + arguments("simpleSendWithDefaultTopic", + (Consumer>) (template) -> template.send(message).subscribe(), + true, message), + arguments("simpleSendWithTopic", + (Consumer>) (template) -> template + .send("simpleSendWithTopic", message).subscribe(), + false, message), + arguments("simpleSendWithDefaultTopicAndSchema", + (Consumer>) (template) -> template.send(message, Schema.STRING) + .subscribe(), + true, message), + arguments("simpleSendWithTopicAndSchema", + (Consumer>) (template) -> template + .send("simpleSendWithTopicAndSchema", message, Schema.STRING).subscribe(), + false, message), + arguments("simpleSendNullWithTopicAndSchema", + (Consumer>) (template) -> template + .send("simpleSendNullWithTopicAndSchema", (String) null, Schema.STRING).subscribe(), + false, null), + + arguments("simplePublisherSendWithDefaultTopic", + (Consumer>) (template) -> template.send(messagePublisher) + .subscribe(), + true, message), + arguments("simplePublisherSendWithTopic", + (Consumer>) (template) -> template + .send("simplePublisherSendWithTopic", messagePublisher).subscribe(), + false, message), + arguments("simplePublisherSendWithDefaultTopicAndSchema", + (Consumer>) (template) -> template + .send(messagePublisher, Schema.STRING).subscribe(), + true, message), + arguments("simplePublisherSendWithTopicAndSchema", + (Consumer>) (template) -> template + .send("simplePublisherSendWithTopicAndSchema", messagePublisher, Schema.STRING) + .subscribe(), + false, message), + + arguments("fluentSendWithDefaultTopic", + (Consumer>) (template) -> template.newMessage(message).send() + .subscribe(), + true, message), + arguments("fluentSendWithTopic", + (Consumer>) (template) -> template.newMessage(message) + .withTopic("fluentSendWithTopic").send().subscribe(), + false, message), + arguments("fluentSendWithDefaultTopicAndSchema", + (Consumer>) (template) -> template.newMessage(message) + .withSchema(Schema.STRING).send().subscribe(), + true, message), + arguments("fluentSendNullWithTopicAndSchema", + (Consumer>) (template) -> template.newMessage(null) + .withSchema(Schema.STRING).withTopic("fluentSendNullWithTopicAndSchema").send() + .subscribe(), + false, null), + arguments("fluentPublisherSend", (Consumer>) (template) -> template + .newMessages(messagePublisher).send().subscribe(), true, message)); } - static final class SendTestArgs { + @Test + void sendMessageWithMessageCustomizer() throws Exception { + Consumer> sendFunction = (template) -> template.newMessage("test-message") + .withMessageCustomizer((mb) -> mb.key("test-key")).send().subscribe(); + Message msg = sendAndConsume(sendFunction, "sendMessageWithMessageCustomizer", Schema.STRING, + "test-message", true); + assertThat(msg.getKey()).isEqualTo("test-key"); + } - private boolean simpleApi; + @Test + void sendMessageWithSenderCustomizer() throws Exception { + Consumer> sendFunction = (template) -> template.newMessage("test-message") + .withSenderCustomizer((sb) -> sb.producerName("test-producer")).send().subscribe(); + Message msg = sendAndConsume(sendFunction, "sendMessageWithSenderCustomizer", Schema.STRING, + "test-message", true); + assertThat(msg.getProducerName()).isEqualTo("test-producer"); + } - private boolean explicitTopic; + @Test + void sendMessageWithCustomTopicMapping() throws Exception { + String topic = "sendMessageWithCustomTopicMapping"; - private boolean explicitSchema; + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, + new MutableReactiveMessageSenderSpec(), null); - private boolean messageCustomizer; + DefaultTopicResolver topicResolver = new DefaultTopicResolver(); + topicResolver.addCustomTopicMapping(String.class, topic); + ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory, + new DefaultSchemaResolver(), topicResolver); - private boolean senderCustomizer; + Consumer> sendFunction = (template) -> template.send("test-message").subscribe(); + sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.STRING, "test-message"); + } - private SendTestArgs(boolean simpleApi) { - this.simpleApi = simpleApi; + @Test + void sendMessageWithCustomSchemaMapping() throws Exception { + String topic = "sendMessageWithCustomSchemaMapping"; + + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, + new MutableReactiveMessageSenderSpec(), null); + + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); + schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); + ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory, schemaResolver, + new DefaultTopicResolver()); + + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + Consumer> sendFunction = (template) -> template.send(topic, foo).subscribe(); + sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.JSON(Foo.class), foo); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("sendMessageFailedTestProvider") + void sendMessageFailed(String testName, ThrowingConsumer> sendFunction) { + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, + new MutableReactiveMessageSenderSpec(), null); + ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory); + assertThatIllegalArgumentException().isThrownBy(() -> sendFunction.accept(pulsarTemplate)); + } + + static Stream sendMessageFailedTestProvider() { + String message = "test-message"; + return Stream.of( + arguments("sendWithoutTopic", + (ThrowingConsumer>) (template) -> template.send(message)), + arguments("sendNullWithoutSchema", + (ThrowingConsumer>) (template) -> template + .send("sendNullWithoutSchema", (String) null))); + } + + @Test + void sendNullWithDefaultTopicFails() { + MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec(); + spec.setTopicName("sendNullWithDefaultTopicFails"); + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, spec, + null); + ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory); + assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send((String) null, Schema.STRING)); + } + + private Message sendAndConsume(Consumer> sendFunction, String topic, + Schema schema, T expectedValue, Boolean withDefaultTopic) throws Exception { + MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); + if (withDefaultTopic) { + senderSpec.setTopicName(topic); } + ReactivePulsarSenderFactory senderFactory = new DefaultReactivePulsarSenderFactory<>(client, senderSpec, + null); - static SendTestArgs simple() { - return new SendTestArgs(true); + ReactivePulsarTemplate pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory); + + return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue); + } + + private Message sendAndConsume(ReactivePulsarTemplate template, + Consumer> sendFunction, String topic, Schema schema, T expectedValue) + throws Exception { + try (org.apache.pulsar.client.api.Consumer consumer = client.newConsumer(schema).topic(topic) + .subscriptionName(topic + "-sub").subscribe()) { + sendFunction.accept(template); + + Message msg = consumer.receive(3, TimeUnit.SECONDS); + assertThat(msg).isNotNull(); + assertThat(msg.getValue()).isEqualTo(expectedValue); + return msg; } - - static SendTestArgs fluent() { - return new SendTestArgs(false); - } - - SendTestArgs topic() { - this.explicitTopic = true; - return this; - } - - SendTestArgs schema() { - this.explicitSchema = true; - return this; - } - - SendTestArgs messageCustomizer() { - this.messageCustomizer = true; - return this; - } - - SendTestArgs senderCustomizer() { - this.senderCustomizer = true; - return this; - } - } record Foo(String foo, String bar) { diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java index 879bc26f..e92c7b85 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java @@ -162,7 +162,7 @@ public class PulsarMessageChannelBinder extends } } // TODO if schema == null then default lookup bean Schema w/ name == binding - return this.schemaResolver.getSchema(schemaType, resolvableType); + return this.schemaResolver.resolveSchema(schemaType, resolvableType).get().orElse(null); } @Override diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java index e0eda44f..12c567d3 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java @@ -17,9 +17,13 @@ package org.springframework.pulsar.spring.cloud.stream.binder; import static org.assertj.core.api.AssertionsForClassTypes.assertThatExceptionOfType; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; +import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.SchemaType; import org.junit.jupiter.api.Nested; @@ -31,6 +35,7 @@ import org.junit.jupiter.params.provider.EnumSource.Mode; import org.springframework.core.ResolvableType; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.Resolved; import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.spring.cloud.stream.binder.provisioning.PulsarTopicProvisioner; @@ -50,21 +55,24 @@ public class PulsarMessageChannelBinderResolveSchemaTests { @ParameterizedTest @EnumSource(mode = Mode.MATCH_NONE, names = "^(AUTO.*|AVRO|JSON|KEY_VALUE|NONE|PROTOBUF.*)$") void primitiveSchemaTypes(SchemaType schemaType) { + doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(schemaType, null); binder.resolveSchema(schemaType, null, null, null); - verify(resolver).getSchema(schemaType, null); + verify(resolver).resolveSchema(schemaType, null); } @ParameterizedTest @EnumSource(mode = Mode.MATCH_ALL, names = "^(JSON|AVRO|PROTOBUF)$") void structSchemaTypes(SchemaType schemaType) { + doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(schemaType), any()); binder.resolveSchema(schemaType, Foo.class, null, null); - verify(resolver).getSchema(schemaType, ResolvableType.forClass(Foo.class)); + verify(resolver).resolveSchema(schemaType, ResolvableType.forClass(Foo.class)); } @Test void keyValueSchemaType() { + doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.KEY_VALUE), any()); binder.resolveSchema(SchemaType.KEY_VALUE, null, Foo.class, Bar.class); - verify(resolver).getSchema(SchemaType.KEY_VALUE, + verify(resolver).resolveSchema(SchemaType.KEY_VALUE, ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class)); } @@ -94,21 +102,24 @@ public class PulsarMessageChannelBinderResolveSchemaTests { @Test void withMesssageType() { + doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.NONE), any()); binder.resolveSchema(SchemaType.NONE, Foo.class, null, null); - verify(resolver).getSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); + verify(resolver).resolveSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); } @Test void withKeyAndValueTypes() { + doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.NONE), any()); binder.resolveSchema(SchemaType.NONE, null, Foo.class, Bar.class); - verify(resolver).getSchema(SchemaType.NONE, + verify(resolver).resolveSchema(SchemaType.NONE, ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class)); } @Test void withMessageTypeAndKeyAndValueTypes() { + doReturn(Resolved.of(Schema.STRING)).when(resolver).resolveSchema(eq(SchemaType.NONE), any()); binder.resolveSchema(SchemaType.NONE, Foo.class, String.class, Bar.class); - verify(resolver).getSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); + verify(resolver).resolveSchema(SchemaType.NONE, ResolvableType.forClass(Foo.class)); } @Test diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index eb7e4c10..898ac671 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -26,7 +26,6 @@ import org.apache.pulsar.client.api.DeadLetterPolicy; import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.Messages; import org.apache.pulsar.client.api.RedeliveryBackoff; -import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.core.MethodParameter; @@ -143,10 +142,8 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver(); SchemaType schemaType = pulsarContainerProperties.getSchemaType(); ResolvableType messageType = resolvableType(messageParameter); - Schema schema = schemaResolver.getSchema(schemaType, messageType); - if (schema != null) { - pulsarContainerProperties.setSchema(schema); - } + schemaResolver.resolveSchema(schemaType, messageType).ifResolved(pulsarContainerProperties::setSchema); + // Make sure the schemaType is updated to match the current schema if (pulsarContainerProperties.getSchema() != null) { SchemaType type = pulsarContainerProperties.getSchema().getSchemaInfo().getType(); @@ -159,7 +156,7 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo || StringUtils.hasText(pulsarContainerProperties.getTopicsPattern()); if (!hasTopicInfo) { topicResolver.resolveTopic(null, messageType.getRawClass(), () -> null) - .ifPresent((topic) -> pulsarContainerProperties.setTopics(new String[] { topic })); + .ifResolved((topic) -> pulsarContainerProperties.setTopics(new String[] { topic })); } container.setNegativeAckRedeliveryBackoff(this.negativeAckRedeliveryBackoff); 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 73188986..f3403731 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 @@ -126,8 +126,7 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory protected String resolveTopicName(String userSpecifiedTopic) { String defaultTopic = Objects.toString(getProducerConfig().get("topicName"), null); - return this.topicResolver.resolveTopic(userSpecifiedTopic, () -> defaultTopic).orElseThrow( - () -> new IllegalArgumentException("Topic must be specified when no default topic is configured")); + return this.topicResolver.resolveTopic(userSpecifiedTopic, () -> defaultTopic).orElseThrow(); } @Override diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java index 752030ba..f4ec6b4b 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java @@ -123,16 +123,22 @@ public class DefaultSchemaResolver implements SchemaResolver { } @Override - public Schema getSchema(Class messageClass, boolean returnDefault) { + public Resolved> resolveSchema(@Nullable Class messageClass, boolean returnDefault) { + if (messageClass == null) { + return Resolved.failed("Schema must be specified when the message is null"); + } Schema schema = BASE_SCHEMA_MAPPINGS.get(messageClass); if (schema == null) { schema = getCustomSchemaOrMaybeDefault(messageClass, returnDefault); } - return schema != null ? castToType(schema) : null; + if (schema == null) { + return Resolved.failed("Schema not specified and no schema found for " + messageClass); + } + return Resolved.of(castToType(schema)); } @Nullable - protected Schema getCustomSchemaOrMaybeDefault(Class messageClass, boolean returnDefault) { + protected Schema getCustomSchemaOrMaybeDefault(@Nullable Class messageClass, boolean returnDefault) { Schema schema = this.customSchemaMappings.get(messageClass); if (schema == null && returnDefault) { if (messageClass != null) { @@ -150,49 +156,55 @@ public class DefaultSchemaResolver implements SchemaResolver { @Override @SuppressWarnings("unchecked") - public Schema getSchema(SchemaType schemaType, @Nullable ResolvableType messageType) { - Schema schema = switch (schemaType) { - case STRING -> Schema.STRING; - case BOOLEAN -> Schema.BOOL; - case INT8 -> Schema.INT8; - case INT16 -> Schema.INT16; - case INT32 -> Schema.INT32; - case INT64 -> Schema.INT64; - case FLOAT -> Schema.FLOAT; - case DOUBLE -> Schema.DOUBLE; - case DATE -> Schema.DATE; - case TIME -> Schema.TIME; - case TIMESTAMP -> Schema.TIMESTAMP; - case BYTES -> Schema.BYTES; - case INSTANT -> Schema.INSTANT; - case LOCAL_DATE -> Schema.LOCAL_DATE; - case LOCAL_TIME -> Schema.LOCAL_TIME; - case LOCAL_DATE_TIME -> Schema.LOCAL_DATE_TIME; - case JSON -> JSONSchema.of(requireNonNullMessageType(schemaType, messageType)); - case AVRO -> AvroSchema.of(requireNonNullMessageType(schemaType, messageType)); - case PROTOBUF -> { - Class messageClass = requireNonNullMessageType(schemaType, messageType); - yield ProtobufSchema.of((Class) messageClass); - } - case KEY_VALUE -> { - requireNonNullMessageType(schemaType, messageType); - yield getMessageKeyValueSchema(messageType); - } - case NONE -> { - if (messageType == null) { - yield Schema.BYTES; + public Resolved> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType) { + try { + Schema schema = switch (schemaType) { + case STRING -> Schema.STRING; + case BOOLEAN -> Schema.BOOL; + case INT8 -> Schema.INT8; + case INT16 -> Schema.INT16; + case INT32 -> Schema.INT32; + case INT64 -> Schema.INT64; + case FLOAT -> Schema.FLOAT; + case DOUBLE -> Schema.DOUBLE; + case DATE -> Schema.DATE; + case TIME -> Schema.TIME; + case TIMESTAMP -> Schema.TIMESTAMP; + case BYTES -> Schema.BYTES; + case INSTANT -> Schema.INSTANT; + case LOCAL_DATE -> Schema.LOCAL_DATE; + case LOCAL_TIME -> Schema.LOCAL_TIME; + case LOCAL_DATE_TIME -> Schema.LOCAL_DATE_TIME; + case JSON -> JSONSchema.of(requireNonNullMessageType(schemaType, messageType)); + case AVRO -> AvroSchema.of(requireNonNullMessageType(schemaType, messageType)); + case PROTOBUF -> { + Class messageClass = requireNonNullMessageType(schemaType, messageType); + yield ProtobufSchema.of((Class) messageClass); } - if (KeyValue.class.isAssignableFrom(messageType.getRawClass())) { + case KEY_VALUE -> { + requireNonNullMessageType(schemaType, messageType); yield getMessageKeyValueSchema(messageType); } - yield getSchema(messageType.getRawClass(), false); - } - default -> throw new IllegalArgumentException("Unsupported schema type: " + schemaType.name()); - }; - return schema != null ? castToType(schema) : null; + case NONE -> { + if (messageType == null || messageType.getRawClass() == null) { + yield Schema.BYTES; + } + if (KeyValue.class.isAssignableFrom(messageType.getRawClass())) { + yield getMessageKeyValueSchema(messageType); + } + yield resolveSchema(messageType.getRawClass(), false).orElseThrow(); + } + default -> throw new IllegalArgumentException("Unsupported schema type: " + schemaType.name()); + }; + return Resolved.of(castToType(schema)); + } + catch (RuntimeException e) { + return Resolved.failed(e); + } } - private Class requireNonNullMessageType(SchemaType schemaType, ResolvableType messageType) { + @Nullable + private Class requireNonNullMessageType(SchemaType schemaType, @Nullable ResolvableType messageType) { return Objects.requireNonNull(messageType, "messageType must be specified for " + schemaType.name()) .getRawClass(); } @@ -200,8 +212,8 @@ public class DefaultSchemaResolver implements SchemaResolver { private Schema getMessageKeyValueSchema(ResolvableType messageType) { Class keyClass = messageType.resolveGeneric(0); Class valueClass = messageType.resolveGeneric(1); - Schema> keySchema = this.getSchema(keyClass); - Schema> valueSchema = this.getSchema(valueClass); + Schema keySchema = this.resolveSchema(keyClass).orElseThrow(); + Schema valueSchema = this.resolveSchema(valueClass).orElseThrow(); return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java index e60acc78..594d15a1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java @@ -19,7 +19,6 @@ package org.springframework.pulsar.core; import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; -import java.util.Optional; import java.util.function.Supplier; import org.springframework.lang.Nullable; @@ -70,28 +69,40 @@ public class DefaultTopicResolver implements TopicResolver { } @Override - public Optional resolveTopic(@Nullable String userSpecifiedTopic, Supplier defaultTopicSupplier) { - return doResolveTopic(userSpecifiedTopic, null, defaultTopicSupplier); + public Resolved resolveTopic(@Nullable String userSpecifiedTopic, Supplier defaultTopicSupplier) { + if (StringUtils.hasText(userSpecifiedTopic)) { + return Resolved.of(userSpecifiedTopic); + } + String defaultTopic = defaultTopicSupplier.get(); + if (defaultTopic == null) { + return Resolved.failed("Topic must be specified when no default topic is configured"); + } + return Resolved.of(defaultTopic); } @Override - public Optional resolveTopic(@Nullable String userSpecifiedTopic, T message, + public Resolved resolveTopic(@Nullable String userSpecifiedTopic, @Nullable T message, Supplier defaultTopicSupplier) { - return doResolveTopic(userSpecifiedTopic, message.getClass(), defaultTopicSupplier); + return doResolveTopic(userSpecifiedTopic, message != null ? message.getClass() : null, defaultTopicSupplier); } @Override - public Optional resolveTopic(@Nullable String userSpecifiedTopic, Class messageType, + public Resolved resolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class messageType, Supplier defaultTopicSupplier) { return doResolveTopic(userSpecifiedTopic, messageType, defaultTopicSupplier); } - private Optional doResolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class messageType, + protected Resolved doResolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class messageType, Supplier defaultTopicSupplier) { if (StringUtils.hasText(userSpecifiedTopic)) { - return Optional.of(userSpecifiedTopic); + return Resolved.of(userSpecifiedTopic); } - return Optional.ofNullable(this.customTopicMappings.getOrDefault(messageType, defaultTopicSupplier.get())); + if (messageType == null) { + return Resolved.failed("Topic must be specified when the message is null"); + } + String topic = this.customTopicMappings.getOrDefault(messageType, defaultTopicSupplier.get()); + return topic == null ? Resolved.failed("Topic must be specified when no default topic is configured") + : Resolved.of(topic); } } 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 fa4247fe..636ef68e 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 @@ -40,7 +40,7 @@ public interface PulsarOperations { * @return the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - MessageId send(T message) throws PulsarClientException; + MessageId send(@Nullable T message) throws PulsarClientException; /** * Sends a message to the default topic in a blocking manner. @@ -50,7 +50,7 @@ public interface PulsarOperations { * @return the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - MessageId send(T message, @Nullable Schema schema) throws PulsarClientException; + MessageId send(@Nullable T message, @Nullable Schema schema) throws PulsarClientException; /** * Sends a message to the specified topic in a blocking manner. @@ -60,7 +60,7 @@ public interface PulsarOperations { * @return the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - MessageId send(@Nullable String topic, T message) throws PulsarClientException; + MessageId send(@Nullable String topic, @Nullable T message) throws PulsarClientException; /** * Sends a message to the specified topic in a blocking manner. @@ -72,7 +72,8 @@ public interface PulsarOperations { * @return the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - MessageId send(@Nullable String topic, T message, @Nullable Schema schema) throws PulsarClientException; + MessageId send(@Nullable String topic, @Nullable T message, @Nullable Schema schema) + throws PulsarClientException; /** * Sends a message to the default topic in a non-blocking manner. @@ -80,7 +81,7 @@ public interface PulsarOperations { * @return a future that holds the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - CompletableFuture sendAsync(T message) throws PulsarClientException; + CompletableFuture sendAsync(@Nullable T message) throws PulsarClientException; /** * Sends a message to the default topic in a non-blocking manner. @@ -90,7 +91,8 @@ public interface PulsarOperations { * @return a future that holds the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - CompletableFuture sendAsync(T message, @Nullable Schema schema) throws PulsarClientException; + CompletableFuture sendAsync(@Nullable T message, @Nullable Schema schema) + throws PulsarClientException; /** * Sends a message to the specified topic in a non-blocking manner. @@ -100,7 +102,7 @@ public interface PulsarOperations { * @return a future that holds the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - CompletableFuture sendAsync(@Nullable String topic, T message) throws PulsarClientException; + CompletableFuture sendAsync(@Nullable String topic, @Nullable T message) throws PulsarClientException; /** * Sends a message to the specified topic in a non-blocking manner. @@ -112,7 +114,7 @@ public interface PulsarOperations { * @return a future that holds the id assigned by the broker to the published message * @throws PulsarClientException if an error occurs */ - CompletableFuture sendAsync(@Nullable String topic, T message, @Nullable Schema schema) + CompletableFuture sendAsync(@Nullable String topic, @Nullable T message, @Nullable Schema schema) throws PulsarClientException; /** @@ -120,7 +122,7 @@ public interface PulsarOperations { * @param message the payload of the message * @return the builder to configure and send the message */ - SendMessageBuilder newMessage(T message); + SendMessageBuilder newMessage(@Nullable T message); /** * Builder that can be used to configure and send a message. Provides more 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 babe1509..82289d93 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 @@ -114,48 +114,51 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { } @Override - public MessageId send(T message) throws PulsarClientException { + public MessageId send(@Nullable T message) throws PulsarClientException { return doSend(null, message, null, null, null, null); } @Override - public MessageId send(T message, @Nullable Schema schema) throws PulsarClientException { + public MessageId send(@Nullable T message, @Nullable Schema schema) throws PulsarClientException { return doSend(null, message, schema, null, null, null); } @Override - public MessageId send(@Nullable String topic, T message) throws PulsarClientException { + public MessageId send(@Nullable String topic, @Nullable T message) throws PulsarClientException { return doSend(topic, message, null, null, null, null); } @Override - public MessageId send(@Nullable String topic, T message, @Nullable Schema schema) throws PulsarClientException { + public MessageId send(@Nullable String topic, @Nullable T message, @Nullable Schema schema) + throws PulsarClientException { return doSend(topic, message, schema, null, null, null); } @Override - public CompletableFuture sendAsync(T message) throws PulsarClientException { + public CompletableFuture sendAsync(@Nullable T message) throws PulsarClientException { return doSendAsync(null, message, null, null, null, null); } @Override - public CompletableFuture sendAsync(T message, @Nullable Schema schema) throws PulsarClientException { + public CompletableFuture sendAsync(@Nullable T message, @Nullable Schema schema) + throws PulsarClientException { return doSendAsync(null, message, schema, null, null, null); } @Override - public CompletableFuture sendAsync(@Nullable String topic, T message) throws PulsarClientException { + public CompletableFuture sendAsync(@Nullable String topic, @Nullable T message) + throws PulsarClientException { return doSendAsync(topic, message, null, null, null, null); } @Override - public CompletableFuture sendAsync(@Nullable String topic, T message, @Nullable Schema schema) - throws PulsarClientException { + public CompletableFuture sendAsync(@Nullable String topic, @Nullable T message, + @Nullable Schema schema) throws PulsarClientException { return doSendAsync(topic, message, schema, null, null, null); } @Override - public SendMessageBuilder newMessage(T message) { + public SendMessageBuilder newMessage(@Nullable T message) { return new SendMessageBuilderImpl<>(this, message); } @@ -164,7 +167,7 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { this.beanName = beanName; } - private MessageId doSend(@Nullable String topic, T message, @Nullable Schema schema, + private MessageId doSend(@Nullable String topic, @Nullable T message, @Nullable Schema schema, @Nullable Collection encryptionKeys, @Nullable TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, @Nullable ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { @@ -177,13 +180,12 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { } } - private CompletableFuture doSendAsync(@Nullable String topic, T message, @Nullable Schema schema, - @Nullable Collection encryptionKeys, + private CompletableFuture doSendAsync(@Nullable String topic, @Nullable T message, + @Nullable Schema schema, @Nullable Collection encryptionKeys, @Nullable TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, @Nullable ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { String defaultTopic = Objects.toString(this.producerFactory.getProducerConfig().get("topicName"), null); - String topicName = this.topicResolver.resolveTopic(topic, message, () -> defaultTopic).orElseThrow( - () -> new IllegalArgumentException("Topic must be specified when no default topic is configured")); + String topicName = this.topicResolver.resolveTopic(topic, message, () -> defaultTopic).orElseThrow(); this.logger.trace(() -> "Sending msg to '%s' topic".formatted(topicName)); PulsarMessageSenderContext senderContext = PulsarMessageSenderContext.newContext(topicName, this.beanName); @@ -192,13 +194,19 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { observation.start(); Producer producer = prepareProducerForSend(topicName, message, schema, encryptionKeys, producerCustomizer); - TypedMessageBuilder messageBuilder = producer.newMessage().value(message); - if (typedMessageBuilderCustomizer != null) { - typedMessageBuilderCustomizer.customize(messageBuilder); + TypedMessageBuilder messageBuilder; + try { + messageBuilder = producer.newMessage().value(message); + if (typedMessageBuilderCustomizer != null) { + typedMessageBuilderCustomizer.customize(messageBuilder); + } + // propagate props to message + senderContext.properties().forEach(messageBuilder::property); + } + catch (Exception e) { + ProducerUtils.closeProducerAsync(producer, this.logger); + throw e; } - // propagate props to message - senderContext.properties().forEach(messageBuilder::property); - return messageBuilder.sendAsync().whenComplete((msgId, ex) -> { if (ex == null) { this.logger.trace(() -> "Sent msg to '%s' topic".formatted(topicName)); @@ -227,13 +235,10 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { DefaultPulsarTemplateObservationConvention.INSTANCE, () -> senderContext, this.observationRegistry); } - private Producer prepareProducerForSend(@Nullable String topic, T message, @Nullable Schema schema, + private Producer prepareProducerForSend(@Nullable String topic, @Nullable T message, @Nullable Schema schema, @Nullable Collection encryptionKeys, @Nullable ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { - if (schema == null) { - schema = Objects.requireNonNull(this.schemaResolver.getSchema(message), - "Schema must not be null - expecting at least a default schema"); - } + Schema resolvedSchema = schema == null ? this.schemaResolver.resolveSchema(message).orElseThrow() : schema; List> customizers = new ArrayList<>(); if (!CollectionUtils.isEmpty(this.interceptors)) { customizers.add(builder -> this.interceptors.forEach(builder::intercept)); @@ -241,13 +246,14 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { if (producerCustomizer != null) { customizers.add(producerCustomizer); } - return this.producerFactory.createProducer(schema, topic, encryptionKeys, customizers); + return this.producerFactory.createProducer(resolvedSchema, topic, encryptionKeys, customizers); } public static class SendMessageBuilderImpl implements SendMessageBuilder { private final PulsarTemplate template; + @Nullable private final T message; @Nullable @@ -265,7 +271,7 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { @Nullable private ProducerBuilderCustomizer producerCustomizer; - SendMessageBuilderImpl(PulsarTemplate template, T message) { + SendMessageBuilderImpl(PulsarTemplate template, @Nullable T message) { this.template = template; this.message = message; } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java new file mode 100644 index 00000000..28c611e4 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java @@ -0,0 +1,72 @@ +/* + * Copyright 2023-2023 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 java.util.Optional; +import java.util.function.Consumer; + +import org.springframework.lang.Nullable; + +/** + * A resolved value or an exception if it could not be resolved. + * + * @param the resolved type + * @author Christophe Bornet + */ +public final class Resolved { + + @Nullable + private final T value; + + @Nullable + private final RuntimeException exception; + + private Resolved(@Nullable T value, @Nullable RuntimeException exception) { + this.value = value; + this.exception = exception; + } + + public static Resolved of(T value) { + return new Resolved(value, null); + } + + public static Resolved failed(String reason) { + return new Resolved(null, new IllegalArgumentException(reason)); + } + + public static Resolved failed(RuntimeException e) { + return new Resolved(null, e); + } + + public Optional get() { + return Optional.ofNullable(this.value); + } + + public void ifResolved(Consumer action) { + if (this.value != null) { + action.accept(this.value); + } + } + + public T orElseThrow() { + if (this.value == null && this.exception != null) { + throw this.exception; + } + return this.value; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java index 82b155c3..26596ed8 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java @@ -35,9 +35,8 @@ public interface SchemaResolver { * @param message the message instance * @return the schema to use or {@code null} if no schema could be resolved */ - @Nullable - default Schema getSchema(T message) { - return getSchema(message.getClass()); + default Resolved> resolveSchema(@Nullable T message) { + return resolveSchema(message == null ? null : message.getClass()); } /** @@ -46,9 +45,8 @@ public interface SchemaResolver { * @param messageType the message type * @return the schema to use or {@code null} if no schema could be resolved */ - @Nullable - default Schema getSchema(Class messageType) { - return getSchema(messageType, true); + default Resolved> resolveSchema(@Nullable Class messageType) { + return resolveSchema(messageType, true); } /** @@ -60,8 +58,7 @@ public interface SchemaResolver { * @return the schema to use or the default schema if no schema could be resolved and * {@code returnDefault} is {@code true} - otherwise {@code null} */ - @Nullable - Schema getSchema(Class messageType, boolean returnDefault); + Resolved> resolveSchema(@Nullable Class messageType, boolean returnDefault); /** * Get the schema to use given a schema type and a message type. @@ -70,8 +67,7 @@ public interface SchemaResolver { * @param messageType the message type * @return the schema to use */ - @Nullable - Schema getSchema(SchemaType schemaType, @Nullable ResolvableType messageType); + Resolved> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType); /** * Callback interface that can be implemented by beans wishing to customize the schema diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/TopicResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/TopicResolver.java index ad7cf286..d7716059 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/TopicResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/TopicResolver.java @@ -16,7 +16,6 @@ package org.springframework.pulsar.core; -import java.util.Optional; import java.util.function.Supplier; import org.springframework.lang.Nullable; @@ -35,7 +34,7 @@ public interface TopicResolver { * returns {@code null} to signal no default) * @return the topic to use or {@code empty} if no topic could be resolved */ - Optional resolveTopic(@Nullable String userSpecifiedTopic, Supplier defaultTopicSupplier); + Resolved resolveTopic(@Nullable String userSpecifiedTopic, Supplier defaultTopicSupplier); /** * Resolve the topic name to use for the given message. @@ -46,7 +45,7 @@ public interface TopicResolver { * returns {@code null} to signal no default) * @return the topic to use or {@code empty} if no topic could be resolved */ - Optional resolveTopic(@Nullable String userSpecifiedTopic, T message, + Resolved resolveTopic(@Nullable String userSpecifiedTopic, @Nullable T message, Supplier defaultTopicSupplier); /** @@ -57,7 +56,7 @@ public interface TopicResolver { * returns {@code null} to signal no default) * @return the topic to use or {@code empty} if no topic could be resolved */ - Optional resolveTopic(@Nullable String userSpecifiedTopic, Class messageType, + Resolved resolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class messageType, Supplier defaultTopicSupplier); } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java index c616179b..f05a3da8 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java @@ -98,7 +98,7 @@ class DefaultSchemaResolverTests { @ParameterizedTest @MethodSource("primitiveTypeMessagesProvider") void primitiveTypeMessages(T message, Schema expectedSchema) { - assertThat(resolver.getSchema(message)).isEqualTo(expectedSchema); + assertThat(resolver.resolveSchema(message).orElseThrow()).isEqualTo(expectedSchema); } static Stream primitiveTypeMessagesProvider() { @@ -137,9 +137,9 @@ class DefaultSchemaResolverTests { Schema fooSchema = Schema.AVRO(Foo.class); resolver.addCustomSchemaMapping(Foo.class, fooSchema); resolver.addCustomSchemaMapping(Bar.class, Schema.STRING); - assertThat(resolver.getSchema(new Foo("foo1"))).isSameAs(fooSchema); - assertThat(resolver.getSchema(new Bar<>("bar1"))).isEqualTo(Schema.STRING); - assertThat(resolver.getSchema(new Zaa("zaa1")).getSchemaInfo()) + assertThat(resolver.resolveSchema(new Foo("foo1")).orElseThrow()).isSameAs(fooSchema); + assertThat(resolver.resolveSchema(new Bar<>("bar1")).orElseThrow()).isEqualTo(Schema.STRING); + assertThat(resolver.resolveSchema(new Zaa("zaa1")).orElseThrow().getSchemaInfo()) .isEqualTo(Schema.JSON(Zaa.class).getSchemaInfo()); } @@ -151,7 +151,7 @@ class DefaultSchemaResolverTests { @ParameterizedTest @MethodSource("primitiveMessageTypesProvider") void primitiveMessageTypes(Class messageType, Schema expectedSchema) { - assertThat(resolver.getSchema(messageType)).isEqualTo(expectedSchema); + assertThat(resolver.resolveSchema(messageType).orElseThrow()).isEqualTo(expectedSchema); } static Stream primitiveMessageTypesProvider() { @@ -189,13 +189,15 @@ class DefaultSchemaResolverTests { @Test void customMessageTypes() { - assertThat(resolver.getSchema(Foo.class, false)).isNull(); - assertThat(resolver.getSchema(Foo.class, true).getSchemaInfo()) + assertThatExceptionOfType(IllegalArgumentException.class) + .isThrownBy(() -> resolver.resolveSchema(Foo.class, false).orElseThrow()); + assertThat(resolver.resolveSchema(Foo.class, true).orElseThrow().getSchemaInfo()) .isEqualTo(Schema.JSON(Foo.class).getSchemaInfo()); resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); - assertThat(resolver.getSchema(Foo.class, false)).isEqualTo(Schema.STRING); - assertThat(resolver.getSchema(Bar.class, false)).isNull(); - assertThat(resolver.getSchema(Bar.class, true)).isEqualTo(Schema.BYTES); + assertThat(resolver.resolveSchema(Foo.class, false).orElseThrow()).isEqualTo(Schema.STRING); + assertThatExceptionOfType(IllegalArgumentException.class) + .isThrownBy(() -> resolver.resolveSchema(Bar.class, false).orElseThrow()); + assertThat(resolver.resolveSchema(Bar.class, true).orElseThrow()).isEqualTo(Schema.BYTES); } } @@ -206,7 +208,7 @@ class DefaultSchemaResolverTests { @ParameterizedTest @MethodSource("primitiveSchemasProvider") void primitiveSchemas(SchemaType schemaType, Schema expectedSchema) { - assertThat(resolver.getSchema(schemaType, null)).isEqualTo(expectedSchema); + assertThat(resolver.resolveSchema(schemaType, null).orElseThrow()).isEqualTo(expectedSchema); } static Stream primitiveSchemasProvider() { @@ -234,17 +236,17 @@ class DefaultSchemaResolverTests { @Test void structSchemas() { - assertThat(resolver.getSchema(SchemaType.JSON, ResolvableType.forType(Foo.class))) + assertThat(resolver.resolveSchema(SchemaType.JSON, ResolvableType.forType(Foo.class)).orElseThrow()) .isInstanceOf(JSONSchema.class) .hasFieldOrPropertyWithValue("schema.fullName", sanitizedClassName(Foo.class)); - assertThat(resolver.getSchema(SchemaType.AVRO, ResolvableType.forType(Foo.class))) + assertThat(resolver.resolveSchema(SchemaType.AVRO, ResolvableType.forType(Foo.class)).orElseThrow()) .isInstanceOf(AvroSchema.class) .hasFieldOrPropertyWithValue("schema.fullName", sanitizedClassName(Foo.class)); - assertThat(resolver.getSchema(SchemaType.PROTOBUF, ResolvableType.forType(Person.class))) + assertThat(resolver.resolveSchema(SchemaType.PROTOBUF, ResolvableType.forType(Person.class)).orElseThrow()) .isInstanceOf(ProtobufSchema.class) .hasFieldOrPropertyWithValue("schema.fullName", sanitizedClassName(Proto.Person.class)); ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, String.class, Integer.class); - assertThat(resolver.getSchema(SchemaType.KEY_VALUE, kvType)) + assertThat(resolver.resolveSchema(SchemaType.KEY_VALUE, kvType).orElseThrow()) .asInstanceOf(InstanceOfAssertFactories.type(KeyValueSchema.class)).satisfies((keyValueSchema -> { assertThat(keyValueSchema.getKeySchema()).isEqualTo(Schema.STRING); assertThat(keyValueSchema.getValueSchema()).isEqualTo(Schema.INT32); @@ -255,7 +257,8 @@ class DefaultSchemaResolverTests { @ParameterizedTest @EnumSource(value = SchemaType.class, names = { "JSON", "AVRO", "PROTOBUF", "KEY_VALUE" }) void structSchemasRequireMessageType(SchemaType schemaType) { - assertThatExceptionOfType(NullPointerException.class).isThrownBy(() -> resolver.getSchema(schemaType, null)) + assertThatExceptionOfType(NullPointerException.class) + .isThrownBy(() -> resolver.resolveSchema(schemaType, null).orElseThrow()) .withMessage("messageType must be specified for " + schemaType.name()); } @@ -263,7 +266,7 @@ class DefaultSchemaResolverTests { @EnumSource(value = SchemaType.class, names = { "PROTOBUF_NATIVE", "AUTO", "AUTO_CONSUME", "AUTO_PUBLISH" }) void unsupportedSchemaTypes(SchemaType unsupportedType) { assertThatExceptionOfType(IllegalArgumentException.class) - .isThrownBy(() -> resolver.getSchema(unsupportedType, null)) + .isThrownBy(() -> resolver.resolveSchema(unsupportedType, null).orElseThrow()) .withMessage("Unsupported schema type: " + unsupportedType.name()); } @@ -276,20 +279,21 @@ class DefaultSchemaResolverTests { @Test void nullMessageType() { - assertThat(resolver.getSchema(SchemaType.NONE, null)).isEqualTo(Schema.BYTES); + assertThat(resolver.resolveSchema(SchemaType.NONE, null).orElseThrow()).isEqualTo(Schema.BYTES); } @Test void primitiveMessageType() { - assertThat(resolver.getSchema(SchemaType.NONE, ResolvableType.forType(String.class))) + assertThat(resolver.resolveSchema(SchemaType.NONE, ResolvableType.forType(String.class)).orElseThrow()) .isEqualTo(Schema.STRING); } @Test void customMessageType() { - assertThat(resolver.getSchema(SchemaType.NONE, ResolvableType.forType(Foo.class))).isNull(); + assertThatExceptionOfType(IllegalArgumentException.class).isThrownBy( + () -> resolver.resolveSchema(SchemaType.NONE, ResolvableType.forType(Foo.class)).orElseThrow()); resolver.addCustomSchemaMapping(Foo.class, Schema.STRING); - assertThat(resolver.getSchema(SchemaType.NONE, ResolvableType.forType(Foo.class))) + assertThat(resolver.resolveSchema(SchemaType.NONE, ResolvableType.forType(Foo.class)).orElseThrow()) .isEqualTo(Schema.STRING); } @@ -297,7 +301,7 @@ class DefaultSchemaResolverTests { void primitiveKeyValueMessageType() { ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, String.class, Integer.class); - assertThat(resolver.getSchema(SchemaType.NONE, kvType)) + assertThat(resolver.resolveSchema(SchemaType.NONE, kvType).orElseThrow()) .asInstanceOf(InstanceOfAssertFactories.type(KeyValueSchema.class)) .satisfies((keyValueSchema -> { assertThat(keyValueSchema.getKeySchema()).isEqualTo(Schema.STRING); @@ -309,7 +313,7 @@ class DefaultSchemaResolverTests { @Test void customKeyValueMessageTypeDefaultsToJSONSchema() { ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Zaa.class); - assertThat(resolver.getSchema(SchemaType.NONE, kvType)) + assertThat(resolver.resolveSchema(SchemaType.NONE, kvType).orElseThrow()) .asInstanceOf(InstanceOfAssertFactories.type(KeyValueSchema.class)) .satisfies((keyValueSchema -> { assertThat(keyValueSchema.getKeySchema().getSchemaInfo()) @@ -327,7 +331,7 @@ class DefaultSchemaResolverTests { resolver.addCustomSchemaMapping(Foo.class, fooSchema); resolver.addCustomSchemaMapping(Bar.class, barSchema); ResolvableType kvType = ResolvableType.forClassWithGenerics(KeyValue.class, Foo.class, Bar.class); - assertThat(resolver.getSchema(SchemaType.NONE, kvType)) + assertThat(resolver.resolveSchema(SchemaType.NONE, kvType).orElseThrow()) .asInstanceOf(InstanceOfAssertFactories.type(KeyValueSchema.class)) .satisfies((keyValueSchema -> { assertThat(keyValueSchema.getKeySchema()).isSameAs(fooSchema); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java index 69674c70..dae734d2 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java @@ -19,7 +19,6 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.AssertionsForClassTypes.assertThat; import static org.junit.jupiter.params.provider.Arguments.arguments; -import java.util.Optional; import java.util.stream.Stream; import org.assertj.core.api.InstanceOfAssertFactories; @@ -59,7 +58,7 @@ class DefaultTopicResolverTests { @MethodSource("resolveNoMessageInfoProvider") void resolveNoMessageInfo(String testName, @Nullable String userTopic, @Nullable String defaultTopic, @Nullable String expectedTopic) { - assertThatTopicIsExpected(resolver.resolveTopic(userTopic, () -> defaultTopic), expectedTopic); + assertThat(resolver.resolveTopic(userTopic, () -> defaultTopic).get().orElse(null)).isEqualTo(expectedTopic); } static Stream resolveNoMessageInfoProvider() { @@ -77,7 +76,8 @@ class DefaultTopicResolverTests { @MethodSource("resolveByMessageInstanceProvider") void resolveByMessageInstance(String testName, @Nullable String userTopic, T message, @Nullable String defaultTopic, @Nullable String expectedTopic) { - assertThatTopicIsExpected(resolver.resolveTopic(userTopic, message, () -> defaultTopic), expectedTopic); + assertThat(resolver.resolveTopic(userTopic, message, () -> defaultTopic).get().orElse(null)) + .isEqualTo(expectedTopic); } static Stream resolveByMessageInstanceProvider() { @@ -99,7 +99,8 @@ class DefaultTopicResolverTests { @MethodSource("resolveByMessageTypeProvider") void resolveByMessageType(String testName, @Nullable String userTopic, Class messageType, @Nullable String defaultTopic, @Nullable String expectedTopic) { - assertThatTopicIsExpected(resolver.resolveTopic(userTopic, messageType, () -> defaultTopic), expectedTopic); + assertThat(resolver.resolveTopic(userTopic, messageType, () -> defaultTopic).get().orElse(null)) + .isEqualTo(expectedTopic); } static Stream resolveByMessageTypeProvider() { @@ -110,7 +111,7 @@ class DefaultTopicResolverTests { arguments("complexMessageWithUserTopic", userTopic, Foo.class, defaultTopic, userTopic), arguments("complexMessageNoUserTopic", null, Foo.class, defaultTopic, fooTopic), arguments("nullMessageWithUserTopicAndDefault", userTopic, null, defaultTopic, userTopic), - arguments("nullMessageWithDefault", null, null, defaultTopic, defaultTopic), + arguments("nullMessageWithDefault", null, null, defaultTopic, null), arguments("noMatchWithUserTopicAndDefault", userTopic, Bar.class, defaultTopic, userTopic), arguments("noMatchWithUserTopic", userTopic, Bar.class, null, userTopic), arguments("noMatchWithDefault", null, Bar.class, defaultTopic, defaultTopic), @@ -119,15 +120,6 @@ class DefaultTopicResolverTests { // @formatter:on } - private void assertThatTopicIsExpected(Optional actual, @Nullable String expectedTopic) { - if (expectedTopic == null) { - assertThat(actual).isEmpty(); - } - else { - assertThat(actual).hasValue(expectedTopic); - } - } - @Nested class TopicMappingsAPI { 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 d7215572..a6b2ce08 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -17,6 +17,8 @@ package org.springframework.pulsar.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.params.provider.Arguments.arguments; import static org.mockito.ArgumentMatchers.any; @@ -28,21 +30,23 @@ import static org.mockito.Mockito.when; import java.time.Duration; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.UUID; -import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; -import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Message; 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.interceptor.ProducerInterceptor; import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Named; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -50,8 +54,8 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; -import org.springframework.pulsar.core.PulsarOperations.SendMessageBuilder; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; +import org.springframework.util.function.ThrowingConsumer; /** * Tests for {@link PulsarTemplate}. @@ -63,129 +67,125 @@ import org.springframework.pulsar.test.support.PulsarTestContainerSupport; */ class PulsarTemplateTests implements PulsarTestContainerSupport { - @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"; - TypedMessageBuilderCustomizer messageCustomizer = null; - if (testArgs.messageCustomizer) { - messageCustomizer = (mb) -> mb.key("foo-key"); - } - ProducerBuilderCustomizer producerCustomizer = null; - if (testArgs.producerCustomizer) { - producerCustomizer = (pb) -> pb.producerName("foo-producer"); - } - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.STRING).topic(topic) - .subscriptionName(subscription).subscribe()) { - Map producerConfig = testArgs.explicitTopic ? Collections.emptyMap() - : Collections.singletonMap("topicName", topic); - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - producerConfig); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + private PulsarClient client; - Object sendResponse; - if (testArgs.simpleApi) { - if (testArgs.explicitSchema && testArgs.explicitTopic) { - sendResponse = testArgs.async ? pulsarTemplate.sendAsync(topic, msgPayload, Schema.STRING) - : pulsarTemplate.send(topic, msgPayload, Schema.STRING); - } - else if (testArgs.explicitSchema) { - sendResponse = testArgs.async ? pulsarTemplate.sendAsync(msgPayload, Schema.STRING) - : pulsarTemplate.send(msgPayload, Schema.STRING); - } - else if (testArgs.explicitTopic) { - sendResponse = testArgs.async ? pulsarTemplate.sendAsync(topic, msgPayload) - : pulsarTemplate.send(topic, msgPayload); - } - else { - sendResponse = testArgs.async ? pulsarTemplate.sendAsync(msgPayload) - : pulsarTemplate.send(msgPayload); - } - } - else { - SendMessageBuilder messageBuilder = pulsarTemplate.newMessage(msgPayload); - if (testArgs.explicitTopic) { - messageBuilder = messageBuilder.withTopic(topic); - } - if (testArgs.explicitSchema) { - messageBuilder = messageBuilder.withSchema(Schema.STRING); - } - if (messageCustomizer != null) { - messageBuilder = messageBuilder.withMessageCustomizer(messageCustomizer); - } - if (producerCustomizer != null) { - messageBuilder = messageBuilder.withProducerCustomizer(producerCustomizer); - } - sendResponse = testArgs.async ? 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 (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()); - } - } + @BeforeEach + void setup() throws PulsarClientException { + client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build(); } - private static Stream sendMessageTestProvider() { - return Stream.of(arguments("simpleSend", SendTestArgs.simple().sync()), - arguments("simpleSendWithTopic", SendTestArgs.simple().sync().topic()), - arguments("simpleSendWithSchema", SendTestArgs.simple().sync().schema()), - arguments("simpleSendWithTopicAndSchema", SendTestArgs.simple().sync().topic().schema()), - arguments("simpleAsyncSend", SendTestArgs.simple().async()), - arguments("simpleAsyncSendWithTopic", SendTestArgs.simple().async().topic()), - arguments("simpleAsyncSendWithSchema", SendTestArgs.simple().async().schema()), - arguments("simpleAsyncSendWithTopicAndSchema", SendTestArgs.simple().async().topic().schema()), - arguments("fluentSend", SendTestArgs.fluent().sync()), - arguments("fluentSendWithSchema", SendTestArgs.fluent().sync().schema()), - arguments("fluentSendWithTopic", SendTestArgs.fluent().sync().topic()), - arguments("fluentSendWithMessageCustomizer", SendTestArgs.fluent().sync().messageCustomizer()), - arguments("fluentSendWithProducerCustomizer", SendTestArgs.fluent().sync().producerCustomizer()), - arguments("fluentSendWithTopicAndSchema", SendTestArgs.fluent().sync().topic().schema()), - arguments("fluentSendWithTopicAndSchemaAndCustomizers", - SendTestArgs.fluent().sync().topic().schema().messageCustomizer().producerCustomizer()), - arguments("fluentAsyncSend", SendTestArgs.fluent().async()), - arguments("fluentAsyncSendWithSchema", SendTestArgs.fluent().async().schema()), - arguments("fluentAsyncSendWithTopic", SendTestArgs.fluent().async().topic()), - arguments("fluentAsyncSendWithMessageCustomizer", SendTestArgs.fluent().async().messageCustomizer()), - arguments("fluentAsyncSendWithProducerCustomizer", SendTestArgs.fluent().async().producerCustomizer()), - arguments("fluentAsyncSendWithTopicAndSchema", SendTestArgs.fluent().async().topic().schema()), - arguments("fluentAsyncSendWithTopicAndSchemaAndCustomizers", - SendTestArgs.fluent().async().topic().schema().messageCustomizer().producerCustomizer())); + @AfterEach + void tearDown() throws PulsarClientException { + // 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()); + client.close(); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("sendMessageTestProvider") + void sendMessageTest(String testName, ThrowingConsumer> sendFunction, + Boolean withDefaultTopic, String expectedValue) throws Exception { + sendAndConsume(sendFunction, testName, Schema.STRING, expectedValue, withDefaultTopic); + } + + static Stream sendMessageTestProvider() { + String message = "test-message"; + + return Stream.of( + // Simple send sync + arguments("simpleSendWithDefaultTopic", + (ThrowingConsumer>) (template) -> template.send(message), true, message), + arguments("simpleSendWithTopic", + (ThrowingConsumer>) (template) -> template.send("simpleSendWithTopic", + message), + false, message), + arguments("simpleSendWithDefaultTopicAndSchema", + (ThrowingConsumer>) (template) -> template.send(message, Schema.STRING), + true, message), + arguments("simpleSendWithTopicAndSchema", + (ThrowingConsumer>) (template) -> template + .send("simpleSendWithTopicAndSchema", message, Schema.STRING), + false, message), + arguments("simpleSendNullWithTopicAndSchema", + (ThrowingConsumer>) (template) -> template + .send("simpleSendNullWithTopicAndSchema", null, Schema.STRING), + false, null), + + // Simple send async + arguments("simpleSendAsyncWithDefaultTopic", + (ThrowingConsumer>) (template) -> template.sendAsync(message).get(3, + TimeUnit.SECONDS), + true, message), + arguments("simpleSendAsyncWithTopic", + (ThrowingConsumer>) (template) -> template + .sendAsync("simpleSendAsyncWithTopic", message).get(3, TimeUnit.SECONDS), + false, message), + arguments("simpleSendAsyncWithDefaultTopicAndSchema", + (ThrowingConsumer>) (template) -> template + .sendAsync(message, Schema.STRING).get(3, TimeUnit.SECONDS), + true, message), + arguments("simpleSendAsyncWithTopicAndSchema", + (ThrowingConsumer>) (template) -> template + .sendAsync("simpleSendAsyncWithTopicAndSchema", message, Schema.STRING) + .get(3, TimeUnit.SECONDS), + false, message), + arguments("simpleSendAsyncNullWithTopicAndSchema", + (ThrowingConsumer>) (template) -> template + .sendAsync("simpleSendAsyncNullWithTopicAndSchema", null, Schema.STRING) + .get(3, TimeUnit.SECONDS), + false, null), + + // Fluent send + arguments("fluentSendWithDefaultTopic", + (ThrowingConsumer>) (template) -> template.newMessage(message).send(), + true, message), + arguments("fluentSendWithTopic", + (ThrowingConsumer>) (template) -> template.newMessage(message) + .withTopic("fluentSendWithTopic").send(), + false, message), + arguments("fluentSendWithDefaultTopicAndSchema", + (ThrowingConsumer>) (template) -> template.newMessage(message) + .withSchema(Schema.STRING).send(), + true, message), + arguments("fluentSendNullWithTopicAndSchema", + (ThrowingConsumer>) (template) -> template.newMessage(null) + .withSchema(Schema.STRING).withTopic("fluentSendNullWithTopicAndSchema").send(), + false, null), + arguments("fluentSendAsync", (ThrowingConsumer>) (template) -> template + .newMessage(message).sendAsync().get(3, TimeUnit.SECONDS), true, message) + + ); + } + + @Test + void sendMessageWithMessageCustomizer() throws Exception { + ThrowingConsumer> sendFunction = (template) -> template.newMessage("test-message") + .withMessageCustomizer((mb) -> mb.key("test-key")).send(); + Message msg = sendAndConsume(sendFunction, "sendMessageWithMessageCustomizer", Schema.STRING, + "test-message", true); + assertThat(msg.getKey()).isEqualTo("test-key"); + } + + @Test + void sendMessageWithSenderCustomizer() throws Exception { + ThrowingConsumer> sendFunction = (template) -> template.newMessage("test-message") + .withProducerCustomizer((sb) -> sb.producerName("test-producer")).send(); + Message msg = sendAndConsume(sendFunction, "sendMessageWithSenderCustomizer", Schema.STRING, + "test-message", true); + assertThat(msg.getProducerName()).isEqualTo("test-producer"); } @ParameterizedTest(name = "{0}") @MethodSource("interceptorInvocationTestProvider") void interceptorInvocationTest(String topic, List interceptors) throws Exception { - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - Collections.singletonMap("topicName", topic)); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); - pulsarTemplate.send("test-interceptor"); - for (ProducerInterceptor interceptor : interceptors) { - verify(interceptor, atLeastOnce()).eligible(any(Message.class)); - } + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + Collections.singletonMap("topicName", topic)); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); + pulsarTemplate.send("test-interceptor"); + for (ProducerInterceptor interceptor : interceptors) { + verify(interceptor, atLeastOnce()).eligible(any(Message.class)); } } @@ -198,161 +198,126 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { } @Test - void sendMessageWithSpecificSchema() throws Exception { + void sendNonPrimitiveMessageWithSpecifiedSchema() throws Exception { String topic = "ptt-specificSchema-topic"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.AVRO(Foo.class)).topic(topic) - .subscriptionName("ptt-specificSchema-subs").subscribe()) { - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - Collections.singletonMap("topicName", topic)); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); - Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); - pulsarTemplate.send(foo, Schema.AVRO(Foo.class)); - assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) - .isEqualTo(foo); - } - } + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + ThrowingConsumer> sendFunction = (template) -> template.send(foo, Schema.AVRO(Foo.class)); + sendAndConsume(sendFunction, topic, Schema.AVRO(Foo.class), foo, true); } @Test - void sendMessageWithoutSpecificSchema() throws Exception { + void sendNonPrimitiveMessageWithInferredSchema() throws Exception { String topic = "ptt-nospecificSchema-topic"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) - .subscriptionName("ptt-nospecificSchema-subs").subscribe()) { - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - Collections.singletonMap("topicName", topic)); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); - Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); - pulsarTemplate.send(foo); - assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) - .isEqualTo(foo); - } - } + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + ThrowingConsumer> sendFunction = (template) -> template.send(foo); + sendAndConsume(sendFunction, topic, Schema.JSON(Foo.class), foo, true); } @Test void sendMessageWithSpecificSchemaInferredByCustomTypeMappings() throws Exception { String topic = "ptt-schemaInferred-topic"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) - .subscriptionName("ptt-schemaInferred-subs").subscribe()) { - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - Collections.singletonMap("topicName", topic)); - // Custom schema resolver allows not specifying the schema when sending - DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); - schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), - schemaResolver, new DefaultTopicResolver(), null, null); - Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); - pulsarTemplate.send(foo); - assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) - .isEqualTo(foo); - } - } + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + Collections.singletonMap("topicName", topic)); + // Custom schema resolver allows not specifying the schema when sending + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver(); + schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class)); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), + schemaResolver, new DefaultTopicResolver(), null, null); + + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + ThrowingConsumer> sendFunction = (template) -> template.newMessage(foo).send(); + sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.JSON(Foo.class), foo); } @ParameterizedTest @ValueSource(booleans = { true, false }) void sendMessageTopicInferredByCustomTypeMappings(boolean producerFactoryHasDefaultTopic) throws Exception { String topic = "ptt-topicInferred-" + producerFactoryHasDefaultTopic + "-topic"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) - .subscriptionName("ptt-topicInferred-subs").subscribe()) { - PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, - producerFactoryHasDefaultTopic ? Collections.singletonMap("topicName", "fake-topic") - : Collections.emptyMap()); - // Topic mappings allows not specifying the topic when sending (nor having - // default on producer) - DefaultTopicResolver topicResolver = new DefaultTopicResolver(); - topicResolver.addCustomTopicMapping(Foo.class, topic); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), - new DefaultSchemaResolver(), topicResolver, null, null); - Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); - pulsarTemplate.send(foo, Schema.JSON(Foo.class)); - assertThat(consumer.receiveAsync()).succeedsWithin(Duration.ofSeconds(3)).extracting(Message::getValue) - .isEqualTo(foo); - } - } + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + producerFactoryHasDefaultTopic ? Collections.singletonMap("topicName", "fake-topic") + : Collections.emptyMap()); + // Topic mappings allows not specifying the topic when sending (nor having + // default on producer) + DefaultTopicResolver topicResolver = new DefaultTopicResolver(); + topicResolver.addCustomTopicMapping(Foo.class, topic); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), + new DefaultSchemaResolver(), topicResolver, null, null); + + Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID()); + ThrowingConsumer> sendFunction = (template) -> template.send(foo, Schema.JSON(Foo.class)); + sendAndConsume(pulsarTemplate, sendFunction, topic, Schema.JSON(Foo.class), foo); } @Test @SuppressWarnings("unchecked") void sendMessageWithEncryptionKeys() throws Exception { String topic = "ptt-encryptionKeys-topic"; - try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) - .build()) { - PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); - when(producerFactory.createProducer(Schema.STRING, topic, Set.of("key"), new ArrayList<>())) - .thenReturn(client.newProducer(Schema.STRING).topic(topic).create()); - PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); - pulsarTemplate.newMessage("msg").withTopic(topic).withEncryptionKeys(Set.of("key")).send(); - verify(producerFactory).createProducer(Schema.STRING, topic, Set.of("key"), new ArrayList<>()); - } + PulsarProducerFactory producerFactory = mock(PulsarProducerFactory.class); + when(producerFactory.createProducer(Schema.STRING, topic, Set.of("key"), new ArrayList<>())) + .thenReturn(client.newProducer(Schema.STRING).topic(topic).create()); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); + pulsarTemplate.newMessage("msg").withTopic(topic).withEncryptionKeys(Set.of("key")).send(); + verify(producerFactory).createProducer(Schema.STRING, topic, Set.of("key"), new ArrayList<>()); } - static final class SendTestArgs { + @ParameterizedTest(name = "{0}") + @MethodSource("sendMessageFailedTestProvider") + void sendMessageFailed(String testName, ThrowingConsumer> sendFunction) { + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, new HashMap<>()); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); + assertThatIllegalArgumentException().isThrownBy(() -> sendFunction.accept(pulsarTemplate)); + } - private final boolean simpleApi; + static Stream sendMessageFailedTestProvider() { + String message = "test-message"; + return Stream.of( + arguments("sendWithoutTopic", + (ThrowingConsumer>) (template) -> template.send(message)), + arguments("sendNullWithoutSchema", (ThrowingConsumer>) (template) -> template + .send("sendNullWithoutSchema", (String) null))); + } - private boolean async; + @Test + void sendNullWithDefaultTopicFails() { + HashMap config = new HashMap<>(); + config.put("topicName", "sendNullWithDefaultTopicFails"); + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, config); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); + assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send(null, Schema.STRING)); + } - private boolean explicitTopic; + @Test + void sendWithoutSchemaFails() { + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, new HashMap<>()); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); + // Defaulting to Schema.JSON would prevent this from failing + assertThatExceptionOfType(ClassCastException.class) + .isThrownBy(() -> pulsarTemplate.send("sendWithoutSchemaFails", new Foo("foo", "bar"))); + } - private boolean explicitSchema; - - private boolean messageCustomizer; - - private boolean producerCustomizer; - - private SendTestArgs(boolean simpleApi) { - this.simpleApi = simpleApi; + private Message sendAndConsume(ThrowingConsumer> sendFunction, String topic, + Schema schema, T expectedValue, Boolean withDefaultTopic) throws Exception { + Map config = new HashMap<>(); + if (withDefaultTopic) { + config.put("topicName", topic); } + PulsarProducerFactory senderFactory = new DefaultPulsarProducerFactory<>(client, config); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(senderFactory); + return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue); + } - static SendTestArgs simple() { - return new SendTestArgs(true); + private Message sendAndConsume(PulsarTemplate template, ThrowingConsumer> sendFunction, + String topic, Schema schema, T expectedValue) throws Exception { + try (org.apache.pulsar.client.api.Consumer consumer = client.newConsumer(schema).topic(topic) + .subscriptionName(topic + "-sub").subscribe()) { + sendFunction.accept(template); + Message msg = consumer.receive(3, TimeUnit.SECONDS); + assertThat(msg).isNotNull(); + assertThat(msg.getValue()).isEqualTo(expectedValue); + return msg; } - - static SendTestArgs fluent() { - return new SendTestArgs(false); - } - - SendTestArgs async() { - this.async = true; - return this; - } - - SendTestArgs sync() { - this.async = false; - return this; - } - - SendTestArgs topic() { - this.explicitTopic = true; - return this; - } - - SendTestArgs schema() { - this.explicitSchema = true; - return this; - } - - SendTestArgs messageCustomizer() { - this.messageCustomizer = true; - return this; - } - - SendTestArgs producerCustomizer() { - this.producerCustomizer = true; - return this; - } - } public static class Foo {