Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add support for tombstone/null values

Browse Source 
Fixes #311
This commit is contained in:
Christophe Bornet
2023-02-06 19:28:41 +01:00
committed by Chris Bono
parent 4da2f5025c
commit 33712c6bef
20 changed files with 669 additions and 695 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-pulsar-docs/src/main/asciidoc/topic-resolution.adoc
View File

@@ -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:

9
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java
View File

@@ -138,10 +138,9 @@ public class MethodReactivePulsarListenerEndpoint<V> 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<V> 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<V> customizer1 = b -> b.deadLetterPolicy(this.deadLetterPolicy);

4
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java
View File

@@ -96,9 +96,7 @@ public class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSend
@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> 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<T> sender = this.reactivePulsarClient.messageSender(schema);

10
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarOperations.java
View File

@@ -41,7 +41,7 @@ public interface ReactivePulsarOperations<T> {
* @param message the message to send
* @return the id assigned by the broker to the published message
*/
Mono<MessageId> send(T message);
Mono<MessageId> send(@Nullable T message);
/**
* Sends a message to the specified topic in a reactive manner. default topic
@@ -50,7 +50,7 @@ public interface ReactivePulsarOperations<T> {
* resolution
* @return the id assigned by the broker to the published message
*/
Mono<MessageId> send(T message, @Nullable Schema<T> schema);
Mono<MessageId> send(@Nullable T message, @Nullable Schema<T> schema);
/**
* Sends a message to the specified topic in a reactive manner.
@@ -59,7 +59,7 @@ public interface ReactivePulsarOperations<T> {
* @param message the message to send
* @return the id assigned by the broker to the published message
*/
Mono<MessageId> send(@Nullable String topic, T message);
Mono<MessageId> 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<T> {
* resolution
* @return the id assigned by the broker to the published message
*/
Mono<MessageId> send(@Nullable String topic, T message, @Nullable Schema<T> schema);
Mono<MessageId> send(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema);
/**
* Sends multiple messages to the default topic in a reactive manner.
@@ -119,7 +119,7 @@ public interface ReactivePulsarOperations<T> {
* @param message the payload of the message
* @return the builder to configure and send the message
*/
SendOneMessageBuilder<T> newMessage(T message);
SendOneMessageBuilder<T> newMessage(@Nullable T message);
/**
* Create a {@link SendManyMessageBuilder builder} for configuring and sending

44
spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java
View File

@@ -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<T> implements ReactivePulsarOperations<T> {
}
@Override
public Mono<MessageId> send(T message) {
public Mono<MessageId> send(@Nullable T message) {
return send(null, message);
}
@Override
public Mono<MessageId> send(T message, @Nullable Schema<T> schema) {
public Mono<MessageId> send(@Nullable T message, @Nullable Schema<T> schema) {
return doSend(null, message, schema, null, null);
}
@Override
public Mono<MessageId> send(@Nullable String topic, T message) {
public Mono<MessageId> send(@Nullable String topic, @Nullable T message) {
return doSend(topic, message, null, null, null);
}
@Override
public Mono<MessageId> send(@Nullable String topic, T message, @Nullable Schema<T> schema) {
public Mono<MessageId> send(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema) {
return doSend(topic, message, schema, null, null);
}
@@ -118,7 +116,7 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
}
@Override
public SendOneMessageBuilder<T> newMessage(T message) {
public SendOneMessageBuilder<T> newMessage(@Nullable T message) {
return new SendOneMessageBuilderImpl<>(this, message);
}
@@ -127,10 +125,10 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
return new SendManyMessageBuilderImpl<>(this, messages);
}
private Mono<MessageId> doSend(@Nullable String topic, T message, @Nullable Schema<T> schema,
private Mono<MessageId> doSend(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema,
@Nullable MessageSpecBuilderCustomizer<T> messageSpecBuilderCustomizer,
@Nullable ReactiveMessageSenderBuilderCustomizer<T> customizer) {
String topicName = resolveTopic(topic, message.getClass());
String topicName = resolveTopic(topic, message);
this.logger.trace(() -> "Sending reactive msg to '%s' topic".formatted(topicName));
ReactiveMessageSender<T> sender = createMessageSender(topicName, message, schema, customizer);
// @formatter:off
@@ -145,7 +143,7 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
return messages.switchOnFirst((firstSignal, messageFlux) -> {
MessageSpec<T> firstMessage = firstSignal.get();
if (firstMessage != null && firstSignal.isOnNext()) {
String topicName = resolveTopic(topic, firstMessage.getValue().getClass());
String topicName = resolveTopic(topic, firstMessage.getValue());
ReactiveMessageSender<T> sender = createMessageSender(topicName, firstMessage.getValue(), schema,
customizer);
return messageFlux.as(sender::sendMany).doOnError(
@@ -157,21 +155,13 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
});
}
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<String> 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 <T> MessageSpec<T> getMessageSpec(
@Nullable MessageSpecBuilderCustomizer<T> messageSpecBuilderCustomizer, T message) {
@Nullable MessageSpecBuilderCustomizer<T> messageSpecBuilderCustomizer, @Nullable T message) {
MessageSpecBuilder<T> messageSpecBuilder = MessageSpec.builder(message);
if (messageSpecBuilderCustomizer != null) {
messageSpecBuilderCustomizer.customize(messageSpecBuilder);
@@ -179,12 +169,9 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
return messageSpecBuilder.build();
}
private ReactiveMessageSender<T> createMessageSender(@Nullable String topic, T message, @Nullable Schema<T> schema,
@Nullable ReactiveMessageSenderBuilderCustomizer<T> customizer) {
Schema<T> resolvedSchema = schema == null ? this.schemaResolver.getSchema(message) : schema;
if (resolvedSchema == null) {
throw new IllegalArgumentException("Couldn't resolve a schema for the message");
}
private ReactiveMessageSender<T> createMessageSender(@Nullable String topic, @Nullable T message,
@Nullable Schema<T> schema, @Nullable ReactiveMessageSenderBuilderCustomizer<T> customizer) {
Schema<T> resolvedSchema = schema == null ? this.schemaResolver.resolveSchema(message).orElseThrow() : schema;
return this.reactiveMessageSenderFactory.createSender(resolvedSchema, topic, customizer);
}
@@ -228,12 +215,13 @@ public class ReactivePulsarTemplate<T> implements ReactivePulsarOperations<T> {
private static final class SendOneMessageBuilderImpl<T>
extends SendMessageBuilderImpl<SendOneMessageBuilderImpl<T>, T> implements SendOneMessageBuilder<T> {
@Nullable
private final T message;
@Nullable
private MessageSpecBuilderCustomizer<T> messageCustomizer;
SendOneMessageBuilderImpl(ReactivePulsarTemplate<T> template, T message) {
SendOneMessageBuilderImpl(ReactivePulsarTemplate<T> template, @Nullable T message) {
super(template);
this.message = message;
}

437
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java
View File

@@ -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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic).subscriptionName(sub)
.subscribe()) {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
senderSpec.setTopicName(topic);
ReactivePulsarSenderFactory<Foo> producerFactory = new DefaultReactivePulsarSenderFactory<>(client,
senderSpec, null);
ReactivePulsarTemplate<Foo> pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory);
private PulsarClient client;
List<Foo> 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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic).subscriptionName(sub)
.subscribe()) {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
senderSpec.setTopicName(topic);
ReactivePulsarSenderFactory<Foo> 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<Foo> pulsarTemplate = new ReactivePulsarTemplate<>(producerFactory,
schemaResolver, new DefaultTopicResolver());
List<Foo> 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<Foo> foos2 = new ArrayList<>();
for (int i = 0; i < 10; i++) {
CompletableFuture<Message<Foo>> 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<List<String>, ReactivePulsarTemplate<String>> sendHandler) throws Exception {
String topic = "rptt-" + testName + "-topic";
String sub = "rptt-" + testName + "-sub";
try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
.build()) {
try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic).subscriptionName(sub)
.subscribe()) {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
senderSpec.setTopicName(topic + "-fake");
ReactivePulsarSenderFactory<String> 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<String> 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<Arguments> sendManyWithInferredTopicProvider() {
return Stream.of(
arguments("simpleApiNoSchema",
(BiConsumer<List<String>, ReactivePulsarTemplate<String>>) (data, template) -> template
.send(Flux.fromIterable(data).map(MessageSpec::of)).subscribe()),
arguments("simpleApiWithSchema",
(BiConsumer<List<String>, ReactivePulsarTemplate<String>>) (data, template) -> template
.send(Flux.fromIterable(data).map(MessageSpec::of), Schema.STRING).subscribe()),
arguments("fluentApiNoSchema",
(BiConsumer<List<String>, ReactivePulsarTemplate<String>>) (data, template) -> template
.newMessages(Flux.fromIterable(data).map(MessageSpec::of)).send().subscribe()),
arguments("fluentApiWithSchema",
(BiConsumer<List<String>, ReactivePulsarTemplate<String>>) (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<String> messageCustomizer = null;
if (testArgs.messageCustomizer) {
messageCustomizer = (mb) -> mb.key("foo-key");
}
ReactiveMessageSenderBuilderCustomizer<String> senderCustomizer = null;
if (testArgs.senderCustomizer) {
senderCustomizer = (sb) -> sb.producerName("foo-sender");
}
try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
.build()) {
try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic)
.subscriptionName(subscription).subscribe()) {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
if (!testArgs.explicitTopic) {
senderSpec.setTopicName(topic);
}
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
senderSpec, null);
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
Mono<MessageId> 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<String> 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<String> 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<ReactivePulsarTemplate<String>> sendFunction,
Boolean withDefaultTopic, String expectedValue) throws Exception {
sendAndConsume(sendFunction, testName, Schema.STRING, expectedValue, withDefaultTopic);
}
private static Stream<Arguments> 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<Arguments> sendMessageTestProvider() {
String message = "test-message";
Flux<MessageSpec<String>> messagePublisher = Flux.just(MessageSpec.of(message));
return Stream.of(
arguments("simpleSendWithDefaultTopic",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.send(message).subscribe(),
true, message),
arguments("simpleSendWithTopic",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.send("simpleSendWithTopic", message).subscribe(),
false, message),
arguments("simpleSendWithDefaultTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.send(message, Schema.STRING)
.subscribe(),
true, message),
arguments("simpleSendWithTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.send("simpleSendWithTopicAndSchema", message, Schema.STRING).subscribe(),
false, message),
arguments("simpleSendNullWithTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.send("simpleSendNullWithTopicAndSchema", (String) null, Schema.STRING).subscribe(),
false, null),
arguments("simplePublisherSendWithDefaultTopic",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.send(messagePublisher)
.subscribe(),
true, message),
arguments("simplePublisherSendWithTopic",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.send("simplePublisherSendWithTopic", messagePublisher).subscribe(),
false, message),
arguments("simplePublisherSendWithDefaultTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.send(messagePublisher, Schema.STRING).subscribe(),
true, message),
arguments("simplePublisherSendWithTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.send("simplePublisherSendWithTopicAndSchema", messagePublisher, Schema.STRING)
.subscribe(),
false, message),
arguments("fluentSendWithDefaultTopic",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.newMessage(message).send()
.subscribe(),
true, message),
arguments("fluentSendWithTopic",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.newMessage(message)
.withTopic("fluentSendWithTopic").send().subscribe(),
false, message),
arguments("fluentSendWithDefaultTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.newMessage(message)
.withSchema(Schema.STRING).send().subscribe(),
true, message),
arguments("fluentSendNullWithTopicAndSchema",
(Consumer<ReactivePulsarTemplate<String>>) (template) -> template.newMessage(null)
.withSchema(Schema.STRING).withTopic("fluentSendNullWithTopicAndSchema").send()
.subscribe(),
false, null),
arguments("fluentPublisherSend", (Consumer<ReactivePulsarTemplate<String>>) (template) -> template
.newMessages(messagePublisher).send().subscribe(), true, message));
}
static final class SendTestArgs {
@Test
void sendMessageWithMessageCustomizer() throws Exception {
Consumer<ReactivePulsarTemplate<String>> sendFunction = (template) -> template.newMessage("test-message")
.withMessageCustomizer((mb) -> mb.key("test-key")).send().subscribe();
Message<String> msg = sendAndConsume(sendFunction, "sendMessageWithMessageCustomizer", Schema.STRING,
"test-message", true);
assertThat(msg.getKey()).isEqualTo("test-key");
}
private boolean simpleApi;
@Test
void sendMessageWithSenderCustomizer() throws Exception {
Consumer<ReactivePulsarTemplate<String>> sendFunction = (template) -> template.newMessage("test-message")
.withSenderCustomizer((sb) -> sb.producerName("test-producer")).send().subscribe();
Message<String> 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<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
new MutableReactiveMessageSenderSpec(), null);
private boolean messageCustomizer;
DefaultTopicResolver topicResolver = new DefaultTopicResolver();
topicResolver.addCustomTopicMapping(String.class, topic);
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory,
new DefaultSchemaResolver(), topicResolver);
private boolean senderCustomizer;
Consumer<ReactivePulsarTemplate<String>> 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<Foo> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
new MutableReactiveMessageSenderSpec(), null);
DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
schemaResolver.addCustomSchemaMapping(Foo.class, Schema.JSON(Foo.class));
ReactivePulsarTemplate<Foo> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory, schemaResolver,
new DefaultTopicResolver());
Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
Consumer<ReactivePulsarTemplate<Foo>> 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<ReactivePulsarTemplate<String>> sendFunction) {
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client,
new MutableReactiveMessageSenderSpec(), null);
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
assertThatIllegalArgumentException().isThrownBy(() -> sendFunction.accept(pulsarTemplate));
}
static Stream<Arguments> sendMessageFailedTestProvider() {
String message = "test-message";
return Stream.of(
arguments("sendWithoutTopic",
(ThrowingConsumer<ReactivePulsarTemplate<String>>) (template) -> template.send(message)),
arguments("sendNullWithoutSchema",
(ThrowingConsumer<ReactivePulsarTemplate<String>>) (template) -> template
.send("sendNullWithoutSchema", (String) null)));
}
@Test
void sendNullWithDefaultTopicFails() {
MutableReactiveMessageSenderSpec spec = new MutableReactiveMessageSenderSpec();
spec.setTopicName("sendNullWithDefaultTopicFails");
ReactivePulsarSenderFactory<String> senderFactory = new DefaultReactivePulsarSenderFactory<>(client, spec,
null);
ReactivePulsarTemplate<String> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send((String) null, Schema.STRING));
}
private <T> Message<T> sendAndConsume(Consumer<ReactivePulsarTemplate<T>> sendFunction, String topic,
Schema<T> schema, T expectedValue, Boolean withDefaultTopic) throws Exception {
MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec();
if (withDefaultTopic) {
senderSpec.setTopicName(topic);
}
ReactivePulsarSenderFactory<T> senderFactory = new DefaultReactivePulsarSenderFactory<>(client, senderSpec,
null);
static SendTestArgs simple() {
return new SendTestArgs(true);
ReactivePulsarTemplate<T> pulsarTemplate = new ReactivePulsarTemplate<>(senderFactory);
return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue);
}
private <T> Message<T> sendAndConsume(ReactivePulsarTemplate<T> template,
Consumer<ReactivePulsarTemplate<T>> sendFunction, String topic, Schema<T> schema, T expectedValue)
throws Exception {
try (org.apache.pulsar.client.api.Consumer<T> consumer = client.newConsumer(schema).topic(topic)
.subscriptionName(topic + "-sub").subscribe()) {
sendFunction.accept(template);
Message<T> 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) {

2
spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java
View File

@@ -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

23
spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinderResolveSchemaTests.java
View File

@@ -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

9
spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java
View File

@@ -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<V> 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<V> 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);

3
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -126,8 +126,7 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
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

98
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java
View File

@@ -123,16 +123,22 @@ public class DefaultSchemaResolver implements SchemaResolver {
}
@Override
public <T> Schema<T> getSchema(Class<?> messageClass, boolean returnDefault) {
public <T> Resolved<Schema<T>> 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 <T> Schema<T> 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<? extends GeneratedMessageV3>) messageClass);
}
case KEY_VALUE -> {
requireNonNullMessageType(schemaType, messageType);
yield getMessageKeyValueSchema(messageType);
}
case NONE -> {
if (messageType == null) {
yield Schema.BYTES;
public <T> Resolved<Schema<T>> 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<? extends GeneratedMessageV3>) 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<? extends Class<?>> keySchema = this.getSchema(keyClass);
Schema<? extends Class<?>> valueSchema = this.getSchema(valueClass);
Schema<?> keySchema = this.resolveSchema(keyClass).orElseThrow();
Schema<?> valueSchema = this.resolveSchema(valueClass).orElseThrow();
return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE);
}

29
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java
View File

@@ -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<String> resolveTopic(@Nullable String userSpecifiedTopic, Supplier<String> defaultTopicSupplier) {
return doResolveTopic(userSpecifiedTopic, null, defaultTopicSupplier);
public Resolved<String> resolveTopic(@Nullable String userSpecifiedTopic, Supplier<String> 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 <T> Optional<String> resolveTopic(@Nullable String userSpecifiedTopic, T message,
public <T> Resolved<String> resolveTopic(@Nullable String userSpecifiedTopic, @Nullable T message,
Supplier<String> defaultTopicSupplier) {
return doResolveTopic(userSpecifiedTopic, message.getClass(), defaultTopicSupplier);
return doResolveTopic(userSpecifiedTopic, message != null ? message.getClass() : null, defaultTopicSupplier);
}
@Override
public Optional<String> resolveTopic(@Nullable String userSpecifiedTopic, Class<?> messageType,
public Resolved<String> resolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class<?> messageType,
Supplier<String> defaultTopicSupplier) {
return doResolveTopic(userSpecifiedTopic, messageType, defaultTopicSupplier);
}
private Optional<String> doResolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class<?> messageType,
protected Resolved<String> doResolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class<?> messageType,
Supplier<String> 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);
}
}

20
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarOperations.java
View File

@@ -40,7 +40,7 @@ public interface PulsarOperations<T> {
* @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<T> {
* @return the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
MessageId send(T message, @Nullable Schema<T> schema) throws PulsarClientException;
MessageId send(@Nullable T message, @Nullable Schema<T> schema) throws PulsarClientException;
/**
* Sends a message to the specified topic in a blocking manner.
@@ -60,7 +60,7 @@ public interface PulsarOperations<T> {
* @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<T> {
* @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<T> schema) throws PulsarClientException;
MessageId send(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema)
throws PulsarClientException;
/**
* Sends a message to the default topic in a non-blocking manner.
@@ -80,7 +81,7 @@ public interface PulsarOperations<T> {
* @return a future that holds the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException;
CompletableFuture<MessageId> 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<T> {
* @return a future that holds the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
CompletableFuture<MessageId> sendAsync(T message, @Nullable Schema<T> schema) throws PulsarClientException;
CompletableFuture<MessageId> sendAsync(@Nullable T message, @Nullable Schema<T> schema)
throws PulsarClientException;
/**
* Sends a message to the specified topic in a non-blocking manner.
@@ -100,7 +102,7 @@ public interface PulsarOperations<T> {
* @return a future that holds the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
CompletableFuture<MessageId> sendAsync(@Nullable String topic, T message) throws PulsarClientException;
CompletableFuture<MessageId> 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<T> {
* @return a future that holds the id assigned by the broker to the published message
* @throws PulsarClientException if an error occurs
*/
CompletableFuture<MessageId> sendAsync(@Nullable String topic, T message, @Nullable Schema<T> schema)
CompletableFuture<MessageId> sendAsync(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema)
throws PulsarClientException;
/**
@@ -120,7 +122,7 @@ public interface PulsarOperations<T> {
* @param message the payload of the message
* @return the builder to configure and send the message
*/
SendMessageBuilder<T> newMessage(T message);
SendMessageBuilder<T> newMessage(@Nullable T message);
/**
* Builder that can be used to configure and send a message. Provides more options

62
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
View File

@@ -114,48 +114,51 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, 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<T> schema) throws PulsarClientException {
public MessageId send(@Nullable T message, @Nullable Schema<T> 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<T> schema) throws PulsarClientException {
public MessageId send(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema)
throws PulsarClientException {
return doSend(topic, message, schema, null, null, null);
}
@Override
public CompletableFuture<MessageId> sendAsync(T message) throws PulsarClientException {
public CompletableFuture<MessageId> sendAsync(@Nullable T message) throws PulsarClientException {
return doSendAsync(null, message, null, null, null, null);
}
@Override
public CompletableFuture<MessageId> sendAsync(T message, @Nullable Schema<T> schema) throws PulsarClientException {
public CompletableFuture<MessageId> sendAsync(@Nullable T message, @Nullable Schema<T> schema)
throws PulsarClientException {
return doSendAsync(null, message, schema, null, null, null);
}
@Override
public CompletableFuture<MessageId> sendAsync(@Nullable String topic, T message) throws PulsarClientException {
public CompletableFuture<MessageId> sendAsync(@Nullable String topic, @Nullable T message)
throws PulsarClientException {
return doSendAsync(topic, message, null, null, null, null);
}
@Override
public CompletableFuture<MessageId> sendAsync(@Nullable String topic, T message, @Nullable Schema<T> schema)
throws PulsarClientException {
public CompletableFuture<MessageId> sendAsync(@Nullable String topic, @Nullable T message,
@Nullable Schema<T> schema) throws PulsarClientException {
return doSendAsync(topic, message, schema, null, null, null);
}
@Override
public SendMessageBuilder<T> newMessage(T message) {
public SendMessageBuilder<T> newMessage(@Nullable T message) {
return new SendMessageBuilderImpl<>(this, message);
}
@@ -164,7 +167,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
this.beanName = beanName;
}
private MessageId doSend(@Nullable String topic, T message, @Nullable Schema<T> schema,
private MessageId doSend(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema,
@Nullable Collection<String> encryptionKeys,
@Nullable TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
@Nullable ProducerBuilderCustomizer<T> producerCustomizer) throws PulsarClientException {
@@ -177,13 +180,12 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
}
}
private CompletableFuture<MessageId> doSendAsync(@Nullable String topic, T message, @Nullable Schema<T> schema,
@Nullable Collection<String> encryptionKeys,
private CompletableFuture<MessageId> doSendAsync(@Nullable String topic, @Nullable T message,
@Nullable Schema<T> schema, @Nullable Collection<String> encryptionKeys,
@Nullable TypedMessageBuilderCustomizer<T> typedMessageBuilderCustomizer,
@Nullable ProducerBuilderCustomizer<T> 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<T> implements PulsarOperations<T>, BeanNameAware {
observation.start();
Producer<T> producer = prepareProducerForSend(topicName, message, schema, encryptionKeys,
producerCustomizer);
TypedMessageBuilder<T> messageBuilder = producer.newMessage().value(message);
if (typedMessageBuilderCustomizer != null) {
typedMessageBuilderCustomizer.customize(messageBuilder);
TypedMessageBuilder<T> 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<T> implements PulsarOperations<T>, BeanNameAware {
DefaultPulsarTemplateObservationConvention.INSTANCE, () -> senderContext, this.observationRegistry);
}
private Producer<T> prepareProducerForSend(@Nullable String topic, T message, @Nullable Schema<T> schema,
private Producer<T> prepareProducerForSend(@Nullable String topic, @Nullable T message, @Nullable Schema<T> schema,
@Nullable Collection<String> encryptionKeys, @Nullable ProducerBuilderCustomizer<T> 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<T> resolvedSchema = schema == null ? this.schemaResolver.resolveSchema(message).orElseThrow() : schema;
List<ProducerBuilderCustomizer<T>> customizers = new ArrayList<>();
if (!CollectionUtils.isEmpty(this.interceptors)) {
customizers.add(builder -> this.interceptors.forEach(builder::intercept));
@@ -241,13 +246,14 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, 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<T> implements SendMessageBuilder<T> {
private final PulsarTemplate<T> template;
@Nullable
private final T message;
@Nullable
@@ -265,7 +271,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T>, BeanNameAware {
@Nullable
private ProducerBuilderCustomizer<T> producerCustomizer;
SendMessageBuilderImpl(PulsarTemplate<T> template, T message) {
SendMessageBuilderImpl(PulsarTemplate<T> template, @Nullable T message) {
this.template = template;
this.message = message;
}

72
spring-pulsar/src/main/java/org/springframework/pulsar/core/Resolved.java Normal file
View File

@@ -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 <T> the resolved type
* @author Christophe Bornet
*/
public final class Resolved<T> {
@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 <T> Resolved<T> of(T value) {
return new Resolved<T>(value, null);
}
public static <T> Resolved<T> failed(String reason) {
return new Resolved<T>(null, new IllegalArgumentException(reason));
}
public static <T> Resolved<T> failed(RuntimeException e) {
return new Resolved<T>(null, e);
}
public Optional<T> get() {
return Optional.ofNullable(this.value);
}
public void ifResolved(Consumer<? super T> 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;
}
}

16
spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java
View File

@@ -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 <T> Schema<T> getSchema(T message) {
return getSchema(message.getClass());
default <T> Resolved<Schema<T>> 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 <T> Schema<T> getSchema(Class<?> messageType) {
return getSchema(messageType, true);
default <T> Resolved<Schema<T>> 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
<T> Schema<T> getSchema(Class<?> messageType, boolean returnDefault);
<T> Resolved<Schema<T>> 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
<T> Schema<T> getSchema(SchemaType schemaType, @Nullable ResolvableType messageType);
<T> Resolved<Schema<T>> resolveSchema(SchemaType schemaType, @Nullable ResolvableType messageType);
/**
* Callback interface that can be implemented by beans wishing to customize the schema

7
spring-pulsar/src/main/java/org/springframework/pulsar/core/TopicResolver.java
View File

@@ -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<String> resolveTopic(@Nullable String userSpecifiedTopic, Supplier<String> defaultTopicSupplier);
Resolved<String> resolveTopic(@Nullable String userSpecifiedTopic, Supplier<String> 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
*/
<T> Optional<String> resolveTopic(@Nullable String userSpecifiedTopic, T message,
<T> Resolved<String> resolveTopic(@Nullable String userSpecifiedTopic, @Nullable T message,
Supplier<String> 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<String> resolveTopic(@Nullable String userSpecifiedTopic, Class<?> messageType,
Resolved<String> resolveTopic(@Nullable String userSpecifiedTopic, @Nullable Class<?> messageType,
Supplier<String> defaultTopicSupplier);
}

52
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java
View File

@@ -98,7 +98,7 @@ class DefaultSchemaResolverTests {
@ParameterizedTest
@MethodSource("primitiveTypeMessagesProvider")
<T> void primitiveTypeMessages(T message, Schema<T> expectedSchema) {
assertThat(resolver.getSchema(message)).isEqualTo(expectedSchema);
assertThat(resolver.resolveSchema(message).orElseThrow()).isEqualTo(expectedSchema);
}
static Stream<Arguments> 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")
<T> void primitiveMessageTypes(Class<?> messageType, Schema<T> expectedSchema) {
assertThat(resolver.getSchema(messageType)).isEqualTo(expectedSchema);
assertThat(resolver.resolveSchema(messageType).orElseThrow()).isEqualTo(expectedSchema);
}
static Stream<Arguments> 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")
<T> void primitiveSchemas(SchemaType schemaType, Schema<T> expectedSchema) {
assertThat(resolver.getSchema(schemaType, null)).isEqualTo(expectedSchema);
assertThat(resolver.resolveSchema(schemaType, null).orElseThrow()).isEqualTo(expectedSchema);
}
static Stream<Arguments> 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);

20
spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java
View File

@@ -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<Arguments> resolveNoMessageInfoProvider() {
@@ -77,7 +76,8 @@ class DefaultTopicResolverTests {
@MethodSource("resolveByMessageInstanceProvider")
<T> 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<Arguments> 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<Arguments> 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<String> actual, @Nullable String expectedTopic) {
if (expectedTopic == null) {
assertThat(actual).isEmpty();
}
else {
assertThat(actual).hasValue(expectedTopic);
}
}
@Nested
class TopicMappingsAPI {

445
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java
View File

@@ -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<String> messageCustomizer = null;
if (testArgs.messageCustomizer) {
messageCustomizer = (mb) -> mb.key("foo-key");
}
ProducerBuilderCustomizer<String> producerCustomizer = null;
if (testArgs.producerCustomizer) {
producerCustomizer = (pb) -> pb.producerName("foo-producer");
}
try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
.build()) {
try (Consumer<String> consumer = client.newConsumer(Schema.STRING).topic(topic)
.subscriptionName(subscription).subscribe()) {
Map<String, Object> producerConfig = testArgs.explicitTopic ? Collections.emptyMap()
: Collections.singletonMap("topicName", topic);
PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
producerConfig);
PulsarTemplate<String> 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<String> 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<Message<String>> receiveMsgFuture = consumer.receiveAsync();
Message<String> 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<Arguments> 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<PulsarTemplate<String>> sendFunction,
Boolean withDefaultTopic, String expectedValue) throws Exception {
sendAndConsume(sendFunction, testName, Schema.STRING, expectedValue, withDefaultTopic);
}
static Stream<Arguments> sendMessageTestProvider() {
String message = "test-message";
return Stream.of(
// Simple send sync
arguments("simpleSendWithDefaultTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.send(message), true, message),
arguments("simpleSendWithTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.send("simpleSendWithTopic",
message),
false, message),
arguments("simpleSendWithDefaultTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.send(message, Schema.STRING),
true, message),
arguments("simpleSendWithTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.send("simpleSendWithTopicAndSchema", message, Schema.STRING),
false, message),
arguments("simpleSendNullWithTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.send("simpleSendNullWithTopicAndSchema", null, Schema.STRING),
false, null),
// Simple send async
arguments("simpleSendAsyncWithDefaultTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.sendAsync(message).get(3,
TimeUnit.SECONDS),
true, message),
arguments("simpleSendAsyncWithTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.sendAsync("simpleSendAsyncWithTopic", message).get(3, TimeUnit.SECONDS),
false, message),
arguments("simpleSendAsyncWithDefaultTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.sendAsync(message, Schema.STRING).get(3, TimeUnit.SECONDS),
true, message),
arguments("simpleSendAsyncWithTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.sendAsync("simpleSendAsyncWithTopicAndSchema", message, Schema.STRING)
.get(3, TimeUnit.SECONDS),
false, message),
arguments("simpleSendAsyncNullWithTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.sendAsync("simpleSendAsyncNullWithTopicAndSchema", null, Schema.STRING)
.get(3, TimeUnit.SECONDS),
false, null),
// Fluent send
arguments("fluentSendWithDefaultTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.newMessage(message).send(),
true, message),
arguments("fluentSendWithTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.newMessage(message)
.withTopic("fluentSendWithTopic").send(),
false, message),
arguments("fluentSendWithDefaultTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.newMessage(message)
.withSchema(Schema.STRING).send(),
true, message),
arguments("fluentSendNullWithTopicAndSchema",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.newMessage(null)
.withSchema(Schema.STRING).withTopic("fluentSendNullWithTopicAndSchema").send(),
false, null),
arguments("fluentSendAsync", (ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.newMessage(message).sendAsync().get(3, TimeUnit.SECONDS), true, message)
);
}
@Test
void sendMessageWithMessageCustomizer() throws Exception {
ThrowingConsumer<PulsarTemplate<String>> sendFunction = (template) -> template.newMessage("test-message")
.withMessageCustomizer((mb) -> mb.key("test-key")).send();
Message<String> msg = sendAndConsume(sendFunction, "sendMessageWithMessageCustomizer", Schema.STRING,
"test-message", true);
assertThat(msg.getKey()).isEqualTo("test-key");
}
@Test
void sendMessageWithSenderCustomizer() throws Exception {
ThrowingConsumer<PulsarTemplate<String>> sendFunction = (template) -> template.newMessage("test-message")
.withProducerCustomizer((sb) -> sb.producerName("test-producer")).send();
Message<String> 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<ProducerInterceptor> interceptors) throws Exception {
try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl())
.build()) {
PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors);
pulsarTemplate.send("test-interceptor");
for (ProducerInterceptor interceptor : interceptors) {
verify(interceptor, atLeastOnce()).eligible(any(Message.class));
}
PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<String> 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<Foo> consumer = client.newConsumer(Schema.AVRO(Foo.class)).topic(topic)
.subscriptionName("ptt-specificSchema-subs").subscribe()) {
PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<Foo> 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<PulsarTemplate<Foo>> 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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
.subscriptionName("ptt-nospecificSchema-subs").subscribe()) {
PulsarProducerFactory<Foo> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<Foo> 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<PulsarTemplate<Foo>> 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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
.subscriptionName("ptt-schemaInferred-subs").subscribe()) {
PulsarProducerFactory<Foo> 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<Foo> 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<Foo> 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<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(),
schemaResolver, new DefaultTopicResolver(), null, null);
Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
ThrowingConsumer<PulsarTemplate<Foo>> 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<Foo> consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic)
.subscriptionName("ptt-topicInferred-subs").subscribe()) {
PulsarProducerFactory<Foo> 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<Foo> 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<Foo> 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<Foo> pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(),
new DefaultSchemaResolver(), topicResolver, null, null);
Foo foo = new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID());
ThrowingConsumer<PulsarTemplate<Foo>> 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<String> producerFactory = mock(PulsarProducerFactory.class);
when(producerFactory.createProducer(Schema.STRING, topic, Set.of("key"), new ArrayList<>()))
.thenReturn(client.newProducer(Schema.STRING).topic(topic).create());
PulsarTemplate<String> 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<String> producerFactory = mock(PulsarProducerFactory.class);
when(producerFactory.createProducer(Schema.STRING, topic, Set.of("key"), new ArrayList<>()))
.thenReturn(client.newProducer(Schema.STRING).topic(topic).create());
PulsarTemplate<String> 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<PulsarTemplate<String>> sendFunction) {
PulsarProducerFactory<String> senderFactory = new DefaultPulsarProducerFactory<>(client, new HashMap<>());
PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(senderFactory);
assertThatIllegalArgumentException().isThrownBy(() -> sendFunction.accept(pulsarTemplate));
}
private final boolean simpleApi;
static Stream<Arguments> sendMessageFailedTestProvider() {
String message = "test-message";
return Stream.of(
arguments("sendWithoutTopic",
(ThrowingConsumer<PulsarTemplate<String>>) (template) -> template.send(message)),
arguments("sendNullWithoutSchema", (ThrowingConsumer<PulsarTemplate<String>>) (template) -> template
.send("sendNullWithoutSchema", (String) null)));
}
private boolean async;
@Test
void sendNullWithDefaultTopicFails() {
HashMap<String, Object> config = new HashMap<>();
config.put("topicName", "sendNullWithDefaultTopicFails");
PulsarProducerFactory<String> senderFactory = new DefaultPulsarProducerFactory<>(client, config);
PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(senderFactory);
assertThatIllegalArgumentException().isThrownBy(() -> pulsarTemplate.send(null, Schema.STRING));
}
private boolean explicitTopic;
@Test
void sendWithoutSchemaFails() {
PulsarProducerFactory<Foo> senderFactory = new DefaultPulsarProducerFactory<>(client, new HashMap<>());
PulsarTemplate<Foo> 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 <T> Message<T> sendAndConsume(ThrowingConsumer<PulsarTemplate<T>> sendFunction, String topic,
Schema<T> schema, T expectedValue, Boolean withDefaultTopic) throws Exception {
Map<String, Object> config = new HashMap<>();
if (withDefaultTopic) {
config.put("topicName", topic);
}
PulsarProducerFactory<T> senderFactory = new DefaultPulsarProducerFactory<>(client, config);
PulsarTemplate<T> pulsarTemplate = new PulsarTemplate<>(senderFactory);
return sendAndConsume(pulsarTemplate, sendFunction, topic, schema, expectedValue);
}
static SendTestArgs simple() {
return new SendTestArgs(true);
private <T> Message<T> sendAndConsume(PulsarTemplate<T> template, ThrowingConsumer<PulsarTemplate<T>> sendFunction,
String topic, Schema<T> schema, T expectedValue) throws Exception {
try (org.apache.pulsar.client.api.Consumer<T> consumer = client.newConsumer(schema).topic(topic)
.subscriptionName(topic + "-sub").subscribe()) {
sendFunction.accept(template);
Message<T> 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 {