From cffffa6496f894baf00b54df98c9baa4c6a9f53a Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 1 Mar 2024 22:54:43 -0600 Subject: [PATCH] Add producer interceptor cache key tests This commit adds a test to PulsarTemplateTests that ensures producers are properly cached when the template is configured with one or more producer interceptors. --- .../pulsar/core/PulsarTemplateTests.java | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) 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 17f10582..176555bf 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 @@ -35,6 +35,7 @@ import java.util.Objects; import java.util.Set; import java.util.UUID; import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; import java.util.stream.Stream; import org.apache.pulsar.client.api.Message; @@ -221,6 +222,35 @@ class PulsarTemplateTests implements PulsarTestContainerSupport { List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class)))); } + @Test + void interceptorUsedAsCacheKeyProperly() { + var producerFactory = new CachingPulsarProducerFactory(client, null, null, new DefaultTopicResolver(), + Duration.ofSeconds(10L), 10L, 10); + try { + var interceptors = List.of(mock(ProducerInterceptor.class)); + var pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); + assertCacheSize(producerFactory, 0); + IntStream.range(0, 3).forEach((i) -> { + pulsarTemplate.send("test-intercept-topic", "test-interceptor-" + i); + assertCacheSize(producerFactory, 1); + }); + assertCacheSize(producerFactory, 1); + } + finally { + // The CPPF returns producers that do not actually close when the template + // calls close on them - destroy does close the producers though + if (producerFactory != null) { + producerFactory.destroy(); + } + } + } + + private void assertCacheSize(CachingPulsarProducerFactory producerFactory, int expectedSize) { + assertThat(producerFactory).extracting("producerCache.cache.cache") + .asInstanceOf(InstanceOfAssertFactories.MAP) + .hasSize(expectedSize); + } + @ParameterizedTest @ValueSource(booleans = { true, false }) void sendMessageWithTopicInferredByTypeMappings(boolean producerFactoryHasDefaultTopic) throws Exception {