diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java index 226705b6..49b76386 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/DefaultReactivePulsarListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -85,6 +85,7 @@ public class DefaultReactivePulsarListenerContainerFactory implements Reactiv ReactivePulsarListenerEndpoint endpoint) { ReactivePulsarContainerProperties properties = new ReactivePulsarContainerProperties<>(); + properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver()); if (!CollectionUtils.isEmpty(endpoint.getTopics())) { properties.setTopics(endpoint.getTopics()); diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java index 927d19b7..58bb6b23 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/config/MethodReactivePulsarListenerEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -31,6 +31,7 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.client.impl.schema.ProtobufSchema; +import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; @@ -43,7 +44,7 @@ import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; -import org.springframework.pulsar.core.SchemaUtils; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.listener.Acknowledgement; import org.springframework.pulsar.listener.adapter.HandlerAdapter; import org.springframework.pulsar.listener.adapter.PulsarMessagingMessageListenerAdapter; @@ -139,9 +140,14 @@ public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePul messageParameter = parameter.get(); } + // TODO refactor this all out later to SchemaResolver + // ResolvableType messageType = resolvableType(messageParameter); + // Schema schema = schemaResolver.getSchema(schemaType, messageType); + DefaultReactivePulsarMessageListenerContainer containerInstance = (DefaultReactivePulsarMessageListenerContainer) container; ReactivePulsarContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties(); SchemaType schemaType = pulsarContainerProperties.getSchemaType(); + SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver(); if (schemaType != SchemaType.NONE) { switch (schemaType) { case STRING -> pulsarContainerProperties.setSchema((Schema) Schema.STRING); @@ -173,15 +179,22 @@ public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePul pulsarContainerProperties.setSchema((Schema) messageSchema); } case KEY_VALUE -> { - Schema messageSchema = getMessageKeyValueSchema(messageParameter); + Schema messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter); pulsarContainerProperties.setSchema((Schema) messageSchema); } } } else { if (messageParameter != null) { - Schema messageSchema = getMessageSchema(messageParameter, - (messageClass) -> SchemaUtils.getSchema(messageClass, false)); + Schema messageSchema = null; + ResolvableType type = resolvableType(messageParameter); + if (KeyValue.class.isAssignableFrom(type.getRawClass())) { + messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter); + } + else { + messageSchema = getMessageSchema(messageParameter, + (messageClass) -> schemaResolver.getSchema(messageClass, false)); + } if (messageSchema != null) { pulsarContainerProperties.setSchema((Schema) messageSchema); } @@ -207,12 +220,12 @@ public class MethodReactivePulsarListenerEndpoint extends AbstractReactivePul return schemaFactory.apply(messageClass); } - private Schema getMessageKeyValueSchema(MethodParameter messageParameter) { + private Schema getMessageKeyValueSchema(SchemaResolver schemaResolver, MethodParameter messageParameter) { ResolvableType messageType = resolvableType(messageParameter); Class keyClass = messageType.resolveGeneric(0); Class valueClass = messageType.resolveGeneric(1); - Schema> keySchema = SchemaUtils.getSchema(keyClass); - Schema> valueSchema = SchemaUtils.getSchema(valueClass); + Schema> keySchema = schemaResolver.getSchema(keyClass); + Schema> valueSchema = schemaResolver.getSchema(valueClass); return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); } diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java index f46c6514..97e35953 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -26,7 +26,9 @@ import org.apache.pulsar.reactive.client.api.ReactiveMessageSender; import org.reactivestreams.Publisher; import org.springframework.core.log.LogAccessor; -import org.springframework.pulsar.core.SchemaUtils; +import org.springframework.lang.Nullable; +import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.SchemaResolver; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -43,15 +45,29 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { private final ReactivePulsarSenderFactory reactiveMessageSenderFactory; + private final SchemaResolver schemaResolver; + + @Nullable private Schema schema; /** - * Construct a template instance with observation configuration. + * Construct a template instance that uses the default schema resolver. * @param reactiveMessageSenderFactory the factory used to create the backing Pulsar * reactive senders */ public ReactivePulsarTemplate(ReactivePulsarSenderFactory reactiveMessageSenderFactory) { + this(reactiveMessageSenderFactory, new DefaultSchemaResolver()); + } + + /** + * Construct a template instance with a custom schema resolver. + * @param reactiveMessageSenderFactory the factory used to create the backing Pulsar + * @param schemaResolver the schema resolver to use reactive senders + */ + public ReactivePulsarTemplate(ReactivePulsarSenderFactory reactiveMessageSenderFactory, + SchemaResolver schemaResolver) { this.reactiveMessageSenderFactory = reactiveMessageSenderFactory; + this.schemaResolver = schemaResolver; } @Override @@ -106,10 +122,10 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { if (this.schema != null) { /* * If the template has a schema, we can create the message sender right away - * and use ReactiveMessageSender::sendMany to send them as a stream. Otherwise - * we need to wait to get a message to create it and we can't share it between - * messages. So we create one each time and use ReactiveMessageSender::sendOne - * to send messages individually. + * and use ReactiveMessageSender::sendMany to send them as a stream. + * Otherwise, we need to wait to get a message to create it and we can't share + * it between messages. So we create one each time and use + * ReactiveMessageSender::sendOne to send messages individually. */ ReactiveMessageSender sender = createMessageSender(topic, null, null); return messages.map(MessageSpec::of).as(sender::sendMany) @@ -134,7 +150,7 @@ public class ReactivePulsarTemplate implements ReactivePulsarOperations { private ReactiveMessageSender createMessageSender(String topic, T message, ReactiveMessageSenderBuilderCustomizer customizer) { - Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); + Schema schema = this.schema != null ? this.schema : this.schemaResolver.getSchema(message); return this.reactiveMessageSenderFactory.createSender(topic, schema, customizer == null ? Collections.emptyList() : Collections.singletonList(customizer)); } diff --git a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java index 1bd7f599..b73dad87 100644 --- a/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java +++ b/spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -24,6 +24,9 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; +import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.SchemaResolver; + /** * Contains runtime properties for a reactive listener container. * @@ -44,6 +47,8 @@ public class ReactivePulsarContainerProperties { private SchemaType schemaType; + private SchemaResolver schemaResolver = new DefaultSchemaResolver(); + private ReactivePulsarMessageHandler messageHandler; private Duration handlingTimeout = Duration.ofMinutes(2); @@ -84,6 +89,14 @@ public class ReactivePulsarContainerProperties { this.schemaType = schemaType; } + public SchemaResolver getSchemaResolver() { + return this.schemaResolver; + } + + public void setSchemaResolver(SchemaResolver schemaResolver) { + this.schemaResolver = schemaResolver; + } + public Collection getTopics() { return this.topics; } diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java index 680d6aea..45072cbd 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/ReactivePulsarTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -22,8 +22,10 @@ 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.stream.Stream; @@ -39,6 +41,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarTestContainerSupport; import reactor.core.publisher.Flux; @@ -52,12 +55,12 @@ import reactor.core.publisher.Mono; class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { @Test - void sendMessagesWithSpecificSchemaTest() throws Exception { - String topic = "smt-specific-schema-topic-reactive"; + void sendMessagesWithSpecificSchema() throws Exception { + String topic = "smt-specific-schema-reactive-topic"; try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build()) { try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) - .subscriptionName("test-specific-schema-subscription").subscribe()) { + .subscriptionName("smt-specific-schema-reactive-sub").subscribe()) { MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); senderSpec.setTopicName(topic); org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>( @@ -80,6 +83,42 @@ class ReactivePulsarTemplateTests implements PulsarTestContainerSupport { } } + @Test + void sendMessagesWithSpecificSchemaAndCustomTypeMappings() throws Exception { + String topic = "smt-specific-schema-custom-reactive-topic"; + try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) + .subscriptionName("smt-specific-schema-custom-reactive-sub").subscribe()) { + MutableReactiveMessageSenderSpec senderSpec = new MutableReactiveMessageSenderSpec(); + senderSpec.setTopicName(topic); + org.springframework.pulsar.reactive.core.ReactivePulsarSenderFactory producerFactory = new org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory<>( + client, senderSpec, null); + // Custom schema resolver allows not calling setSchema on template + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver( + Collections.singletonMap(Foo.class, Schema.JSON(Foo.class))); + org.springframework.pulsar.reactive.core.ReactivePulsarTemplate pulsarTemplate = new org.springframework.pulsar.reactive.core.ReactivePulsarTemplate<>( + producerFactory, schemaResolver); + + List foos = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + foos.add(new Foo("Foo-" + UUID.randomUUID(), "Bar-" + UUID.randomUUID())); + } + pulsarTemplate.send(Flux.fromIterable(foos)).subscribe(); + + // TODO figure out why ordering is not preserved when template does not + // have schema set + List foos2 = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + CompletableFuture> receiveFuture = consumer.receiveAsync(); + assertThat(receiveFuture).succeedsWithin(Duration.ofSeconds(3)); + foos2.add(receiveFuture.get().getValue()); + } + assertThat(foos).containsExactlyInAnyOrderElementsOf(foos2); + } + } + } + @ParameterizedTest(name = "{0}") @MethodSource("sendMessageTestProvider") void sendMessageTest(String testName, SendTestArgs testArgs) throws Exception { diff --git a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java index 25b51dda..f5afc7ce 100644 --- a/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java +++ b/spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -59,11 +59,13 @@ import org.springframework.messaging.handler.annotation.Header; import org.springframework.pulsar.config.PulsarClientConfiguration; import org.springframework.pulsar.config.PulsarClientFactoryBean; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTestContainerSupport; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; import org.springframework.pulsar.reactive.config.ReactivePulsarListenerContainerFactory; import org.springframework.pulsar.reactive.config.ReactivePulsarListenerEndpointRegistry; @@ -72,6 +74,7 @@ import org.springframework.pulsar.reactive.config.annotation.ReactivePulsarListe import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer; import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; +import org.springframework.pulsar.reactive.listener.ReactivePulsarListenerTests.SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.User2; import org.springframework.pulsar.support.PulsarHeaders; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -492,6 +495,189 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport { } + @Nested + @ContextConfiguration(classes = SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.class) + class SchemaCustomMappingsTestCases { + + static CountDownLatch jsonLatch = new CountDownLatch(3); + static CountDownLatch avroLatch = new CountDownLatch(3); + static CountDownLatch keyvalueLatch = new CountDownLatch(3); + static CountDownLatch protobufLatch = new CountDownLatch(3); + + @Test + void jsonSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(JSONSchema.of(User2.class)); + + for (int i = 0; i < 3; i++) { + template.send("json-custom-schema-topic", new User2("Jason", i)); + } + assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void avroSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(AvroSchema.of(User.class)); + + for (int i = 0; i < 3; i++) { + template.send("avro-custom-schema-topic", new User("Avi", i)); + } + assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void keyvalueSchema() throws Exception { + PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, Collections.emptyMap()); + PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); + Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), + KeyValueEncodingType.INLINE); + template.setSchema(kvSchema); + + for (int i = 0; i < 3; i++) { + template.send("keyvalue-custom-schema-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150))); + } + assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void protobufSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(ProtobufSchema.of(Proto.Person.class)); + + for (int i = 0; i < 3; i++) { + template.send("protobuf-custom-schema-topic", + Proto.Person.newBuilder().setId(i).setName("Paul").build()); + } + assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnableReactivePulsar + @Configuration + static class SchemaCustomMappingsTestConfig { + + @Bean + SchemaResolver customSchemaResolver() { + Map, Schema> customMappings = new HashMap<>(); + customMappings.put(User.class, Schema.AVRO(User.class)); + customMappings.put(User2.class, Schema.JSON(User2.class)); + customMappings.put(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class)); + return new DefaultSchemaResolver(customMappings); + } + + @Bean + ReactivePulsarListenerContainerFactory reactivePulsarListenerContainerFactory( + ReactivePulsarConsumerFactory pulsarConsumerFactory, SchemaResolver schemaResolver) { + ReactivePulsarContainerProperties containerProps = new ReactivePulsarContainerProperties<>(); + containerProps.setSchemaResolver(schemaResolver); + return new DefaultReactivePulsarListenerContainerFactory<>(pulsarConsumerFactory, containerProps); + } + + @ReactivePulsarListener(id = "jsonListener", topics = "json-custom-schema-topic", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenJson(User2 message) { + jsonLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(id = "avroListener", topics = "avro-custom-schema-topic", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenAvro(User message) { + avroLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(id = "keyvalueListener", topics = "keyvalue-custom-schema-topic", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenKeyvalue(KeyValue message) { + keyvalueLatch.countDown(); + return Mono.empty(); + } + + @ReactivePulsarListener(id = "protobufListener", topics = "protobuf-custom-schema-topic", + consumerCustomizer = "subscriptionInitialPositionEarliest") + Mono listenProtobuf(Proto.Person message) { + protobufLatch.countDown(); + return Mono.empty(); + } + + @Bean + ReactiveMessageConsumerBuilderCustomizer subscriptionInitialPositionEarliest() { + return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest); + } + + record User2(String name, int age) { + } + + } + + /** + * Do not convert this to a Record as Avro does not seem to work well w/ records. + */ + static class User { + + private String name; + + private int age; + + User() { + + } + + User(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; + } + User user = (User) 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 + '}'; + } + + } + + } + @Nested @ContextConfiguration(classes = ReactivePulsarListenerTests.PulsarHeadersTest.PulsarListenerWithHeadersConfig.class) class PulsarHeadersTest { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java index 547be0e5..caa7caa7 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAnnotationDrivenConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -28,6 +28,7 @@ import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.config.ConcurrentPulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerBeanNames; import org.springframework.pulsar.core.PulsarConsumerFactory; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.observation.PulsarListenerObservationConvention; import org.springframework.util.unit.DataSize; @@ -55,9 +56,11 @@ public class PulsarAnnotationDrivenConfiguration { ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory( ObjectProvider> consumerFactoryProvider, ObjectProvider observationRegistryProvider, - ObjectProvider observationConventionProvider) { + ObjectProvider observationConventionProvider, + SchemaResolver schemaResolver) { PulsarContainerProperties containerProperties = new PulsarContainerProperties(); + containerProperties.setSchemaResolver(schemaResolver); containerProperties.setSubscriptionType(this.pulsarProperties.getConsumer().getSubscriptionType()); containerProperties.setObservationConvention(observationConventionProvider.getIfUnique()); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 08f6e34c..751e9844 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -33,10 +33,12 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean; import org.springframework.pulsar.core.CachingPulsarProducerFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.function.PulsarFunction; import org.springframework.pulsar.function.PulsarFunctionAdministration; import org.springframework.pulsar.function.PulsarSink; @@ -65,26 +67,26 @@ public class PulsarAutoConfiguration { } @Bean - @ConditionalOnMissingBean(PulsarClientConfiguration.class) + @ConditionalOnMissingBean public PulsarClientConfiguration pulsarClientConfiguration() { return new PulsarClientConfiguration(this.properties.buildClientProperties()); } @Bean - @ConditionalOnMissingBean(PulsarClientFactoryBean.class) + @ConditionalOnMissingBean public PulsarClientFactoryBean pulsarClientFactoryBean(PulsarClientConfiguration pulsarClientConfiguration) { return new PulsarClientFactoryBean(pulsarClientConfiguration); } @Bean - @ConditionalOnMissingBean(PulsarProducerFactory.class) + @ConditionalOnMissingBean @ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "false") public PulsarProducerFactory pulsarProducerFactory(PulsarClient pulsarClient) { return new DefaultPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties()); } @Bean - @ConditionalOnMissingBean(PulsarProducerFactory.class) + @ConditionalOnMissingBean @ConditionalOnProperty(name = "spring.pulsar.producer.cache.enabled", havingValue = "true", matchIfMissing = true) public PulsarProducerFactory cachingPulsarProducerFactory(PulsarClient pulsarClient) { return new CachingPulsarProducerFactory<>(pulsarClient, this.properties.buildProducerProperties(), @@ -94,31 +96,37 @@ public class PulsarAutoConfiguration { } @Bean - @ConditionalOnMissingBean(PulsarTemplate.class) + @ConditionalOnMissingBean public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, - ObjectProvider interceptorsProvider, + ObjectProvider interceptorsProvider, SchemaResolver schemaResolver, ObjectProvider observationRegistryProvider, ObjectProvider observationConventionProvider) { return new PulsarTemplate<>(pulsarProducerFactory, interceptorsProvider.orderedStream().toList(), - this.properties.getTemplate().isObservationsEnabled() ? observationRegistryProvider.getIfUnique() - : null, + schemaResolver, this.properties.getTemplate().isObservationsEnabled() + ? observationRegistryProvider.getIfUnique() : null, observationConventionProvider.getIfUnique()); } @Bean - @ConditionalOnMissingBean(PulsarConsumerFactory.class) + @ConditionalOnMissingBean + public SchemaResolver schemaResolver() { + return new DefaultSchemaResolver(); + } + + @Bean + @ConditionalOnMissingBean public PulsarConsumerFactory pulsarConsumerFactory(PulsarClient pulsarClient) { return new DefaultPulsarConsumerFactory<>(pulsarClient, this.properties.buildConsumerProperties()); } @Bean - @ConditionalOnMissingBean(PulsarAdministration.class) + @ConditionalOnMissingBean public PulsarAdministration pulsarAdministration() { return new PulsarAdministration(this.properties.buildAdminProperties()); } @Bean - @ConditionalOnMissingBean(PulsarFunctionAdministration.class) + @ConditionalOnMissingBean @ConditionalOnProperty(name = "spring.pulsar.function.enabled", havingValue = "true", matchIfMissing = true) public PulsarFunctionAdministration pulsarFunctionAdministration(PulsarAdministration pulsarAdministration, ObjectProvider pulsarFunctions, ObjectProvider pulsarSinks, diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java index a4021116..c072f77e 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAnnotationDrivenConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -23,6 +23,7 @@ import org.springframework.boot.context.properties.PropertyMapper; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.config.PulsarListenerBeanNames; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.reactive.config.DefaultReactivePulsarListenerContainerFactory; import org.springframework.pulsar.reactive.config.annotation.EnableReactivePulsar; import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory; @@ -46,9 +47,11 @@ public class PulsarReactiveAnnotationDrivenConfiguration { @Bean @ConditionalOnMissingBean(name = "reactivePulsarListenerContainerFactory") DefaultReactivePulsarListenerContainerFactory reactivePulsarListenerContainerFactory( - ObjectProvider> consumerFactoryProvider) { + ObjectProvider> consumerFactoryProvider, + SchemaResolver schemaResolver) { ReactivePulsarContainerProperties containerProperties = new ReactivePulsarContainerProperties<>(); + containerProperties.setSchemaResolver(schemaResolver); containerProperties.setSubscriptionType(this.properties.getConsumer().getSubscriptionType()); PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull(); diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java index 4eec43ff..cf51ca3d 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarReactiveAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -32,6 +32,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Import; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.reactive.core.DefaultReactivePulsarConsumerFactory; import org.springframework.pulsar.reactive.core.DefaultReactivePulsarReaderFactory; import org.springframework.pulsar.reactive.core.DefaultReactivePulsarSenderFactory; @@ -113,9 +114,9 @@ public class PulsarReactiveAutoConfiguration { @Bean @ConditionalOnMissingBean - public ReactivePulsarTemplate pulsarReactiveTemplate( - ReactivePulsarSenderFactory reactivePulsarSenderFactory) { - return new ReactivePulsarTemplate<>(reactivePulsarSenderFactory); + public ReactivePulsarTemplate pulsarReactiveTemplate(ReactivePulsarSenderFactory reactivePulsarSenderFactory, + SchemaResolver schemaResolver) { + return new ReactivePulsarTemplate<>(reactivePulsarSenderFactory, schemaResolver); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java index 1108ee3d..4e0a33f1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/ConcurrentPulsarListenerContainerFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -58,6 +58,7 @@ public class ConcurrentPulsarListenerContainerFactory protected ConcurrentPulsarMessageListenerContainer createContainerInstance(PulsarListenerEndpoint endpoint) { PulsarContainerProperties properties = new PulsarContainerProperties(); + properties.setSchemaResolver(this.getContainerProperties().getSchemaResolver()); if (!CollectionUtils.isEmpty(endpoint.getTopics())) { properties.setTopics(endpoint.getTopics().toArray(new String[0])); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index 77f3e9b0..660a0469 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -32,6 +32,7 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; import org.apache.pulsar.client.impl.schema.ProtobufSchema; +import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; @@ -44,7 +45,7 @@ import org.springframework.messaging.converter.SmartMessageConverter; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; -import org.springframework.pulsar.core.SchemaUtils; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.listener.Acknowledgement; import org.springframework.pulsar.listener.ConcurrentPulsarMessageListenerContainer; import org.springframework.pulsar.listener.PulsarConsumerErrorHandler; @@ -148,6 +149,12 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo ConcurrentPulsarMessageListenerContainer containerInstance = (ConcurrentPulsarMessageListenerContainer) container; PulsarContainerProperties pulsarContainerProperties = containerInstance.getContainerProperties(); SchemaType schemaType = pulsarContainerProperties.getSchemaType(); + SchemaResolver schemaResolver = pulsarContainerProperties.getSchemaResolver(); + + // TODO refactor this all out later to SchemaResolver + // ResolvableType messageType = resolvableType(messageParameter); + // Schema schema = schemaResolver.getSchema(schemaType, messageType); + if (schemaType != SchemaType.NONE) { switch (schemaType) { case STRING -> pulsarContainerProperties.setSchema(Schema.STRING); @@ -179,15 +186,23 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo pulsarContainerProperties.setSchema(messageSchema); } case KEY_VALUE -> { - Schema messageSchema = getMessageKeyValueSchema(messageParameter); + Schema messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter); pulsarContainerProperties.setSchema(messageSchema); } } } else { if (messageParameter != null) { - Schema messageSchema = getMessageSchema(messageParameter, - (messageClass) -> SchemaUtils.getSchema(messageClass, false)); + + Schema messageSchema = null; + ResolvableType type = resolvableType(messageParameter); + if (KeyValue.class.isAssignableFrom(type.getRawClass())) { + messageSchema = getMessageKeyValueSchema(schemaResolver, messageParameter); + } + else { + messageSchema = getMessageSchema(messageParameter, + (messageClass) -> schemaResolver.getSchema(messageClass, false)); + } if (messageSchema != null) { pulsarContainerProperties.setSchema(messageSchema); } @@ -210,12 +225,12 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo return schemaFactory.apply(messageClass); } - private Schema getMessageKeyValueSchema(MethodParameter messageParameter) { + private Schema getMessageKeyValueSchema(SchemaResolver schemaResolver, MethodParameter messageParameter) { ResolvableType messageType = resolvableType(messageParameter); Class keyClass = messageType.resolveGeneric(0); Class valueClass = messageType.resolveGeneric(1); - Schema> keySchema = SchemaUtils.getSchema(keyClass); - Schema> valueSchema = SchemaUtils.getSchema(valueClass); + Schema> keySchema = schemaResolver.getSchema(keyClass); + Schema> valueSchema = schemaResolver.getSchema(valueClass); return Schema.KeyValue(keySchema, valueSchema, KeyValueEncodingType.INLINE); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java new file mode 100644 index 00000000..13ff6d3a --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java @@ -0,0 +1,111 @@ +/* + * Copyright 2022-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.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import org.apache.pulsar.client.api.Schema; + +import org.springframework.lang.Nullable; + +/** + * Default schema resolver capable of handling basic message types. + * + *

