From 5aa97e43cdbc7e88cf4786dc0a757035b9c20550 Mon Sep 17 00:00:00 2001 From: Aleksei Arsenev Date: Tue, 4 Jul 2023 21:14:27 +0300 Subject: [PATCH] Add @PulsarTypeMapping for default topic/schema This commit introduces the @PulsarTypeMapping annotation which can be used on message classes to specify default topic and/or schema info. --- .../pages/reference/topic-resolution.adoc | 10 ++++ .../pulsar/annotation/PulsarTypeMapping.java | 55 +++++++++++++++++++ .../pulsar/core/DefaultSchemaResolver.java | 10 ++++ .../pulsar/core/DefaultTopicResolver.java | 20 ++++++- .../core/DefaultSchemaResolverTests.java | 10 ++++ .../core/DefaultTopicResolverTests.java | 9 +++ 6 files changed, 111 insertions(+), 3 deletions(-) create mode 100644 spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarTypeMapping.java diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc index af170e2e..f5d82169 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/topic-resolution.adoc @@ -34,6 +34,16 @@ NOTE: The `message-type` is the fully-qualified name of the message class. WARNING: If the message (or the first message of a `Publisher` input) is `null`, the framework won't be able to determine the topic from it. Another method shall be used to specify the topic if your application is likely to send `null` messages. +=== Specified via annotation +When no topic passed into API and no mappings configured, the system looks for `PulsarTopic` annotation. The following example configures topic for `Baz` class using annotation: + +[source,java,indent=0,subs="verbatim"] +---- +@PulsarTopic("baz-topic") +record Baz(String value) { +} +---- + === Custom topic resolver The preferred method of adding mappings is via the property mentioned above. However, if more control is needed you can replace the default resolver by proving your own implementation, for example: diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarTypeMapping.java b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarTypeMapping.java new file mode 100644 index 00000000..3ec14973 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/annotation/PulsarTypeMapping.java @@ -0,0 +1,55 @@ +/* + * Copyright 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.annotation; + +import org.apache.pulsar.common.schema.SchemaType; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Specifies default topic and schema for class. + * + * @author Aleksei Arsenev + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@Documented +public @interface PulsarTypeMapping { + + /** + * Default topic for class. + * @return topic + */ + String topic() default ""; + + /** + * Default schema type for class. + * @return schema type + */ + SchemaType schemaType() default SchemaType.NONE; + + /** + * Message key type (must be specified when schema type is {@code KEY_VALUE}) + * @return message key type + */ + Class messageKeyType() default Void.class; + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java index 802ffbc1..250bbe2d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolver.java @@ -39,8 +39,10 @@ import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; import org.springframework.core.ResolvableType; +import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.log.LogAccessor; import org.springframework.lang.Nullable; +import org.springframework.pulsar.annotation.PulsarTypeMapping; /** * Default schema resolver capable of handling basic message types. @@ -138,6 +140,14 @@ public class DefaultSchemaResolver implements SchemaResolver { @Nullable protected Schema getCustomSchemaOrMaybeDefault(@Nullable Class messageClass, boolean returnDefault) { Schema schema = this.customSchemaMappings.get(messageClass); + if (schema == null && messageClass != null) { + PulsarTypeMapping annotation = AnnotationUtils.findAnnotation(messageClass, PulsarTypeMapping.class); + if (annotation != null && annotation.schemaType() != SchemaType.NONE) { + var resolvedSchema = resolveSchema(annotation.schemaType(), messageClass, annotation.messageKeyType()); + resolvedSchema.ifResolved(objectSchema -> addCustomSchemaMapping(messageClass, objectSchema)); + schema = resolvedSchema.get().orElse(null); + } + } if (schema == null && returnDefault) { if (messageClass != null) { try { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java index 594d15a1..afbacb8a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultTopicResolver.java @@ -17,11 +17,13 @@ package org.springframework.pulsar.core; import java.util.Collections; -import java.util.LinkedHashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Supplier; +import org.springframework.core.annotation.AnnotationUtils; import org.springframework.lang.Nullable; +import org.springframework.pulsar.annotation.PulsarTypeMapping; import org.springframework.util.StringUtils; /** @@ -35,7 +37,7 @@ import org.springframework.util.StringUtils; */ public class DefaultTopicResolver implements TopicResolver { - private final Map, String> customTopicMappings = new LinkedHashMap<>(); + private final Map, String> customTopicMappings = new ConcurrentHashMap<>(); /** * Adds a custom mapping from message type to topic. @@ -100,7 +102,19 @@ public class DefaultTopicResolver implements TopicResolver { if (messageType == null) { return Resolved.failed("Topic must be specified when the message is null"); } - String topic = this.customTopicMappings.getOrDefault(messageType, defaultTopicSupplier.get()); + + String topic = this.getCustomTopicMappings().get(messageType); + if (topic == null) { + PulsarTypeMapping annotation = AnnotationUtils.findAnnotation(messageType, PulsarTypeMapping.class); + if (annotation != null && !annotation.topic().isBlank()) { + this.addCustomTopicMapping(messageType, annotation.topic()); + topic = annotation.topic(); + } + } + + if (topic == null) { + topic = defaultTopicSupplier.get(); + } return topic == null ? Resolved.failed("Topic must be specified when no default topic is configured") : Resolved.of(topic); } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java index 09e6799d..8c772f8f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultSchemaResolverTests.java @@ -49,6 +49,7 @@ import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.MethodSource; import org.springframework.core.ResolvableType; +import org.springframework.pulsar.annotation.PulsarTypeMapping; import org.springframework.pulsar.listener.Proto; import org.springframework.pulsar.listener.Proto.Person; @@ -201,6 +202,11 @@ class DefaultSchemaResolverTests { assertThat(resolver.resolveSchema(Bar.class, true).orElseThrow()).isEqualTo(Schema.BYTES); } + + @Test + void annotatedMessageType() { + assertThat(resolver.resolveSchema(Zaz.class, false).orElseThrow()).isEqualTo(Schema.STRING); + } } @Nested @@ -371,4 +377,8 @@ class DefaultSchemaResolverTests { record Zaa(String value) { } + @PulsarTypeMapping(schemaType = SchemaType.STRING) + record Zaz(String value) { + } + } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java index 937aa829..1c3ba11f 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultTopicResolverTests.java @@ -30,6 +30,7 @@ import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; import org.springframework.lang.Nullable; +import org.springframework.pulsar.annotation.PulsarTypeMapping; /** * Unit tests for {@link DefaultTopicResolver}. @@ -44,6 +45,8 @@ class DefaultTopicResolverTests { private static final String fooTopic = "foo-topic1"; + private static final String bazTopic = "baz-topic1"; + private static final String stringTopic = "string-topic1"; private DefaultTopicResolver resolver = new DefaultTopicResolver(); @@ -111,6 +114,8 @@ class DefaultTopicResolverTests { arguments("complexMessageWithUserTopic", userTopic, Foo.class, defaultTopic, userTopic), arguments("complexMessageNoUserTopic", null, Foo.class, defaultTopic, fooTopic), arguments("nullMessageWithUserTopicAndDefault", userTopic, null, defaultTopic, userTopic), + arguments("annotationMessageWithUserTopic", userTopic, Baz.class, defaultTopic, userTopic), + arguments("annotationMessageNoUserTopic", null, Baz.class, defaultTopic, bazTopic), arguments("nullMessageWithDefault", null, null, defaultTopic, null), arguments("noMatchWithUserTopicAndDefault", userTopic, Bar.class, defaultTopic, userTopic), arguments("noMatchWithUserTopic", userTopic, Bar.class, null, userTopic), @@ -166,4 +171,8 @@ class DefaultTopicResolverTests { record Bar(String value) { } + @PulsarTypeMapping(topic = bazTopic) + record Baz(String value) { + } + }