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

Allow custom object mapper for messages (#755)

Browse Source 
This commit adds support for a user-provided Jackson ObjectMapper to be
used when de/serializing JSON messages.

Additionally, adds Gradle test fixtures to the spring-pulsar module and
deprecates the UserRecord and UserPojo in spring-pulsar-test in favor
of their equivalent in the test fixture.

See #723
This commit is contained in:
Chris Bono
2024-07-29 21:57:27 -05:00
committed by GitHub
parent 3fcb22de91
commit 70e9e60f39
32 changed files with 864 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-pulsar-docs/src/main/antora/modules/ROOT/nav.adoc
View File

@@ -21,6 +21,7 @@
*** xref:reference/reactive-pulsar/reactive-message-consumption.adoc[]
*** xref:reference/tombstones-reactive.adoc[]
** xref:reference/topic-resolution.adoc[]
** xref:reference/custom-object-mapper.adoc[]
** xref:reference/pulsar-admin.adoc[]
** xref:reference/pulsar-function.adoc[]
** xref:reference/observability.adoc[]

48
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/custom-object-mapper.adoc Normal file
View File

@@ -0,0 +1,48 @@
[[custom-object-mapper]]
= Custom Object Mapper
include::../attributes/attributes.adoc[]
Pulsar uses an internal Jackson `ObjectMapper` when de/serializing JSON messages.
If you instead want to provide your own object mapper instance, you can register a `SchemaResolverCustomizer` and set your mapper on the `DefaultSchemaResolver` as follows:
[source,java,indent=0,subs="verbatim"]
----
@Bean
SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
return (DefaultSchemaResolver schemaResolver) -> {
var myObjectMapper = obtainMyObjectMapper();
schemaResolver.setObjectMapper(myObjectMapper);
};
}
----
This results in your object mapper being used to de/serialize all JSON messages that go through the schema resolution process (i.e. in cases where you do not pass a schema in directly when producing/consuming messages).
Under the hood, the resolver creates a special JSON schema which leverages the custom mapper and is used as the schema for all resolved JSON messages.
If you need to pass schema instances directly you can use the `JSONSchemaUtil` to create schemas that respect the custom mapper.
The following example shows how to do this when sending a message with the `PulsarTemplate` variant that takes a schema parameter:
[source,java,indent=0,subs="verbatim"]
----
void sendMessage(PulsarTemplate<MyPojo> template, MyPojo toSend) {
var myObjectMapper = obtainMyObjectMapper();
var schema = JSONSchemaUtil.schemaForTypeWithObjectMapper(MyPojo.class, myObjectMapper);
template.send(toSend, schema);
}
----
[CAUTION]
====
Pulsar configures its default object mapper in a particular way.
Unless you have a specific reason to not do so, it is highly recommended that you configure your mapper with these same options as follows:
[source,java,indent=0,subs="verbatim"]
----
myObjectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
myObjectMapper.configure(DeserializationFeature.READ_UNKNOWN_ENUM_VALUES_AS_NULL, false);
myObjectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
----
====
NOTE: A later version of the framework may instead provide a customizer that operates on the default mapper rather than requiring a separate instance.

10
spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc
View File

@@ -1,5 +1,15 @@
= What's new?
[[what-s-new-in-1-2-since-1-1]]
== What's New in 1.2 Since 1.1
:page-section-summary-toc: 1
This section covers the changes made from version 1.1 to version 1.2.
=== Custom Object Mapper
You can provide your own Jackson `ObjectMapper` that Pulsar will use when producing and consuming JSON messages.
See xref:./reference/custom-object-mapper.adoc[Custom Object Mapper] for more details.
[[what-s-new-in-1-1-since-1-0]]
== What's New in 1.1 Since 1.0
:page-section-summary-toc: 1

1
spring-pulsar-reactive/spring-pulsar-reactive.gradle
View File

