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

Stop using Class as map key

Browse Source 
This commit updates the DefaultTopicResolver and DefaulSchemaResolver to use
String class names rather than Class instances as map keys for their custom
mappings.

Resolves #1078
This commit is contained in:
Chris Bono
2025-04-21 13:03:23 -05:00
committed by Chris Bono
parent f4a803e4bd
commit 605bcde87e
4 changed files with 85 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -23,12 +23,12 @@ 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.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.schema.AvroSchema;
@@ -92,7 +92,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
BASE_SCHEMA_MAPPINGS.put(LocalTime.class, Schema.LOCAL_TIME);
}
private final Map<Class<?>, Schema<?>> customSchemaMappings = new LinkedHashMap<>();
private final Map<String, Schema<?>> customSchemaMappings = new LinkedHashMap<>();
private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
@@ -122,7 +122,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
*/
@Nullable
public Schema<?> addCustomSchemaMapping(Class<?> messageType, Schema<?> schema) {
return this.customSchemaMappings.put(messageType, schema);
return this.customSchemaMappings.put(this.toMessageTypeMapKey(messageType), schema);
}
/**
@@ -133,15 +133,30 @@ public class DefaultSchemaResolver implements SchemaResolver {
*/
@Nullable
public Schema<?> removeCustomMapping(Class<?> messageType) {
return this.customSchemaMappings.remove(messageType);
return this.customSchemaMappings.remove(this.toMessageTypeMapKey(messageType));
}
/**
* Gets the currently registered custom mappings from message type to schema.
* @return unmodifiable map of custom mappings
* Gets the currently registered custom mapping for the specified message type.
* @return optional custom topic registered for the message type
* @deprecated deprecated in favor of {@link #getCustomSchemaMapping(Class)} (Class)}
*/
@Deprecated(since = "1.2.5", forRemoval = true)
public Map<Class<?>, Schema<?>> getCustomSchemaMappings() {
return Collections.unmodifiableMap(this.customSchemaMappings);
Map<Class<?>, Schema<?>> copyOfMappings = new HashMap<>();
this.customSchemaMappings.entrySet()
.stream()
.map((e) -> copyOfMappings.put(this.fromMessageTypeMapKey(e.getKey()), e.getValue()));
return copyOfMappings;
}
/**
* Gets the currently registered custom mapping for the specified message type.
* @param messageType the message type
* @return optional custom topic registered for the message type
*/
public Optional<Schema<?>> getCustomSchemaMapping(Class<?> messageType) {
return Optional.ofNullable(this.customSchemaMappings.get(this.toMessageTypeMapKey(messageType)));
}
@Override
@@ -162,7 +177,7 @@ public class DefaultSchemaResolver implements SchemaResolver {
@Nullable
protected Schema<?> getCustomSchemaOrMaybeDefault(@Nullable Class<?> messageClass, boolean returnDefault) {
// Check for custom schema mapping
Schema<?> schema = this.customSchemaMappings.get(messageClass);
var schema = this.getCustomSchemaMapping(messageClass).orElse(null);
// If no custom schema mapping found, look for @PulsarMessage (if enabled)
if (this.usePulsarMessageAnnotations && schema == null && messageClass != null) {
@@ -289,4 +304,17 @@ public class DefaultSchemaResolver implements SchemaResolver {
return (Schema<X>) rawSchema;
}
private Class<?> fromMessageTypeMapKey(String messageTypeKey) {
try {
return Class.forName(messageTypeKey);
}
catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
private String toMessageTypeMapKey(Class<?> messageType) {
return messageType.getName();
}
}

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

@@ -16,9 +16,10 @@
package org.springframework.pulsar.core;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.function.Supplier;
import org.springframework.beans.BeansException;
@@ -45,7 +46,7 @@ public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
private final LogAccessor logger = new LogAccessor(this.getClass());
private final Map<Class<?>, String> customTopicMappings = new LinkedHashMap<>();
private final Map<String, String> customTopicMappings = new LinkedHashMap<>();
private final PulsarMessageAnnotationRegistry pulsarMessageAnnotationRegistry = new PulsarMessageAnnotationRegistry();
@@ -86,7 +87,7 @@ public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
*/
@Nullable
public String addCustomTopicMapping(Class<?> messageType, String topic) {
return this.customTopicMappings.put(messageType, topic);
return this.customTopicMappings.put(this.toMessageTypeMapKey(messageType), topic);
}
/**
@@ -97,15 +98,31 @@ public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
*/
@Nullable
public String removeCustomMapping(Class<?> messageType) {
return this.customTopicMappings.remove(messageType);
return this.customTopicMappings.remove(this.toMessageTypeMapKey(messageType));
}
/**
* Gets the currently registered custom mappings from message type to topic.
* Gets the currently registered custom mappings from message type class name to
* topic.
* @return unmodifiable map of custom mappings
* @deprecated deprecated in favor of {@link #getCustomTopicMapping(Class)}
*/
@Deprecated(since = "1.2.5", forRemoval = true)
public Map<Class<?>, String> getCustomTopicMappings() {
return Collections.unmodifiableMap(this.customTopicMappings);
Map<Class<?>, String> copyOfMappings = new HashMap<>();
this.customTopicMappings.entrySet()
.stream()
.map((e) -> copyOfMappings.put(this.fromMessageTypeMapKey(e.getKey()), e.getValue()));
return copyOfMappings;
}
/**
* Gets the currently registered custom mapping for the specified message type.
* @param messageType the message type
* @return optional custom topic registered for the message type
*/
public Optional<String> getCustomTopicMapping(Class<?> messageType) {
return Optional.ofNullable(this.customTopicMappings.get(this.toMessageTypeMapKey(messageType)));
}
@Override
@@ -141,7 +158,7 @@ public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
return Resolved.failed("Topic must be specified when the message is null");
}
// Check for custom topic mapping
String topic = this.customTopicMappings.get(messageType);
String topic = this.customTopicMappings.get(this.toMessageTypeMapKey(messageType));
// If no custom topic mapping found, look for @PulsarMessage (if enabled)
if (this.usePulsarMessageAnnotations && topic == null) {
@@ -174,6 +191,19 @@ public class DefaultTopicResolver implements TopicResolver, BeanFactoryAware {
.orElseThrow(() -> "Failed to resolve topic expression: %s".formatted(v));
}
private Class<?> fromMessageTypeMapKey(String messageTypeKey) {
try {
return Class.forName(messageTypeKey);
}
catch (ClassNotFoundException e) {
throw new RuntimeException(e);
}
}
private String toMessageTypeMapKey(Class<?> messageType) {
return messageType.getName();
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
if (beanFactory instanceof ConfigurableBeanFactory configurableBeanFactory) {

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

@@ -79,6 +79,7 @@ class DefaultSchemaResolverTests {
@Nested
class CustomSchemaMappingsAPI {
@SuppressWarnings("removal")
@Test
void noMappingsByDefault() {
assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP).isEmpty();
@@ -88,14 +89,13 @@ class DefaultSchemaResolverTests {
void addMappings() {
Schema<?> previouslyMappedSchema = resolver.addCustomSchemaMapping(Foo.class, Schema.STRING);
assertThat(previouslyMappedSchema).isNull();
assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
.containsEntry(Foo.class, Schema.STRING);
assertThat(resolver.getCustomSchemaMapping(Foo.class)).hasValue(Schema.STRING);
previouslyMappedSchema = resolver.addCustomSchemaMapping(Foo.class, Schema.BOOL);
assertThat(previouslyMappedSchema).isEqualTo(Schema.STRING);
assertThat(resolver.getCustomSchemaMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
.containsEntry(Foo.class, Schema.BOOL);
assertThat(resolver.getCustomSchemaMapping(Foo.class)).hasValue(Schema.BOOL);
}
@SuppressWarnings("removal")
@Test
void removeMappings() {
Schema<?> previouslyMappedSchema = resolver.removeCustomMapping(Foo.class);
@@ -423,10 +423,11 @@ class DefaultSchemaResolverTests {
.endsWith(JsonMsgType.class.getSimpleName());
// verify added to custom mappings
assertThat(resolver.getCustomSchemaMappings().get(JsonMsgType.class)).isSameAs(resolvedSchema);
assertThat(resolver.getCustomSchemaMapping(JsonMsgType.class))
.hasValueSatisfying((v) -> assertThat(v).isSameAs(resolvedSchema));
// verify subsequent calls skip resolution again
assertThat(resolver.resolveSchema(JsonMsgType.class, false).orElseThrow()).isSameAs(resolvedSchema);
assertThat(resolver.resolveSchema(JsonMsgType.class, false).orElseThrow()).isEqualTo(resolvedSchema);
verify(resolver, times(1)).getAnnotatedSchemaType(JsonMsgType.class);
}

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

@@ -166,7 +166,7 @@ class DefaultTopicResolverTests {
assertThat(resolver.resolveTopic(null, Baz.class, () -> defaultTopic).value().orElse(null))
.isEqualTo(bazTopic);
// verify added to custom mappings
assertThat(resolver.getCustomTopicMappings().get(Baz.class)).isEqualTo(bazTopic);
assertThat(resolver.getCustomTopicMapping(Baz.class)).hasValue(bazTopic);
// verify subsequent calls skip resolution again
assertThat(resolver.resolveTopic(null, Baz.class, () -> defaultTopic).value().orElse(null))
.isEqualTo(bazTopic);
@@ -242,6 +242,7 @@ class DefaultTopicResolverTests {
resolver = new DefaultTopicResolver();
}
@SuppressWarnings("removal")
@Test
void noMappingsByDefault() {
resolver = new DefaultTopicResolver();
@@ -254,14 +255,13 @@ class DefaultTopicResolverTests {
String topic2 = "bar-topic";
String previouslyMappedTopic = resolver.addCustomTopicMapping(Foo.class, topic1);
assertThat(previouslyMappedTopic).isNull();
assertThat(resolver.getCustomTopicMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
.containsEntry(Foo.class, topic1);
assertThat(resolver.getCustomTopicMapping(Foo.class)).hasValue(topic1);
previouslyMappedTopic = resolver.addCustomTopicMapping(Foo.class, topic2);
assertThat(previouslyMappedTopic).isEqualTo(topic1);
assertThat(resolver.getCustomTopicMappings()).asInstanceOf(InstanceOfAssertFactories.MAP)
.containsEntry(Foo.class, topic2);
assertThat(resolver.getCustomTopicMapping(Foo.class)).hasValue(topic2);
}
@SuppressWarnings("removal")
@Test
void removeMappings() {
String previouslyMappedTopic = resolver.removeCustomMapping(Foo.class);