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

Cache producer interceptors in PulsarTemplate

Browse Source 
This commit modifies the PulsarTemplate to only adapt the list of
producer interceptors into a list of producer customizers once in
order to properly take the interceptors into account when caching
producers.

Fixes #593
This commit is contained in:
JonasG
2024-02-29 13:27:24 +01:00
committed by Chris Bono
parent 1f2f6ed46d
commit ecec5bb854
2 changed files with 16 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarTemplate.java
View File

@@ -63,16 +63,16 @@ public class PulsarTemplate<T>
private final PulsarProducerFactory<T> producerFactory;
private final List<ProducerInterceptor> interceptors;
private final SchemaResolver schemaResolver;
private final TopicResolver topicResolver;
private final List<ProducerBuilderCustomizer<T>> interceptorsCustomizers;
/**
* Whether to record observations.
*/
private boolean observationEnabled;
private final boolean observationEnabled;
/**
* The registry to record observations with.
@@ -121,10 +121,19 @@ public class PulsarTemplate<T>
public PulsarTemplate(PulsarProducerFactory<T> producerFactory, List<ProducerInterceptor> interceptors,
SchemaResolver schemaResolver, TopicResolver topicResolver, boolean observationEnabled) {
this.producerFactory = producerFactory;
this.interceptors = interceptors;
this.schemaResolver = schemaResolver;
this.topicResolver = topicResolver;
this.observationEnabled = observationEnabled;
if (!CollectionUtils.isEmpty(interceptors)) {
this.interceptorsCustomizers = interceptors.stream().map(this::adaptInterceptorToCustomizer).toList();
}
else {
this.interceptorsCustomizers = null;
}
}
private ProducerBuilderCustomizer<T> adaptInterceptorToCustomizer(ProducerInterceptor interceptor) {
return b -> b.intercept(interceptor);
}
@Override
@@ -279,8 +288,8 @@ public class PulsarTemplate<T>
@Nullable Collection<String> encryptionKeys, @Nullable ProducerBuilderCustomizer<T> producerCustomizer) {
Schema<T> resolvedSchema = schema == null ? this.schemaResolver.resolveSchema(message).orElseThrow() : schema;
List<ProducerBuilderCustomizer<T>> customizers = new ArrayList<>();
if (!CollectionUtils.isEmpty(this.interceptors)) {
customizers.add(builder -> this.interceptors.forEach(builder::intercept));
if (!CollectionUtils.isEmpty(this.interceptorsCustomizers)) {
customizers.addAll(this.interceptorsCustomizers);
}
if (producerCustomizer != null) {
customizers.add(producerCustomizer);

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

@@ -204,7 +204,7 @@ class PulsarTemplateTests implements PulsarTestContainerSupport {
@ParameterizedTest(name = "{0}")
@MethodSource("interceptorInvocationTestProvider")
void interceptorInvocationTest(String topic, List<ProducerInterceptor> interceptors) throws Exception {
void interceptorInvocationTest(String topic, List<ProducerInterceptor> interceptors) {
PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client, topic);
PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors);
pulsarTemplate.send("test-interceptor");