+ * Additional message types can be configured with the {@link #DefaultSchemaResolver(Map)} + * constructor. + * + * @author Soby Chacko + * @author Alexander Preuß + * @author Chris Bono + */ +public class DefaultSchemaResolver implements SchemaResolver { + + private static final Map, Schema> BASE_SCHEMA_MAPPINGS = new HashMap<>(); + static { + BASE_SCHEMA_MAPPINGS.put(byte[].class, Schema.BYTES); + BASE_SCHEMA_MAPPINGS.put(String.class, Schema.STRING); + BASE_SCHEMA_MAPPINGS.put(Byte.class, Schema.INT8); + BASE_SCHEMA_MAPPINGS.put(byte.class, Schema.INT8); + BASE_SCHEMA_MAPPINGS.put(Short.class, Schema.INT16); + BASE_SCHEMA_MAPPINGS.put(short.class, Schema.INT16); + BASE_SCHEMA_MAPPINGS.put(Integer.class, Schema.INT32); + BASE_SCHEMA_MAPPINGS.put(int.class, Schema.INT32); + BASE_SCHEMA_MAPPINGS.put(Long.class, Schema.INT64); + BASE_SCHEMA_MAPPINGS.put(long.class, Schema.INT64); + BASE_SCHEMA_MAPPINGS.put(Boolean.class, Schema.BOOL); + BASE_SCHEMA_MAPPINGS.put(boolean.class, Schema.BOOL); + BASE_SCHEMA_MAPPINGS.put(ByteBuffer.class, Schema.BYTEBUFFER); + BASE_SCHEMA_MAPPINGS.put(ByteBuffer.allocate(0).getClass(), Schema.BYTEBUFFER); + BASE_SCHEMA_MAPPINGS.put(ByteBuffer.allocateDirect(0).getClass(), Schema.BYTEBUFFER); + BASE_SCHEMA_MAPPINGS.put(Date.class, Schema.DATE); + BASE_SCHEMA_MAPPINGS.put(Double.class, Schema.DOUBLE); + BASE_SCHEMA_MAPPINGS.put(double.class, Schema.DOUBLE); + BASE_SCHEMA_MAPPINGS.put(Float.class, Schema.FLOAT); + BASE_SCHEMA_MAPPINGS.put(float.class, Schema.FLOAT); + BASE_SCHEMA_MAPPINGS.put(Instant.class, Schema.INSTANT); + BASE_SCHEMA_MAPPINGS.put(LocalDate.class, Schema.LOCAL_DATE); + BASE_SCHEMA_MAPPINGS.put(LocalDateTime.class, Schema.LOCAL_DATE_TIME); + BASE_SCHEMA_MAPPINGS.put(LocalTime.class, Schema.LOCAL_TIME); + } + + private final Map, Schema> customSchemaMappings; + + /** + * Constructs a resolver with no custom type mappings. + */ + public DefaultSchemaResolver() { + this(Collections.emptyMap()); + } + + /** + * Constructs a resolver with custom type mappings. + * @param customTypeSchemaMappings additional type to schema mappings to use + */ + public DefaultSchemaResolver(Map, Schema> customTypeSchemaMappings) { + this.customSchemaMappings = Objects.requireNonNull(customTypeSchemaMappings); + } + + @Override + public Schema getSchema(Class messageClass, boolean returnDefault) { + Schema schema = BASE_SCHEMA_MAPPINGS.get(messageClass); + if (schema == null) { + schema = getCustomSchemaOrMaybeDefault(messageClass, returnDefault); + } + return schema != null ? castToType(schema) : null; + } + + @Nullable + private Schema getCustomSchemaOrMaybeDefault(Class messageClass, boolean returnDefault) { + return this.customSchemaMappings.getOrDefault(messageClass, (returnDefault ? Schema.BYTES : null)); + } + + @SuppressWarnings({ "unchecked" }) + private Schema castToType(Schema rawSchema) { + return (Schema) rawSchema; + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java index e092914c..645c349c 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -18,6 +18,7 @@ package org.springframework.pulsar.core; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -58,6 +59,8 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { @Nullable private final List interceptors; + private final SchemaResolver schemaResolver; + @Nullable private final ObservationRegistry observationRegistry; @@ -70,37 +73,40 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { private Schema schema; /** - * Construct a template instance. + * Construct a template instance without interceptors that uses the default schema + * resolver. * @param producerFactory the factory used to create the backing Pulsar producers. */ public PulsarTemplate(PulsarProducerFactory producerFactory) { - this(producerFactory, null); + this(producerFactory, Collections.emptyList()); } /** - * Construct a template instance with optional interceptors. + * Construct a template instance with interceptors that uses the default schema + * resolver. * @param producerFactory the factory used to create the backing Pulsar producers. * @param interceptors the interceptors to add to the producer. */ - public PulsarTemplate(PulsarProducerFactory producerFactory, @Nullable List interceptors) { - this(producerFactory, interceptors, null, null); + public PulsarTemplate(PulsarProducerFactory producerFactory, List interceptors) { + this(producerFactory, interceptors, new DefaultSchemaResolver(), null, null); } /** - * Construct a template instance with optional interceptors and observation - * configuration. + * Construct a template instance with optional observation configuration. * @param producerFactory the factory used to create the backing Pulsar producers - * @param interceptors the optional list of interceptors to add to the producer + * @param interceptors the list of interceptors to add to the producer + * @param schemaResolver the schema resolver to use * @param observationRegistry the registry to record observations with or {@code null} * to not record observations * @param observationConvention the optional custom observation convention to use when * recording observations */ - public PulsarTemplate(PulsarProducerFactory producerFactory, @Nullable List interceptors, - @Nullable ObservationRegistry observationRegistry, + public PulsarTemplate(PulsarProducerFactory producerFactory, List interceptors, + SchemaResolver schemaResolver, @Nullable ObservationRegistry observationRegistry, @Nullable PulsarTemplateObservationConvention observationConvention) { this.producerFactory = producerFactory; this.interceptors = interceptors; + this.schemaResolver = schemaResolver; this.observationRegistry = observationRegistry; this.observationConvention = observationConvention; } @@ -204,7 +210,7 @@ public class PulsarTemplate implements PulsarOperations, BeanNameAware { private Producer prepareProducerForSend(@Nullable String topic, T message, @Nullable Collection encryptionKeys, @Nullable ProducerBuilderCustomizer producerCustomizer) throws PulsarClientException { - Schema schema = this.schema != null ? this.schema : SchemaUtils.getSchema(message); + Schema schema = this.schema != null ? this.schema : this.schemaResolver.getSchema(message); List> customizers = new ArrayList<>(); if (!CollectionUtils.isEmpty(this.interceptors)) { customizers.add(builder -> this.interceptors.forEach(builder::intercept)); diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java new file mode 100644 index 00000000..6ba6fc0c --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaResolver.java @@ -0,0 +1,64 @@ +/* + * 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 org.apache.pulsar.client.api.Schema; + +import org.springframework.lang.Nullable; + +/** + * Resolves schema to use for message types. + * + * @author Chris Bono + */ +public interface SchemaResolver { + + /** + * Get the schema to use for a particular message. + * @param the schema type + * @param message the message instance + * @return the schema to use or {@code null} if no schema could be resolved + */ + @Nullable + default Schema getSchema(T message) { + return getSchema(message.getClass()); + } + + /** + * Get the schema to use for a message type. + * @param the schema type + * @param messageType the message type + * @return the schema to use or {@code null} if no schema could be resolved + */ + @Nullable + default Schema getSchema(Class messageType) { + return getSchema(messageType, true); + } + + /** + * Get the schema to use for a message type. + * @param the schema type + * @param messageType the message type + * @param returnDefault whether to return default schema if no schema could be + * resolved + * @return the schema to use or the default schema if no schema could be resolved and + * {@code returnDefault} is {@code true} - otherwise {@code null} + */ + @Nullable + Schema getSchema(Class messageType, boolean returnDefault); + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java deleted file mode 100644 index 74c185e3..00000000 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/SchemaUtils.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Copyright 2022 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.pulsar.core; - -import org.apache.pulsar.client.api.Schema; - -/** - * Utility class for Pulsar schema inference. - * - * @author Soby Chacko - * @author Alexander Preuß - */ -public final class SchemaUtils { - - private SchemaUtils() { - } - - public static Schema getSchema(T message) { - return getSchema(message.getClass()); - } - - public static Schema getSchema(Class messageClass) { - return getSchema(messageClass, true); - } - - @SuppressWarnings("unchecked") - public static Schema getSchema(Class messageClass, boolean returnDefault) { - return switch (messageClass.getName()) { - case "java.lang.String" -> (Schema) Schema.STRING; - case "[B" -> (Schema) Schema.BYTES; - case "java.lang.Byte", "byte" -> (Schema) Schema.INT8; - case "java.lang.Short", "short" -> (Schema) Schema.INT16; - case "java.lang.Integer", "int" -> (Schema) Schema.INT32; - case "java.lang.Long", "long" -> (Schema) Schema.INT64; - case "java.lang.Boolean", "boolean" -> (Schema) Schema.BOOL; - case "java.nio.ByteBuffer" -> (Schema) Schema.BYTEBUFFER; - case "java.util.Date" -> (Schema) Schema.DATE; - case "java.lang.Double", "double" -> (Schema) Schema.DOUBLE; - case "java.lang.Float", "float" -> (Schema) Schema.FLOAT; - case "java.time.Instant" -> (Schema) Schema.INSTANT; - case "java.time.LocalDate" -> (Schema) Schema.LOCAL_DATE; - case "java.time.LocalDateTime" -> (Schema) Schema.LOCAL_DATE_TIME; - case "java.time.LocalTime" -> (Schema) Schema.LOCAL_TIME; - default -> (returnDefault ? (Schema) Schema.BYTES : null); - }; - } - -} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java index d8819e31..5476add0 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarContainerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -24,6 +24,8 @@ import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.core.task.AsyncTaskExecutor; +import org.springframework.pulsar.core.DefaultSchemaResolver; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.observation.PulsarListenerObservationConvention; import org.springframework.util.Assert; @@ -32,6 +34,7 @@ import org.springframework.util.Assert; * * @author Soby Chacko * @author Alexander Preuß + * @author Chris Bono */ public class PulsarContainerProperties { @@ -51,6 +54,8 @@ public class PulsarContainerProperties { private SchemaType schemaType; + private SchemaResolver schemaResolver; + private Object messageListener; private AsyncTaskExecutor consumerTaskExecutor; @@ -72,11 +77,13 @@ public class PulsarContainerProperties { public PulsarContainerProperties(String... topics) { this.topics = topics.clone(); this.topicsPattern = null; + this.schemaResolver = new DefaultSchemaResolver(); } public PulsarContainerProperties(String topicPattern) { this.topicsPattern = topicPattern; this.topics = null; + this.schemaResolver = new DefaultSchemaResolver(); } public Object getMessageListener() { @@ -169,14 +176,6 @@ public class PulsarContainerProperties { this.consumerStartTimeout = consumerStartTimeout; } - public Schema getSchema() { - return this.schema; - } - - public void setSchema(Schema schema) { - this.schema = schema; - } - public String[] getTopics() { return this.topics; } @@ -201,6 +200,14 @@ public class PulsarContainerProperties { this.subscriptionName = subscriptionName; } + public Schema getSchema() { + return this.schema; + } + + public void setSchema(Schema schema) { + this.schema = schema; + } + public SchemaType getSchemaType() { return this.schemaType; } @@ -209,6 +216,14 @@ public class PulsarContainerProperties { this.schemaType = schemaType; } + public SchemaResolver getSchemaResolver() { + return this.schemaResolver; + } + + public void setSchemaResolver(SchemaResolver schemaResolver) { + this.schemaResolver = schemaResolver; + } + public Properties getPulsarConsumerProperties() { return this.pulsarConsumerProperties; } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java new file mode 100644 index 00000000..2121c8ab --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java @@ -0,0 +1,163 @@ +/* + * 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 static org.assertj.core.api.AssertionsForClassTypes.assertThat; +import static org.junit.jupiter.params.provider.Arguments.arguments; + +import java.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.LocalTime; +import java.util.Collections; +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.stream.Stream; + +import org.apache.pulsar.client.api.Schema; +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; + +/** + * Unit tests for {@link DefaultSchemaResolver}. + * + * @author Chris Bono + */ +class DefaultSchemaResolverTests { + + @ParameterizedTest + @MethodSource("schemaByMessageInstanceProvider") + void getSchemaByMessageInstance(T message, Schema expectedSchema) { + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + assertThat(resolver.getSchema(message)).isEqualTo(expectedSchema); + } + + static Stream schemaByMessageInstanceProvider() { + // @formatter:off + return Stream.of( + arguments("foo", Schema.STRING), + arguments("foo".getBytes(), Schema.BYTES), + arguments(Byte.valueOf("0"), Schema.INT8), + arguments((byte) 0, Schema.INT8), + arguments(Short.valueOf("0"), Schema.INT16), + arguments((short) 0, Schema.INT16), + arguments(Integer.valueOf("0"), Schema.INT32), + arguments(0, Schema.INT32), + arguments(Long.valueOf("0"), Schema.INT64), + arguments(0L, Schema.INT64), + arguments(Boolean.TRUE, Schema.BOOL), + arguments(ByteBuffer.wrap("foo".getBytes()), Schema.BYTEBUFFER), + arguments(ByteBuffer.allocateDirect(10), Schema.BYTEBUFFER), + arguments(new Date(), Schema.DATE), + arguments(Double.valueOf("2.2"), Schema.DOUBLE), + arguments(2.3d, Schema.DOUBLE), + arguments(Float.valueOf("2.4"), Schema.FLOAT), + arguments(2.5f, Schema.FLOAT), + arguments(Instant.now(), Schema.INSTANT), + arguments(LocalDate.now(), Schema.LOCAL_DATE), + arguments(LocalDateTime.now(), Schema.LOCAL_DATE_TIME), + arguments(LocalTime.NOON, Schema.LOCAL_TIME) + ); + // @formatter:on + } + + @Test + void getSchemaByMessageInstanceCustomTypes() { + Schema fooSchema = Schema.AVRO(Foo.class); + Map, Schema> customTypes = new HashMap<>(); + customTypes.put(Foo.class, fooSchema); + customTypes.put(Bar.class, Schema.STRING); + DefaultSchemaResolver resolver = new DefaultSchemaResolver(customTypes); + assertThat(resolver.getSchema(new Foo("foo1"))).isSameAs(fooSchema); + assertThat(resolver.getSchema(new Bar<>("bar1"))).isEqualTo(Schema.STRING); + assertThat(resolver.getSchema(new Zaa("zaa1"))).isEqualTo(Schema.BYTES); // default + } + + @ParameterizedTest + @MethodSource("schemaByMessageTypeProvider") + void getSchemaByMessageType(Class messageType, Schema expectedSchema) { + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + assertThat(resolver.getSchema(messageType)).isEqualTo(expectedSchema); + } + + static Stream schemaByMessageTypeProvider() { + // @formatter:off + return Stream.of( + arguments(String.class, Schema.STRING), + arguments(byte[].class, Schema.BYTES), + arguments(Byte.class, Schema.INT8), + arguments(byte.class, Schema.INT8), + arguments(Short.class, Schema.INT16), + arguments(short.class, Schema.INT16), + arguments(Integer.class, Schema.INT32), + arguments(int.class, Schema.INT32), + arguments(Long.class, Schema.INT64), + arguments(long.class, Schema.INT64), + arguments(Boolean.class, Schema.BOOL), + arguments(boolean.class, Schema.BOOL), + arguments(ByteBuffer.class, Schema.BYTEBUFFER), + arguments(ByteBuffer.wrap("foo".getBytes()).getClass(), Schema.BYTEBUFFER), + arguments(ByteBuffer.allocateDirect(10).getClass(), Schema.BYTEBUFFER), + arguments(Date.class, Schema.DATE), + arguments(Double.class, Schema.DOUBLE), + arguments(double.class, Schema.DOUBLE), + arguments(Float.class, Schema.FLOAT), + arguments(float.class, Schema.FLOAT), + arguments(Instant.class, Schema.INSTANT), + arguments(LocalDate.class, Schema.LOCAL_DATE), + arguments(LocalDateTime.class, Schema.LOCAL_DATE_TIME), + arguments(LocalTime.class, Schema.LOCAL_TIME) + ); + // @formatter:on + } + + @Test + void getSchemaByMessageTypeCustomTypes() { + Schema fooSchema = Schema.AVRO(Foo.class); + Map, Schema> customTypes = new HashMap<>(); + customTypes.put(Foo.class, fooSchema); + customTypes.put(Bar.class, Schema.STRING); + DefaultSchemaResolver resolver = new DefaultSchemaResolver(customTypes); + assertThat(resolver.getSchema(Foo.class)).isSameAs(fooSchema); + assertThat(resolver.getSchema(Bar.class)).isEqualTo(Schema.STRING); + assertThat(resolver.getSchema(Zaa.class)).isEqualTo(Schema.BYTES); // default + } + + @Test + void getSchemaByMessageTypeWithoutDefaults() { + DefaultSchemaResolver resolver = new DefaultSchemaResolver(); + assertThat(resolver.getSchema(Foo.class, false)).isNull(); + + resolver = new DefaultSchemaResolver(Collections.singletonMap(Foo.class, Schema.STRING)); + assertThat(resolver.getSchema(Foo.class, false)).isEqualTo(Schema.STRING); + assertThat(resolver.getSchema(Bar.class, false)).isNull(); + } + + record Foo(String value) { + } + + record Bar (T value) { + } + + record Zaa(String value) { + } + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java index 0b10f814..ba5051b1 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarTemplateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -89,7 +89,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) .build()) { try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) - .subscriptionName("test-specific-schema-subscription").subscribe()) { + .subscriptionName("smt-specific-schema-subscription").subscribe()) { PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, Collections.singletonMap("topicName", topic)); PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory); @@ -102,6 +102,29 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { } } + @Test + void sendMessageWithSpecificSchemaAndCustomTypeMappings() throws Exception { + String topic = "smt-specific-schema-custom-topic"; + try (PulsarClient client = PulsarClient.builder().serviceUrl(PulsarTestContainerSupport.getPulsarBrokerUrl()) + .build()) { + try (Consumer consumer = client.newConsumer(Schema.JSON(Foo.class)).topic(topic) + .subscriptionName("smt-specific-schema-custom-subscription").subscribe()) { + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + Collections.singletonMap("topicName", topic)); + // Custom schema resolver allows not calling setSchema on template + DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver( + Collections.singletonMap(Foo.class, Schema.JSON(Foo.class))); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, Collections.emptyList(), + schemaResolver, 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); + } + } + } + @Test @SuppressWarnings("unchecked") void sendMessageWithEncryptionKeys() throws Exception { diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index 16d3068d..ecfa80de 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -64,12 +64,14 @@ import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.config.PulsarListenerEndpointRegistry; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; import org.springframework.pulsar.core.PulsarTestContainerSupport; import org.springframework.pulsar.core.PulsarTopic; +import org.springframework.pulsar.core.SchemaResolver; import org.springframework.pulsar.support.PulsarHeaders; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -560,59 +562,179 @@ public class PulsarListenerTests implements PulsarTestContainerSupport { } - static class User { + } - private String name; + /** + * Do not convert this to a Record as Avro does not seem to work well w/ records. + */ + static class User { - private int age; + private String name; - User() { + private int age; + User() { + } + + User(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; + } + User user = (User) 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 + '}'; + } + + } + + @Nested + @ContextConfiguration(classes = SchemaCustomMappingsTestCases.SchemaCustomMappingsTestConfig.class) + class SchemaCustomMappingsTestCases { + + static CountDownLatch jsonLatch = new CountDownLatch(3); + static CountDownLatch avroLatch = new CountDownLatch(3); + static CountDownLatch keyvalueLatch = new CountDownLatch(3); + static CountDownLatch protobufLatch = new CountDownLatch(3); + + @Test + void jsonSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(Schema.JSON(User2.class)); + + for (int i = 0; i < 3; i++) { + template.send("json-custom-mappings-topic", new User2("Jason", i)); + } + assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void avroSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(AvroSchema.of(User.class)); + + for (int i = 0; i < 3; i++) { + template.send("avro-custom-mappings-topic", new User("Avi", i)); + } + assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void keyvalueSchema() throws Exception { + PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( + pulsarClient, Collections.emptyMap()); + PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); + Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.JSON(User2.class), + KeyValueEncodingType.INLINE); + template.setSchema(kvSchema); + + for (int i = 0; i < 3; i++) { + template.send("keyvalue-custom-mappings-topic", new KeyValue<>("Kevin", new User2("Kevin", 5150))); + } + assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void protobufSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(ProtobufSchema.of(Proto.Person.class)); + + for (int i = 0; i < 3; i++) { + template.send("protobuf-custom-mappings-topic", + Proto.Person.newBuilder().setId(i).setName("Paul").build()); + } + assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @EnablePulsar + @Configuration + static class SchemaCustomMappingsTestConfig { + + @Bean + SchemaResolver customSchemaResolver() { + Map, Schema> customMappings = new HashMap<>(); + customMappings.put(User.class, Schema.AVRO(User.class)); + customMappings.put(User2.class, Schema.JSON(User2.class)); + customMappings.put(Proto.Person.class, Schema.PROTOBUF(Proto.Person.class)); + return new DefaultSchemaResolver(customMappings); } - User(String name, int age) { - this.name = name; - this.age = age; + @Bean + PulsarListenerContainerFactory pulsarListenerContainerFactory( + PulsarConsumerFactory pulsarConsumerFactory, SchemaResolver schemaResolver) { + PulsarContainerProperties containerProps = new PulsarContainerProperties(); + containerProps.setSchemaResolver(schemaResolver); + ConcurrentPulsarListenerContainerFactory pulsarListenerContainerFactory = new ConcurrentPulsarListenerContainerFactory<>( + pulsarConsumerFactory, containerProps, null); + return pulsarListenerContainerFactory; } - public String getName() { - return name; + @PulsarListener(id = "jsonListener", topics = "json-custom-mappings-topic", + subscriptionName = "subscription-4", properties = { "subscriptionInitialPosition=Earliest" }) + void listenJson(User2 message) { + jsonLatch.countDown(); } - public void setName(String name) { - this.name = name; + @PulsarListener(id = "avroListener", topics = "avro-custom-mappings-topic", + subscriptionName = "subscription-6", properties = { "subscriptionInitialPosition=Earliest" }) + void listenAvro(User message) { + avroLatch.countDown(); } - public int getAge() { - return age; + @PulsarListener(id = "keyvalueListener", topics = "keyvalue-custom-mappings-topic", + subscriptionName = "subscription-8", properties = { "subscriptionInitialPosition=Earliest" }) + void listenKeyvalue(KeyValue message) { + keyvalueLatch.countDown(); } - public void setAge(int age) { - this.age = age; + @PulsarListener(id = "protobufListener", topics = "protobuf-custom-mappings-topic", + subscriptionName = "subscription-10", properties = { "subscriptionInitialPosition=Earliest" }) + void listenProtobuf(Proto.Person message) { + protobufLatch.countDown(); } - @Override - public boolean equals(Object o) { - if (this == o) { - return true; - } - if (o == null || getClass() != o.getClass()) { - return false; - } - User user = (User) 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 + '}'; - } + } + record User2(String name, int age) { } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java index 3f034fe4..c9e568d9 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -40,6 +40,7 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean; import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; @@ -146,7 +147,8 @@ public class ObservationIntegrationTests extends SampleTestRunner implements Pul @Bean public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, ObservationRegistry observationRegistry) { - return new PulsarTemplate<>(pulsarProducerFactory, null, observationRegistry, null); + return new PulsarTemplate<>(pulsarProducerFactory, null, new DefaultSchemaResolver(), observationRegistry, + null); } @Bean diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java index 9f5ee4d0..ed00f79f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/observation/ObservationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * Copyright 2022-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. @@ -45,6 +45,7 @@ import org.springframework.pulsar.config.PulsarClientFactoryBean; import org.springframework.pulsar.config.PulsarListenerContainerFactory; import org.springframework.pulsar.core.DefaultPulsarConsumerFactory; import org.springframework.pulsar.core.DefaultPulsarProducerFactory; +import org.springframework.pulsar.core.DefaultSchemaResolver; import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; @@ -186,7 +187,7 @@ public class ObservationTests implements PulsarTestContainerSupport { @Bean(name = "observationTestsTemplate") PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, ObservationRegistry observationRegistry) { - return new PulsarTemplate<>(pulsarProducerFactory, null, observationRegistry, + return new PulsarTemplate<>(pulsarProducerFactory, null, new DefaultSchemaResolver(), observationRegistry, new DefaultPulsarTemplateObservationConvention() { @Override public KeyValues getLowCardinalityKeyValues(PulsarMessageSenderContext context) { diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index f20df39e..c3a5d455 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -7,6 +7,7 @@ +