@@ -33,6 +33,7 @@ dependencies {
optional libs.json.path
testImplementation project(':spring-pulsar-test')
testImplementation(testFixtures(project(":spring-pulsar")))
testRuntimeOnly libs.logback.classic
testImplementation 'io.projectreactor:reactor-test'
testImplementation 'org.assertj:assertj-core'

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

@@ -47,8 +47,10 @@ import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.lang.Nullable;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultTopicResolver;
import org.springframework.pulsar.core.JSONSchemaUtil;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.model.json.UserRecordObjectMapper;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.util.function.ThrowingConsumer;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -332,6 +334,25 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport {
}
@Nested
class CustomObjectMapperTests {
@Test
void sendWithCustomJsonSchema() throws Exception {
// Prepare the schema with custom object mapper
var objectMapper = UserRecordObjectMapper.withSer();
var schema = JSONSchemaUtil.schemaForTypeWithObjectMapper(UserRecord.class, objectMapper);
var topic = "rptt-custom-object-mapper-topic";
var user = new UserRecord("elFoo", 21);
// serializer adds '-ser' to name and 10 to age
var expectedUser = new UserRecord("elFoo-ser", 31);
ThrowingConsumer<ReactivePulsarTemplate<UserRecord>> sendFunction = (
template) -> template.send(topic, user, schema).subscribe();
sendAndConsume(sendFunction, topic, schema, expectedUser, false);
}
}
public static class Foo {
private String foo;

57
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainerTests.java
View File

@@ -22,6 +22,7 @@ import java.time.Duration;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.PulsarClient;
@@ -37,10 +38,13 @@ import org.assertj.core.api.InstanceOfAssertFactories;
import org.junit.jupiter.api.Test;
import org.springframework.core.log.LogAccessor;
import org.springframework.pulsar.core.JSONSchemaUtil;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory;
import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory;
import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.model.json.UserRecordObjectMapper;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import reactor.core.publisher.Flux;
@@ -306,28 +310,69 @@ class DefaultReactivePulsarMessageListenerContainerTests implements PulsarTestCo
}
}
@Test
void oneByOneMessageHandlerWithCustomObjectMapper() throws Exception {
var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
ReactivePulsarMessageListenerContainer<UserRecord> container = null;
try {
// Prepare the schema with custom object mapper
var objectMapper = UserRecordObjectMapper.withDeser();
var schema = JSONSchemaUtil.schemaForTypeWithObjectMapper(UserRecord.class, objectMapper);
var reactivePulsarClient = AdaptedReactivePulsarClientFactory.create(pulsarClient);
var topic = topicNameForTest("com-topic");
var consumerFactory = createAndPrepareConsumerFactory(topic, schema, reactivePulsarClient);
var containerProperties = new ReactivePulsarContainerProperties<UserRecord>();
containerProperties.setSchema(schema);
var latch = new CountDownLatch(1);
AtomicReference<UserRecord> consumedRecordRef = new AtomicReference<>();
containerProperties.setMessageHandler((ReactivePulsarOneByOneMessageHandler<UserRecord>) (msg) -> {
consumedRecordRef.set(msg.getValue());
return Mono.fromRunnable(latch::countDown);
});
container = new DefaultReactivePulsarMessageListenerContainer<>(consumerFactory, containerProperties);
container.start();
var sentUserRecord = new UserRecord("person", 51);
// deser adds '-deser' to name and 5 to age
var expectedConsumedUser = new UserRecord("person-deser", 56);
createPulsarTemplate(topic, reactivePulsarClient).send(sentUserRecord).subscribe();
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(consumedRecordRef).hasValue(expectedConsumedUser);
}
finally {
safeStopContainer(container);
pulsarClient.close();
}
}
private String topicNameForTest(String suffix) {
return "drpmlct-" + suffix;
}
private DefaultReactivePulsarConsumerFactory<String> createAndPrepareConsumerFactory(String topic,
ReactivePulsarClient reactivePulsarClient) {
ReactiveMessageConsumerBuilderCustomizer<String> defaultConfig = (builder) -> {
return this.createAndPrepareConsumerFactory(topic, Schema.STRING, reactivePulsarClient);
}
private <T> DefaultReactivePulsarConsumerFactory<T> createAndPrepareConsumerFactory(String topic, Schema<T> schema,
ReactivePulsarClient reactivePulsarClient) {
ReactiveMessageConsumerBuilderCustomizer<T> defaultConfig = (builder) -> {
builder.topic(topic);
builder.subscriptionName(topic + "-sub");
};
var consumerFactory = new DefaultReactivePulsarConsumerFactory<>(reactivePulsarClient, List.of(defaultConfig));
var consumerFactory = new DefaultReactivePulsarConsumerFactory<T>(reactivePulsarClient, List.of(defaultConfig));
// Ensure subscription is created
consumerFactory.createConsumer(Schema.STRING).consumeNothing().block(Duration.ofSeconds(5));
consumerFactory.createConsumer(schema).consumeNothing().block(Duration.ofSeconds(5));
return consumerFactory;
}
private ReactivePulsarTemplate<String> createPulsarTemplate(String topic,
private <T> ReactivePulsarTemplate<T> createPulsarTemplate(String topic,
ReactivePulsarClient reactivePulsarClient) {
var producerFactory = DefaultReactivePulsarSenderFactory.<String>builderFor(reactivePulsarClient)
var producerFactory = DefaultReactivePulsarSenderFactory.<T>builderFor(reactivePulsarClient)
.withDefaultTopic(topic)
.build();
return new ReactivePulsarTemplate<>(producerFactory);
return new ReactivePulsarTemplate<T>(producerFactory);
}
private void safeStopContainer(ReactivePulsarMessageListenerContainer<?> container) {

4
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerAutoConsumeSchemaTests.java
View File

@@ -48,8 +48,8 @@ import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer;
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerAutoConsumeSchemaTests.ReactivePulsarListenerAutoConsumeSchemaTestsConfig;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.pulsar.test.model.UserPojo;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
import reactor.core.publisher.Mono;

4
spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java
View File

@@ -74,8 +74,8 @@ import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.
import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
import org.springframework.pulsar.reactive.support.MessageUtils;
import org.springframework.pulsar.support.PulsarHeaders;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.pulsar.test.model.UserPojo;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.util.ObjectUtils;

5
spring-pulsar-sample-apps/sample-imperative-produce-consume/build.gradle
View File

@@ -21,7 +21,10 @@ ext['pulsar.version'] = "${pulsarVersion}"
dependencies {
implementation 'org.springframework.boot:spring-boot-starter-pulsar'
developmentOnly 'org.springframework.boot:spring-boot-docker-compose'
testImplementation project(':spring-pulsar-test')
// temporary until JsonSchemaUtil published
implementation project(':spring-pulsar')
implementation(testFixtures(project(":spring-pulsar")))
implementation project(':spring-pulsar-test')
testRuntimeOnly 'ch.qos.logback:logback-classic'
testImplementation "org.springframework.boot:spring-boot-starter-test"
testImplementation "org.springframework.boot:spring-boot-testcontainers"

44
spring-pulsar-sample-apps/sample-imperative-produce-consume/src/main/java/com/example/ImperativeProduceAndConsumeApp.java
View File

@@ -27,8 +27,12 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.core.PulsarTopic;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.model.json.UserRecordObjectMapper;
@SpringBootApplication
public class ImperativeProduceAndConsumeApp {
@@ -55,7 +59,7 @@ public class ImperativeProduceAndConsumeApp {
};
}
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub")
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC + "-sub")
void consumePrimitiveMessagesFromPulsarTopic(String msg) {
LOG.info("++++++CONSUME {}------", msg);
}
@@ -79,7 +83,7 @@ public class ImperativeProduceAndConsumeApp {
};
}
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub")
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC + "-sub")
void consumeComplexMessagesFromPulsarTopic(Foo msg) {
LOG.info("++++++CONSUME {}------", msg);
}
@@ -108,7 +112,7 @@ public class ImperativeProduceAndConsumeApp {
};
}
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub")
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC + "-sub")
void consumePartitionedMessagesFromPulsarTopic(String msg) {
LOG.info("++++++CONSUME {}------", msg);
}
@@ -132,7 +136,7 @@ public class ImperativeProduceAndConsumeApp {
};
}
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC+"-sub", batch = true)
@PulsarListener(topics = TOPIC, subscriptionName = TOPIC + "-sub", batch = true)
void consumeBatchMessagesFromPulsarTopic(List<Foo> messages) {
messages.forEach((msg) -> LOG.info("++++++CONSUME {}------", msg));
}
@@ -162,6 +166,38 @@ public class ImperativeProduceAndConsumeApp {
}
@Configuration(proxyBeanMethods = false)
static class ProduceConsumeCustomObjectMapper {
private static final String TOPIC = "produce-consume-custom-object-mapper";
@Bean
SchemaResolver.SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
return (DefaultSchemaResolver schemaResolver) -> {
var objectMapper = UserRecordObjectMapper.withSerAndDeser();
schemaResolver.setObjectMapper(objectMapper);
};
}
@Bean
ApplicationRunner sendWithCustomObjectMapper(PulsarTemplate<UserRecord> template) {
return (args) -> {
for (int i = 0; i < 10; i++) {
var user = new UserRecord("user-" + i, 30);
template.send(TOPIC, user);
LOG.info("++++++PRODUCE {}------", user);
}
};
}
@PulsarListener(topics = TOPIC)
void consumeWithCustomObjectMapper(UserRecord user) {
LOG.info("++++++CONSUME {}------", user);
}
}
record Foo(String name, Integer value) {
}

