From ecec5bb854f61fe1fa1a597d16c9893a4de39a00 Mon Sep 17 00:00:00 2001
From: JonasG <jonas.grgt@gmail.com>
Date: Thu, 29 Feb 2024 13:27:24 +0100
Subject: [PATCH] Cache producer interceptors in PulsarTemplate

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
---
 .../pulsar/core/PulsarTemplate.java           | 21 +++++++++++++------
 .../pulsar/core/PulsarTemplateTests.java      |  2 +-
 2 files changed, 16 insertions(+), 7 deletions(-)

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 11109e7b..1af30b4f 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
@@ -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);
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 f0e561eb..17f10582 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
@@ -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");