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

Make SchemaResolverCustomizer generic

Browse Source
This commit is contained in:
Chris Bono
2023-01-20 14:14:15 -06:00
committed by Soby Chacko
parent ea1597a0a1
commit 34a9101076
6 changed files with 27 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-listener.adoc
View File

@@ -15,7 +15,7 @@ The following example shows a schema resolver customizer that adds mappings for
[source, java]
----
@Bean
public DefaultSchemaResolverCustomizer schemaResolverCustomizer() {
public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
return (schemaResolver) -> {
schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class));

2
spring-pulsar-docs/src/main/asciidoc/schema-info/schema-info-template.adoc
View File

@@ -21,7 +21,7 @@ The following example shows a schema resolver customizer that adds mappings for
[source, java]
----
@Bean
public DefaultSchemaResolverCustomizer schemaResolverCustomizer() {
public SchemaResolverCustomizer<DefaultSchemaResolver> schemaResolverCustomizer() {
return (schemaResolver) -> {
schemaResolver.addCustomSchemaMapping(User.class, Schema.AVRO(User.class));
schemaResolver.addCustomSchemaMapping(Address.class, Schema.JSON(Address.class));

5
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java
View File

@@ -36,12 +36,12 @@ 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.DefaultSchemaResolverCustomizer;
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.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.function.PulsarFunction;
import org.springframework.pulsar.function.PulsarFunctionAdministration;
import org.springframework.pulsar.function.PulsarSink;
@@ -112,7 +112,8 @@ public class PulsarAutoConfiguration {
@Bean
@ConditionalOnMissingBean(SchemaResolver.class)
public DefaultSchemaResolver schemaResolver(Optional<DefaultSchemaResolverCustomizer> schemaResolverCustomizer) {
public DefaultSchemaResolver schemaResolver(
Optional<SchemaResolverCustomizer<DefaultSchemaResolver>> schemaResolverCustomizer) {
DefaultSchemaResolver schemaResolver = new DefaultSchemaResolver();
schemaResolverCustomizer.ifPresent((customizer) -> customizer.customize(schemaResolver));
return schemaResolver;

7
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -49,12 +49,12 @@ import org.springframework.pulsar.config.PulsarListenerEndpointRegistry;
import org.springframework.pulsar.core.CachingPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultPulsarProducerFactory;
import org.springframework.pulsar.core.DefaultSchemaResolver;
import org.springframework.pulsar.core.DefaultSchemaResolverCustomizer;
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.core.SchemaResolver.SchemaResolverCustomizer;
import org.springframework.pulsar.function.PulsarFunctionAdministration;
import org.springframework.pulsar.listener.AckMode;
import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -144,8 +144,9 @@ class PulsarAutoConfigurationTests {
void defaultSchemaResolverCanBeCustomized() {
record Foo() {
}
DefaultSchemaResolverCustomizer customizer = (sr) -> sr.addCustomSchemaMapping(Foo.class, Schema.STRING);
this.contextRunner.withBean("schemaResolverCustomizer", DefaultSchemaResolverCustomizer.class, () -> customizer)
SchemaResolverCustomizer<DefaultSchemaResolver> customizer = (sr) -> sr.addCustomSchemaMapping(Foo.class,
Schema.STRING);
this.contextRunner.withBean("schemaResolverCustomizer", SchemaResolverCustomizer.class, () -> customizer)
.run((context) -> assertThat(context).hasNotFailed().getBean(DefaultSchemaResolver.class)
.extracting(DefaultSchemaResolver::getCustomSchemaMappings, InstanceOfAssertFactories.MAP)
.containsEntry(Foo.class, Schema.STRING));

33
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultSchemaResolverCustomizer.java
View File

@@ -1,33 +0,0 @@
/*
* 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.core;
/**
* The interface to customize a {@link DefaultSchemaResolver}.
*
* @author Chris Bono
*/
@FunctionalInterface
public interface DefaultSchemaResolverCustomizer {
/**
* Customizes a {@link DefaultSchemaResolver}.
* @param schemaResolver the target of the customization
*/
void customize(DefaultSchemaResolver schemaResolver);
}

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

@@ -73,4 +73,22 @@ public interface SchemaResolver {
@Nullable
<T> Schema<T> getSchema(SchemaType schemaType, @Nullable ResolvableType messageType);
/**
* Callback interface that can be implemented by beans wishing to customize the schema
* resolver before it is fully initialized, in particular to tune its configuration.
*
* @param <T> the type of the {@link SchemaResolver}
* @author Chris Bono
*/
@FunctionalInterface
interface SchemaResolverCustomizer<T extends SchemaResolver> {
/**
* Customize the schema resolver.
* @param schemaResolver the schema resolver to customize
*/
void customize(T schemaResolver);
}
}