42
spring-pulsar-sample-apps/sample-imperative-produce-consume/src/test/java/com/example/ImperativeProduceAndConsumeAppTests.java
View File

@@ -16,14 +16,13 @@
package com.example;
import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import java.util.stream.IntStream;
import com.example.ImperativeProduceAndConsumeApp.Bar;
import com.example.ImperativeProduceAndConsumeApp.Foo;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -31,11 +30,13 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
import com.example.ImperativeProduceAndConsumeApp.Bar;
import com.example.ImperativeProduceAndConsumeApp.Foo;
@SpringBootTest
@ExtendWith(OutputCaptureExtension.class)
@@ -49,7 +50,7 @@ class ImperativeProduceAndConsumeAppTests implements PulsarTestContainerSupport
@Test
void produceConsumeWithPrimitiveMessageType(CapturedOutput output) {
verifyProduceConsume(output,10, (i) -> "ProduceConsumeWithPrimitiveMessageType:" + i);
verifyProduceConsume(output, 10, (i) -> "ProduceConsumeWithPrimitiveMessageType:" + i);
}
@Test
@@ -60,27 +61,42 @@ class ImperativeProduceAndConsumeAppTests implements PulsarTestContainerSupport
@Test
void produceConsumeWithPartitions(CapturedOutput output) {
verifyProduceConsume(output,10, (i) -> "ProduceConsumeWithPartitions:" + i);
verifyProduceConsume(output, 10, (i) -> "ProduceConsumeWithPartitions:" + i);
}
@Test
void produceConsumeBatchListener(CapturedOutput output) {
verifyProduceConsume(output,100,
(i) -> new Foo("ProduceConsumeBatchListener", i));
verifyProduceConsume(output, 100, (i) -> new Foo("ProduceConsumeBatchListener", i));
}
@Test
void produceConsumeDefaultMappings(CapturedOutput output) {
verifyProduceConsume(output,10, (i) -> new Bar("ProduceConsumeDefaultMappings:" + i));
verifyProduceConsume(output, 10, (i) -> new Bar("ProduceConsumeDefaultMappings:" + i));
}
@Test
void produceConsumeCustomObjectMapper(CapturedOutput output) {
// base age is 30 then ser adds 10 then deser adds 5
var expectedAge = 30 + 10 + 5;
verifyProduceConsume(output, 10,
(i) -> new UserRecord("user-%d".formatted(i), 30),
(i) -> new UserRecord("user-%d-ser-deser".formatted(i), expectedAge));
}
private void verifyProduceConsume(CapturedOutput output, int numExpectedMessages,
Function<Integer, Object> expectedMessageFactory) {
List < String > expectedOutput = new ArrayList<>();
this.verifyProduceConsume(output, numExpectedMessages, expectedMessageFactory, expectedMessageFactory);
}
private void verifyProduceConsume(CapturedOutput output, int numExpectedMessages,
Function<Integer, Object> expectedProducedMessageFactory,
Function<Integer, Object> expectedConsumedMessageFactory) {
var expectedOutput = new ArrayList<String>();
IntStream.range(0, numExpectedMessages).forEachOrdered((i) -> {
var msg = expectedMessageFactory.apply(i);
expectedOutput.add("++++++PRODUCE %s------".formatted(msg));
expectedOutput.add("++++++CONSUME %s------".formatted(msg));
var expectedProducedMsg = expectedProducedMessageFactory.apply(i);
var expectedConsumedMsg = expectedConsumedMessageFactory.apply(i);
expectedOutput.add("++++++PRODUCE %s------".formatted(expectedProducedMsg));
expectedOutput.add("++++++CONSUME %s------".formatted(expectedConsumedMsg));
});
Awaitility.waitAtMost(Duration.ofSeconds(15))
.untilAsserted(() -> assertThat(output).contains(expectedOutput));

5
spring-pulsar-sample-apps/sample-reactive/build.gradle
View File

@@ -23,7 +23,10 @@ ext['pulsar-reactive.version'] = "${pulsarReactiveVersion}"
dependencies {
implementation "org.springframework.boot:spring-boot-starter-pulsar-reactive"
developmentOnly 'org.springframework.boot:spring-boot-docker-compose'
testImplementation project(':spring-pulsar-test')
// temporary until JsonSchemaUtil published
implementation project(':spring-pulsar')
implementation(testFixtures(project(":spring-pulsar")))
implementation project(':spring-pulsar-test')
testRuntimeOnly 'ch.qos.logback:logback-classic'
testImplementation "org.springframework.boot:spring-boot-starter-test"
testImplementation "org.springframework.boot:spring-boot-testcontainers"

34
spring-pulsar-sample-apps/sample-reactive/src/main/java/com/example/ReactiveSpringPulsarBootApp.java
View File

@@ -31,9 +31,13 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.SchemaResolver;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListener;
import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListenerMessageConsumerBuilderCustomizer;
import org.springframework.pulsar.reactive.core.ReactivePulsarTemplate;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.model.json.UserRecordObjectMapper;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -125,6 +129,36 @@ public class ReactiveSpringPulsarBootApp {
}
@Configuration(proxyBeanMethods = false)
static class ProduceConsumeCustomObjectMapper {
private static final String TOPIC = "sample-reactive-custom-object-mapper";
@Bean
SchemaResolver.SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
return (DefaultSchemaResolver schemaResolver) -> {
var objectMapper = UserRecordObjectMapper.withSerAndDeser();
schemaResolver.setObjectMapper(objectMapper);
};
}
@Bean
ApplicationRunner sendWithCustomObjectMapper(ReactivePulsarTemplate<UserRecord> template) {
return (args) -> Flux.range(0, 10)
.map((i) -> MessageSpec.of(new UserRecord("user-" + i, 30)))
.as(messages -> template.send(TOPIC, messages))
.doOnNext((msr) -> LOG.info("++++++PRODUCE {}------", msr.getMessageSpec().getValue()))
.subscribe();
}
@ReactivePulsarListener(topics = TOPIC, consumerCustomizer = "subscriptionInitialPositionEarliest")
public Mono<Void> listenSimple(UserRecord user) {
LOG.info("++++++CONSUME {}------", user);
return Mono.empty();
}
}
record Foo(String foo, String bar) {
}

31
spring-pulsar-sample-apps/sample-reactive/src/test/java/com/example/ReactiveSpringPulsarBootAppTests.java
View File

@@ -16,13 +16,14 @@
package com.example;
import static org.assertj.core.api.Assertions.assertThat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Function;
import java.util.stream.IntStream;
import com.example.ReactiveSpringPulsarBootApp.Foo;
import org.awaitility.Awaitility;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -30,11 +31,12 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.system.CapturedOutput;
import org.springframework.boot.test.system.OutputCaptureExtension;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import static org.assertj.core.api.Assertions.assertThat;
import com.example.ReactiveSpringPulsarBootApp.Foo;
@SpringBootTest
@ExtendWith(OutputCaptureExtension.class)
@@ -61,13 +63,30 @@ class ReactiveSpringPulsarBootAppTests implements PulsarTestContainerSupport {
verifyProduceConsume(output,10, (i) -> "ReactiveTemplateWithImperativeListener:" + i);
}
@Test
void produceConsumeCustomObjectMapper(CapturedOutput output) {
// base age is 30 then ser adds 10 then deser adds 5
var expectedAge = 30 + 10 + 5;
verifyProduceConsume(output, 10,
(i) -> new UserRecord("user-%d".formatted(i), 30),
(i) -> new UserRecord("user-%d-ser-deser".formatted(i), expectedAge));
}
private void verifyProduceConsume(CapturedOutput output, int numExpectedMessages,
Function<Integer, Object> expectedMessageFactory) {
List < String > expectedOutput = new ArrayList<>();
this.verifyProduceConsume(output, numExpectedMessages, expectedMessageFactory, expectedMessageFactory);
}
private void verifyProduceConsume(CapturedOutput output, int numExpectedMessages,
Function<Integer, Object> expectedProducedMessageFactory,
Function<Integer, Object> expectedConsumedMessageFactory) {
List<String> expectedOutput = new ArrayList<>();
IntStream.range(0, numExpectedMessages).forEachOrdered((i) -> {
var msg = expectedMessageFactory.apply(i);
expectedOutput.add("++++++PRODUCE %s------".formatted(msg));
expectedOutput.add("++++++CONSUME %s------".formatted(msg));
var expectedProducedMsg = expectedProducedMessageFactory.apply(i);
var expectedConsumedMsg = expectedConsumedMessageFactory.apply(i);
expectedOutput.add("++++++PRODUCE %s------".formatted(expectedProducedMsg));
expectedOutput.add("++++++CONSUME %s------".formatted(expectedConsumedMsg));
});
Awaitility.waitAtMost(Duration.ofSeconds(15))
.untilAsserted(() -> assertThat(output).contains(expectedOutput));

4
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/model/UserPojo.java
View File

@@ -23,7 +23,11 @@ import java.util.Objects;
* <p>
* <b>WARN</b> Do not convert this to a Record as this is used for Avro tests and Avro
* does not work well w/ records yet.
*
* @deprecated this class is replaced with Gradle test fixtures and is only meant to be
* used internally.
*/
@Deprecated(since = "1.2.0", forRemoval = true)
public class UserPojo {
private String name;

3
spring-pulsar-test/src/main/java/org/springframework/pulsar/test/support/model/UserRecord.java
View File

@@ -21,6 +21,9 @@ package org.springframework.pulsar.test.support.model;
*
* @param name the user's name
* @param age the user's age
* @deprecated this class is replaced with Gradle test fixtures and is only meant to be
* used internally.
*/
@Deprecated(since = "1.2.0", forRemoval = true)
public record UserRecord(String name, int age) {
}

3
spring-pulsar/spring-pulsar.gradle
View File

@@ -1,5 +1,6 @@
plugins {
id 'org.springframework.pulsar.spring-module'
id 'java-test-fixtures'
}
description = 'Spring Pulsar Core'
@@ -51,4 +52,6 @@ dependencies {
testImplementation "org.testcontainers:mysql"
testImplementation 'mysql:mysql-connector-java:8.0.33'
// Used by UserRecordDe/serializer in test fixtures
testFixturesApi 'com.fasterxml.jackson.core:jackson-databind'
}

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

@@ -44,6 +44,8 @@ import org.springframework.lang.Nullable;
import org.springframework.pulsar.annotation.PulsarMessage;
import org.springframework.util.Assert;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* Default schema resolver capable of handling basic message types.
*
@@ -96,6 +98,12 @@ public class DefaultSchemaResolver implements SchemaResolver {
private boolean usePulsarMessageAnnotations = true;
private ObjectMapper objectMapper;
public void setObjectMapper(ObjectMapper objectMapper) {
this.objectMapper = objectMapper;
}
/**
* Sets whether to inspect message classes for the
* {@link PulsarMessage @PulsarMessage} annotation during schema resolution.
@@ -168,7 +176,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
if (schema == null && returnDefault) {
if (messageClass != null) {
try {
return Schema.JSON(messageClass);
return jsonSchemaForMessageType(messageClass);
}
catch (Exception e) {
this.logger.debug(e, "Failed to create JSON schema for " + messageClass.getName());
@@ -224,7 +232,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
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 JSON -> jsonSchemaForMessageType(requireNonNullMessageType(schemaType, messageType));
case AVRO -> AvroSchema.of(requireNonNullMessageType(schemaType, messageType));
case PROTOBUF -> {
// WARN! Leave GeneratedMessageV3 fully-qualified as the dependency is
@@ -255,6 +263,13 @@ public class DefaultSchemaResolver implements SchemaResolver {
}
}
private <T> JSONSchema<T> jsonSchemaForMessageType(Class<T> messageType) {
if (this.objectMapper != null) {
return JSONSchemaUtil.schemaForTypeWithObjectMapper(messageType, this.objectMapper);
}
return JSONSchema.of(messageType);
}
@Nullable
private Class<?> requireNonNullMessageType(SchemaType schemaType, @Nullable ResolvableType messageType) {
return Objects.requireNonNull(messageType, "messageType must be specified for " + schemaType.name())

157
spring-pulsar/src/main/java/org/springframework/pulsar/core/JSONSchemaUtil.java Normal file
View File

@@ -0,0 +1,157 @@
/*
* Copyright 2023-2024 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.io.IOException;
import java.io.InputStream;
import java.util.function.Consumer;
import org.apache.pulsar.client.api.SchemaSerializationException;
import org.apache.pulsar.client.api.schema.SchemaDefinitionBuilder;
import org.apache.pulsar.client.api.schema.SchemaReader;
import org.apache.pulsar.client.api.schema.SchemaWriter;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.apache.pulsar.client.impl.schema.SchemaDefinitionBuilderImpl;
import org.springframework.core.log.LogAccessor;
import org.springframework.util.Assert;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
/**
* Factory to create schema definition {@link SchemaDefinitionBuilder builders} that
* provide schema definitions that use custom object mappers when de/serializing objects.
*
* @author Chris Bono
* @since 1.2.0
*/
public interface JSONSchemaUtil {
/**
* Create a new JSON schema that uses the provided object mapper to de/serialize
* objects of the specified type.
* @param objectType the type of objects the resulting schema represents
* @param objectMapper the mapper used to read and write objects from JSON
* @param <T> the type of objects the resulting schema represents
* @return the schema instance
*/
static <T> JSONSchema<T> schemaForTypeWithObjectMapper(Class<T> objectType, ObjectMapper objectMapper) {
return JSONSchemaUtil.schemaForTypeWithObjectMapper(objectType, objectMapper, (b) -> {
});
}
/**
* Create a new JSON schema that uses the provided object mapper to de/serialize
* objects of the specified type.
* @param objectType the type of objects the resulting schema represents
* @param objectMapper the mapper used to read and write objects from JSON
* @param schemaDefinitionBuilderCustomizer the schema definition builder customizer
* @param <T> the type of objects the resulting schema represents
* @return the schema instance
*/
static <T> JSONSchema<T> schemaForTypeWithObjectMapper(Class<T> objectType, ObjectMapper objectMapper,
Consumer<SchemaDefinitionBuilder<T>> schemaDefinitionBuilderCustomizer) {
var reader = new CustomJacksonJsonReader<>(objectMapper, objectType);
var writer = new CustomJacksonJsonWriter<T>(objectMapper);
var schemaDefinitionBuilder = new SchemaDefinitionBuilderImpl<T>().withPojo(objectType)
.withSchemaReader(reader)
.withSchemaWriter(writer);
schemaDefinitionBuilderCustomizer.accept(schemaDefinitionBuilder);
return JSONSchema.of(schemaDefinitionBuilder.build());
}
/**
* Reader implementation for reading objects from JSON using a custom
* {@code ObjectMapper}.
*
* @param <T> object type to read
*/
class CustomJacksonJsonReader<T> implements SchemaReader<T> {
private static final LogAccessor LOG = new LogAccessor(CustomJacksonJsonReader.class);
private final ObjectReader objectReader;
private final Class<T> objectType;
CustomJacksonJsonReader(ObjectMapper objectMapper, Class<T> objectType) {
Assert.notNull(objectMapper, "objectMapper must not be null");
Assert.notNull(objectType, "objectType must not be null");
this.objectReader = objectMapper.readerFor(objectType);
this.objectType = objectType;
}
@Override
public T read(byte[] bytes, int offset, int length) {
try {
return this.objectReader.readValue(bytes, offset, length);
}
catch (IOException e) {
throw new SchemaSerializationException(e);
}
}
@Override
public T read(InputStream inputStream) {
try {
return this.objectReader.readValue(inputStream, this.objectType);
}
catch (IOException e) {
throw new SchemaSerializationException(e);
}
finally {
try {
inputStream.close();
}
catch (IOException e) {
LOG.error(e, () -> "Failed to close input stream on read");
}
}
}
}
/**
* Writer implementation for writing objects as JSON using a custom
* {@code ObjectMapper}.
*
* @param <T> object type to write
*/
class CustomJacksonJsonWriter<T> implements SchemaWriter<T> {
private final ObjectMapper objectMapper;
CustomJacksonJsonWriter(ObjectMapper objectMapper) {
Assert.notNull(objectMapper, "objectMapper must not be null");
this.objectMapper = objectMapper;
}
@Override
public byte[] write(T message) {
try {
return this.objectMapper.writeValueAsBytes(message);
}
catch (JsonProcessingException e) {
throw new SchemaSerializationException(e);
}
}
}
}

2
spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java
View File

@@ -49,7 +49,7 @@ import org.springframework.lang.Nullable;
import org.springframework.pulsar.cache.provider.CacheProvider;
import org.springframework.pulsar.core.CachingPulsarProducerFactory.ProducerCacheKey;
import org.springframework.pulsar.core.CachingPulsarProducerFactory.ProducerWithCloseCallback;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.model.UserPojo;
import org.springframework.test.util.ReflectionTestUtils;
import org.springframework.util.ObjectUtils;

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

@@ -52,12 +52,16 @@ import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.springframework.core.ResolvableType;
import org.springframework.pulsar.annotation.PulsarMessage;
import org.springframework.pulsar.listener.Proto;
import org.springframework.pulsar.listener.Proto.Person;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* Unit tests for {@link DefaultSchemaResolver}.
*
@@ -68,6 +72,10 @@ class DefaultSchemaResolverTests {
private DefaultSchemaResolver resolver = new DefaultSchemaResolver();
private static String sanitizedClassName(Class<?> clazz) {
return clazz.getName().replace("$", ".");
}
@Nested
class CustomSchemaMappingsAPI {
@@ -295,10 +303,6 @@ class DefaultSchemaResolverTests {
.withMessage("Unsupported schema type: " + unsupportedType.name());
}
private String sanitizedClassName(Class<?> clazz) {
return clazz.getName().replace("$", ".");
}
@Nested
class SchemaTypeNone {
@@ -372,6 +376,38 @@ class DefaultSchemaResolverTests {
}
@Nested
class SchemaTypeJson {
@Test
void whenResolverHasObjectMapperThenReturnsCustomJsonSchema() {
var objectMapper = new ObjectMapper();
resolver.setObjectMapper(objectMapper);
var schema = mock(JSONSchema.class);
try (MockedStatic<JSONSchemaUtil> util = Mockito.mockStatic(JSONSchemaUtil.class)) {
util.when(() -> JSONSchemaUtil.schemaForTypeWithObjectMapper(Foo.class, objectMapper))
.thenReturn(schema);
var resolved = resolver.resolveSchema(SchemaType.JSON, ResolvableType.forType(Foo.class));
assertThat(resolved.value()).hasValueSatisfying((s) -> assertThat(s).isSameAs(schema));
util.verify(() -> JSONSchemaUtil.schemaForTypeWithObjectMapper(Foo.class, objectMapper));
}
}
@Test
void whenResolverDoesNotHaveObjectMapperThenReturnsDefaultJsonSchema() {
var objectMapper = new ObjectMapper();
var schema = mock(JSONSchema.class);
try (MockedStatic<JSONSchemaUtil> util = Mockito.mockStatic(JSONSchemaUtil.class)) {
util.when(() -> JSONSchemaUtil.schemaForTypeWithObjectMapper(Foo.class, objectMapper))
.thenReturn(schema);
var resolved = resolver.resolveSchema(SchemaType.JSON, ResolvableType.forType(Foo.class));
assertThat(resolved.value()).hasValueSatisfying((s) -> assertThat(s).isNotSameAs(schema));
util.verifyNoInteractions();
}
}
}
}
@Nested

30
spring-pulsar/src/test/java/org/springframework/pulsar/core/JsonSchemaUtilTests.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* Copyright 2023-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.pulsar.core;
import org.junit.jupiter.api.Test;
/**
* Tests for {@link JSONSchemaUtil}.
*/
class JsonSchemaUtilTests {
@Test
void foo() {
}
}

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

@@ -59,8 +59,9 @@ import org.junit.jupiter.params.provider.ValueSource;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.pulsar.annotation.EnablePulsar;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.model.json.UserRecordObjectMapper;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.util.function.ThrowingConsumer;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -448,6 +449,24 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
}
@Nested
class CustomObjectMapperTests {
@Test
void sendWithCustomJsonSchema() throws Exception {
// Prepare the schema with custom object mapper
var objectMapper = UserRecordObjectMapper.withSer();
var schema = JSONSchemaUtil.schemaForTypeWithObjectMapper(UserRecord.class, objectMapper);
var topic = "ptt-custom-object-mapper-topic";
var user = new UserRecord("elFoo", 21);
// serializer adds '-ser' to name and 10 to age
var expectedUser = new UserRecord("elFoo-ser", 31);
ThrowingConsumer<PulsarTemplate<UserRecord>> sendFunction = (template) -> template.send(user, schema);
sendAndConsume(sendFunction, topic, schema, expectedUser, true);
}
}
public static class Foo {
private String foo;

44
spring-pulsar/src/test/java/org/springframework/pulsar/listener/DefaultPulsarMessageListenerContainerTests.java
View File

@@ -32,6 +32,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -52,7 +53,10 @@ import org.junit.jupiter.api.Test;
import org.springframework.pulsar.core.ConsumerTestUtils;
import org.springframework.pulsar.core.DefaultPulsarConsumerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.JSONSchemaUtil;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.pulsar.test.model.json.UserRecordObjectMapper;
import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
import org.springframework.pulsar.transaction.PulsarAwareTransactionManager;
import org.springframework.test.util.ReflectionTestUtils;
@@ -418,4 +422,44 @@ class DefaultPulsarMessageListenerContainerTests implements PulsarTestContainerS
.withMessage("Transactional batch listeners do not support AckMode.RECORD");
}
@Test
void basicDefaultConsumerWithCustomObjectMapper() throws Exception {
var pulsarClient = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()).build();
var topic = "dpmlct-com-topic";
var pulsarConsumerFactory = new DefaultPulsarConsumerFactory<UserRecord>(pulsarClient,
List.of((consumerBuilder) -> {
consumerBuilder.topic(topic);
consumerBuilder.subscriptionName("dpmlct-com-sub");
}));
var latch = new CountDownLatch(1);
AtomicReference<UserRecord> consumedRecordRef = new AtomicReference<>();
var pulsarContainerProperties = new PulsarContainerProperties();
pulsarContainerProperties.setMessageListener((PulsarRecordMessageListener<UserRecord>) (consumer, msg) -> {
consumedRecordRef.set(msg.getValue());
latch.countDown();
});
// Prepare the schema with custom object mapper
var objectMapper = UserRecordObjectMapper.withDeser();
var schema = JSONSchemaUtil.schemaForTypeWithObjectMapper(UserRecord.class, objectMapper);
pulsarContainerProperties.setSchema(schema);
// Start the container
var container = new DefaultPulsarMessageListenerContainer<>(pulsarConsumerFactory, pulsarContainerProperties);
container.start();
// Send and consume message and ensure the deser was used
var pulsarProducerFactory = new DefaultPulsarProducerFactory<UserRecord>(pulsarClient, topic);
var pulsarTemplate = new PulsarTemplate<>(pulsarProducerFactory);
var sentUserRecord = new UserRecord("person", 51);
// deser adds '-deser' to name and 5 to age
var expectedReceivedUser = new UserRecord("person-deser", 56);
pulsarTemplate.sendAsync(sentUserRecord);
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(consumedRecordRef).hasValue(expectedReceivedUser);
container.stop();
pulsarClient.close();
}
}

4
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerAutoConsumeSchemaTests.java
View File

@@ -45,8 +45,8 @@ import org.springframework.pulsar.annotation.PulsarListener;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.listener.PulsarListenerAutoConsumeSchemaTests.PulsarListenerAutoConsumeSchemaTestsConfig;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.pulsar.test.model.UserPojo;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
/**

4
spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java
View File

@@ -72,8 +72,8 @@ import org.springframework.pulsar.core.TopicResolver;
import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithDefaultType.WithDefaultTypeConfig;
import org.springframework.pulsar.listener.PulsarListenerTests.SubscriptionTypeTests.WithSpecificTypes.WithSpecificTypesConfig;
import org.springframework.pulsar.support.PulsarHeaders;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.pulsar.test.model.UserPojo;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.util.backoff.FixedBackOff;

4
spring-pulsar/src/test/java/org/springframework/pulsar/reader/PulsarReaderAutoConsumeSchemaTests.java
View File

@@ -45,8 +45,8 @@ import org.springframework.pulsar.annotation.PulsarReader;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.PulsarTemplate;
import org.springframework.pulsar.reader.PulsarReaderAutoConsumeSchemaTests.PulsarReaderAutoConsumeSchemaTestsConfig;
import org.springframework.pulsar.test.support.model.UserPojo;
import org.springframework.pulsar.test.support.model.UserRecord;
import org.springframework.pulsar.test.model.UserPojo;
import org.springframework.pulsar.test.model.UserRecord;
import org.springframework.test.context.ContextConfiguration;
/**

79
spring-pulsar/src/testFixtures/java/org/springframework/pulsar/test/model/UserPojo.java Normal file
View File

@@ -0,0 +1,79 @@
/*
* Copyright 2022-2024 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.test.model;
import java.util.Objects;
/**
* Test object (user) defined via standard Java beans get/set methods.
* <p>
* <b>WARN</b> Do not convert this to a Record as this is used for Avro tests and Avro
* does not work well w/ records yet.
*/
public class UserPojo {
private String name;
private int age;
UserPojo() {
}
public UserPojo(String name, int age) {
this.name = name;
this.age = age;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public int getAge() {
return age;
}
public void setAge(int age) {
this.age = age;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
UserPojo user = (UserPojo) o;
return age == user.age && Objects.equals(name, user.name);
}
@Override
public int hashCode() {
return Objects.hash(name, age);
}
@Override
public String toString() {
return "User{" + "name='" + name + '\'' + ", age=" + age + '}';
}
}

26
spring-pulsar/src/testFixtures/java/org/springframework/pulsar/test/model/UserRecord.java Normal file
View File

@@ -0,0 +1,26 @@
/*
* Copyright 2022-2024 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.test.model;
/**
* Test object (user) defined via a Java record.
*
* @param name the user's name
* @param age the user's age
*/
public record UserRecord(String name, int age) {
}

52
spring-pulsar/src/testFixtures/java/org/springframework/pulsar/test/model/json/UserRecordDeserializer.java Normal file
View File

@@ -0,0 +1,52 @@
/*
* Copyright 2023-2024 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.test.model.json;
import java.io.IOException;
import org.springframework.pulsar.test.model.UserRecord;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
/**
* Custom Jackson deserializer for {@link UserRecord}.
*
* @author Chris Bono
* @since 1.2.0
*/
public class UserRecordDeserializer extends StdDeserializer<UserRecord> {
public UserRecordDeserializer() {
this(null);
}
public UserRecordDeserializer(Class<UserRecord> t) {
super(t);
}
@Override
public UserRecord deserialize(JsonParser jp, DeserializationContext ctxt) throws IOException {
JsonNode rootNode = jp.getCodec().readTree(jp);
var name = rootNode.get("name").asText();
var age = rootNode.get("age").asInt();
return new UserRecord(name + "-deser", age + 5);
}
}

58
spring-pulsar/src/testFixtures/java/org/springframework/pulsar/test/model/json/UserRecordObjectMapper.java Normal file
View File

@@ -0,0 +1,58 @@
/*
* Copyright 2023-2024 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.test.model.json;
import org.springframework.pulsar.test.model.UserRecord;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.module.SimpleModule;
/**
* Constructs custom {@link ObjectMapper} instances that leverage the
* {@link UserRecordSerializer} and {@link UserRecordDeserializer}.
*/
public final class UserRecordObjectMapper {
private UserRecordObjectMapper() {
}
public static ObjectMapper withSer() {
var objectMapper = new ObjectMapper();
var module = new SimpleModule();
module.addSerializer(UserRecord.class, new UserRecordSerializer());
objectMapper.registerModule(module);
return objectMapper;
}
public static ObjectMapper withDeser() {
var objectMapper = new ObjectMapper();
var module = new SimpleModule();
module.addDeserializer(UserRecord.class, new UserRecordDeserializer());
objectMapper.registerModule(module);
return objectMapper;
}
public static ObjectMapper withSerAndDeser() {
var objectMapper = new ObjectMapper();
var module = new SimpleModule();
module.addSerializer(UserRecord.class, new UserRecordSerializer());
module.addDeserializer(UserRecord.class, new UserRecordDeserializer());
objectMapper.registerModule(module);
return objectMapper;
}
}

51
spring-pulsar/src/testFixtures/java/org/springframework/pulsar/test/model/json/UserRecordSerializer.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* Copyright 2023-2024 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.test.model.json;
import java.io.IOException;
import org.springframework.pulsar.test.model.UserRecord;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.ser.std.StdSerializer;
/**
* Custom Jackson serializer for {@link UserRecord}.
*
* @author Chris Bono
* @since 1.2.0
*/
public class UserRecordSerializer extends StdSerializer<UserRecord> {
public UserRecordSerializer() {
this(null);
}
public UserRecordSerializer(Class<UserRecord> t) {
super(t);
}
@Override
public void serialize(UserRecord value, JsonGenerator jgen, SerializerProvider provider) throws IOException {
jgen.writeStartObject();
jgen.writeStringField("name", value.name() + "-ser");
jgen.writeNumberField("age", value.age() + 10);
jgen.writeEndObject();
}
}