From e32ed00750611ad5ae6e3aa8fa28a8b01dadfd3b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Alexander=20Preu=C3=9F?= Date: Mon, 8 Aug 2022 13:56:56 +0200 Subject: [PATCH] Add ProducerInterceptor support See #35 --- .../PulsarAutoConfiguration.java | 8 +- .../PulsarAutoConfigurationTests.java | 13 ++ .../core/CachingPulsarProducerFactory.java | 30 +++-- .../core/DefaultPulsarProducerFactory.java | 20 ++- .../pulsar/core/PulsarProducerFactory.java | 16 +++ .../pulsar/core/PulsarTemplate.java | 17 ++- .../CachingPulsarProducerFactoryTests.java | 119 ++++++++++++------ .../core/PulsarProducerFactoryTests.java | 33 ++++- .../pulsar/core/PulsarTemplateTests.java | 28 ++++- 9 files changed, 226 insertions(+), 58 deletions(-) diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index cfa912a8..39fdd85b 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -17,7 +17,9 @@ package org.springframework.pulsar.autoconfigure; import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; @@ -40,6 +42,7 @@ import org.springframework.pulsar.core.PulsarTemplate; * * @author Soby Chacko * @author Chris Bono + * @author Alexander Preuß */ @AutoConfiguration @ConditionalOnClass(PulsarTemplate.class) @@ -84,8 +87,9 @@ public class PulsarAutoConfiguration { @Bean @ConditionalOnMissingBean(PulsarTemplate.class) - public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory) { - return new PulsarTemplate<>(pulsarProducerFactory); + public PulsarTemplate pulsarTemplate(PulsarProducerFactory pulsarProducerFactory, + ObjectProvider interceptors) { + return new PulsarTemplate<>(pulsarProducerFactory, interceptors.orderedStream().toList()); } @Bean diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 4c0c6edd..e60652b6 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -21,6 +21,8 @@ import static org.mockito.Mockito.mock; import java.util.concurrent.TimeUnit; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; +import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -47,6 +49,7 @@ import org.springframework.pulsar.listener.DefaultPulsarMessageListenerContainer * Autoconfiguration tests for {@link PulsarAutoConfiguration}. * * @author Chris Bono + * @author Alexander Preuß */ @SuppressWarnings("unchecked") class PulsarAutoConfigurationTests { @@ -154,6 +157,16 @@ class PulsarAutoConfigurationTests { .isSameAs(listenerAnnotationBeanPostProcessor)); } + @Test + void customProducerInterceptorIsUsedInPulsarTemplate() { + ProducerInterceptor interceptor = mock(ProducerInterceptor.class); + this.contextRunner.withBean("customProducerInterceptor", ProducerInterceptor.class, () -> interceptor) + .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) + .extracting("interceptors") + .asInstanceOf(InstanceOfAssertFactories.list(ProducerInterceptor.class)) + .contains(interceptor))); + } + @Nested class ProducerFactoryAutoConfigurationTests { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java index 8656be0e..4c27ca2f 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java @@ -17,6 +17,7 @@ package org.springframework.pulsar.core; import java.time.Duration; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.CompletableFuture; @@ -30,6 +31,7 @@ import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.apache.pulsar.common.protocol.schema.SchemaHash; import org.springframework.aop.framework.AopProxyUtils; @@ -58,6 +60,7 @@ import com.github.benmanes.caffeine.cache.Scheduler; * * @param producer type. * @author Chris Bono + * @author Alexander Preuß */ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactory implements DisposableBean { @@ -89,12 +92,14 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor } @Override - public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) { + public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors) { final String topicName = ProducerUtils.resolveTopicName(topic, this); - ProducerCacheKey producerCacheKey = new ProducerCacheKey<>(schema, topicName, messageRouter); + ProducerCacheKey producerCacheKey = new ProducerCacheKey<>(schema, topicName, messageRouter, + producerInterceptors); return this.producerCache.get(producerCacheKey, (st) -> { try { - return this.doCreateProducer(st.topic, st.schema, st.router); + return this.doCreateProducer(st.topic, st.schema, st.router, producerInterceptors); } catch (PulsarClientException ex) { throw new RuntimeException(ex); @@ -103,9 +108,9 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor } @Override - protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter) - throws PulsarClientException { - Producer producer = super.doCreateProducer(topic, schema, messageRouter); + protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors) throws PulsarClientException { + Producer producer = super.doCreateProducer(topic, schema, messageRouter, producerInterceptors); return wrapProducerWithCloseCallback(producer, (p) -> this.logger.trace(() -> String.format("Client closed producer %s but will skip actual closing", ProducerUtils.formatProducer(producer)))); @@ -166,19 +171,25 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor private final MessageRouter router; + private final List interceptors; + /** * Constructs an instance. * @param schema the schema the producer is configured to use * @param topic the topic the producer is configured to send to * @param router the custom message router the producer is configured to use + * @param interceptors the list of producer interceptors the producer is + * configured to use */ - ProducerCacheKey(Schema schema, String topic, @Nullable MessageRouter router) { + ProducerCacheKey(Schema schema, String topic, @Nullable MessageRouter router, + @Nullable List interceptors) { Assert.notNull(schema, () -> "'schema' must be non-null"); Assert.notNull(topic, () -> "'topic' must be non-null"); this.schema = schema; this.schemaHash = SchemaHash.of(this.schema); this.topic = topic; this.router = router; + this.interceptors = interceptors; } @Override @@ -191,12 +202,13 @@ public class CachingPulsarProducerFactory extends DefaultPulsarProducerFactor } ProducerCacheKey that = (ProducerCacheKey) o; return this.topic.equals(that.topic) && this.schemaHash.equals(that.schemaHash) - && Objects.equals(this.router, that.router); + && Objects.equals(this.router, that.router) && Objects.equals(this.interceptors, that.interceptors); } @Override public int hashCode() { - return this.topic.hashCode() + this.schemaHash.hashCode() + Objects.hashCode(this.router); + return this.topic.hashCode() + this.schemaHash.hashCode() + Objects.hashCode(this.router) + + Objects.hashCode(this.interceptors); } } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java index 59a939f7..f645879a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java @@ -17,6 +17,7 @@ package org.springframework.pulsar.core; import java.util.HashMap; +import java.util.List; import java.util.Map; import org.apache.commons.logging.LogFactory; @@ -26,6 +27,7 @@ import org.apache.pulsar.client.api.ProducerBuilder; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.springframework.core.log.LogAccessor; import org.springframework.util.CollectionUtils; @@ -36,6 +38,7 @@ import org.springframework.util.CollectionUtils; * @param producer type. * @author Soby Chacko * @author Chris Bono + * @author Alexander Preuß */ public class DefaultPulsarProducerFactory implements PulsarProducerFactory { @@ -54,17 +57,23 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory @Override public Producer createProducer(String topic, Schema schema) throws PulsarClientException { - return createProducer(topic, schema, null); + return createProducer(topic, schema, null, null); } @Override public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException { - return doCreateProducer(topic, schema, messageRouter); + return createProducer(topic, schema, messageRouter, null); } - protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter) - throws PulsarClientException { + @Override + public Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors) throws PulsarClientException { + return doCreateProducer(topic, schema, messageRouter, producerInterceptors); + } + + protected Producer doCreateProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors) throws PulsarClientException { final String resolvedTopic = ProducerUtils.resolveTopicName(topic, this); this.logger.trace(() -> String.format("Creating producer for '%s' topic", resolvedTopic)); final ProducerBuilder producerBuilder = this.pulsarClient.newProducer(schema); @@ -75,6 +84,9 @@ public class DefaultPulsarProducerFactory implements PulsarProducerFactory if (messageRouter != null) { producerBuilder.messageRouter(messageRouter); } + if (!CollectionUtils.isEmpty(producerInterceptors)) { + producerBuilder.intercept(producerInterceptors.toArray(new ProducerInterceptor[0])); + } return producerBuilder.create(); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java index 83766971..ecd4b5f3 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java @@ -16,12 +16,14 @@ package org.springframework.pulsar.core; +import java.util.List; import java.util.Map; import org.apache.pulsar.client.api.MessageRouter; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; /** * The strategy to create a {@link Producer} instance(s). @@ -29,6 +31,7 @@ import org.apache.pulsar.client.api.Schema; * @param producer payload type * @author Soby Chacko * @author Chris Bono + * @author Alexander Preuß */ public interface PulsarProducerFactory { @@ -54,6 +57,19 @@ public interface PulsarProducerFactory { Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException; + /** + * Create a producer. + * @param topic the topic the producer will send messages to or {@code null} to use + * the default topic + * @param schema the schema of the messages to be sent + * @param messageRouter the optional message router to use + * @param producerInterceptors the optional producer interceptors to use + * @return the producer + * @throws PulsarClientException if any error occurs + */ + Producer createProducer(String topic, Schema schema, MessageRouter messageRouter, + List producerInterceptors) throws PulsarClientException; + /** * Return a map of configuration options to use when creating producers. * @return the map of configuration options 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 7e48f09c..cfd29481 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 @@ -16,6 +16,7 @@ package org.springframework.pulsar.core; +import java.util.List; import java.util.concurrent.CompletableFuture; import org.apache.commons.logging.LogFactory; @@ -25,6 +26,7 @@ import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TypedMessageBuilder; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.springframework.core.log.LogAccessor; @@ -42,13 +44,26 @@ public class PulsarTemplate implements PulsarOperations { private final PulsarProducerFactory producerFactory; + private final List interceptors; + /** * Constructs a template instance. * @param producerFactory the producer factory used to create the backing Pulsar * producers. */ public PulsarTemplate(PulsarProducerFactory producerFactory) { + this(producerFactory, null); + } + + /** + * Constructs a template instance. + * @param producerFactory the producer factory used to create the backing Pulsar + * producers. + * @param interceptors the {@link ProducerInterceptor}s to add to the producer. + */ + public PulsarTemplate(PulsarProducerFactory producerFactory, List interceptors) { this.producerFactory = producerFactory; + this.interceptors = interceptors; } @Override @@ -89,7 +104,7 @@ public class PulsarTemplate implements PulsarOperations { private Producer prepareProducerForSend(String topic, T message, MessageRouter messageRouter) throws PulsarClientException { Schema schema = SchemaUtils.getSchema(message); - return this.producerFactory.createProducer(topic, schema, messageRouter); + return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java index 2d7b2497..86bd0469 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java @@ -36,6 +36,7 @@ import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.apache.pulsar.client.impl.schema.StringSchema; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterEach; @@ -58,6 +59,7 @@ import com.github.benmanes.caffeine.cache.Cache; * Tests for {@link CachingPulsarProducerFactory}. * * @author Chris Bono + * @author Alexander Preuß */ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @@ -76,7 +78,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void createProducerMultipleCalls() throws PulsarClientException { PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); - ProducerCacheKey cacheKey = new ProducerCacheKey<>(schema, "topic1", null); + ProducerCacheKey cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null); Producer producer1 = producerFactory.createProducer("topic1", schema); Producer producer2 = producerFactory.createProducer("topic1", new StringSchema()); @@ -111,32 +113,59 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { Schema schema2 = new StringSchema(); MessageRouter router1 = mock(MessageRouter.class); MessageRouter router2 = mock(MessageRouter.class); + List interceptors1 = List.of(mock(ProducerInterceptor.class)); + List interceptors2 = List.of(mock(ProducerInterceptor.class)); PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); - // ask for the same 9 unique combos 3x - should end up w/ only 9 entries in cache + // ask for the same 21 unique combos 3x - should end up w/ only 21 entries in + // cache for (int i = 0; i < 3; i++) { producerFactory.createProducer(topic1, schema1); producerFactory.createProducer(topic1, schema1, router1); producerFactory.createProducer(topic1, schema1, router2); + producerFactory.createProducer(topic1, schema1, router1, interceptors1); + producerFactory.createProducer(topic1, schema1, router1, interceptors2); + producerFactory.createProducer(topic1, schema1, router2, interceptors1); + producerFactory.createProducer(topic1, schema1, router2, interceptors2); producerFactory.createProducer(topic1, schema2); producerFactory.createProducer(topic1, schema2, router1); producerFactory.createProducer(topic1, schema2, router2); + producerFactory.createProducer(topic1, schema2, router1, interceptors1); + producerFactory.createProducer(topic1, schema2, router1, interceptors2); + producerFactory.createProducer(topic1, schema2, router2, interceptors1); + producerFactory.createProducer(topic1, schema2, router2, interceptors2); producerFactory.createProducer(topic2, schema1); producerFactory.createProducer(topic2, schema1, router1); producerFactory.createProducer(topic2, schema1, router2); + producerFactory.createProducer(topic2, schema1, router1, interceptors1); + producerFactory.createProducer(topic2, schema1, router1, interceptors2); + producerFactory.createProducer(topic2, schema1, router2, interceptors1); + producerFactory.createProducer(topic2, schema1, router2, interceptors2); } List> expectedCacheKeys = new ArrayList<>(); - expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, null)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router1)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router2)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, null)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router1)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router2)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, null)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router1)); - expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router2)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, null, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router1, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router1, interceptors1)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router1, interceptors2)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router2, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router2, interceptors1)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic1, router2, interceptors2)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, null, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router1, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router1, interceptors1)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router1, interceptors2)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router2, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router2, interceptors1)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema2, topic1, router2, interceptors2)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, null, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router1, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router1, interceptors1)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router1, interceptors2)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router2, null)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router2, interceptors1)); + expectedCacheKeys.add(new ProducerCacheKey<>(schema1, topic2, router2, interceptors2)); getAssertedProducerCache(producerFactory, expectedCacheKeys); } @@ -144,8 +173,8 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void factoryDestroyCleansUpCacheAndClosesProducers() throws PulsarClientException { CachingPulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); - ProducerCacheKey cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null); - ProducerCacheKey cacheKey2 = new ProducerCacheKey<>(schema, "topic2", null); + ProducerCacheKey cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null, null); + ProducerCacheKey cacheKey2 = new ProducerCacheKey<>(schema, "topic2", null, null); Producer actualProducer1 = actualProducerFrom(producerFactory.createProducer("topic1", schema)); Producer actualProducer2 = actualProducerFrom(producerFactory.createProducer("topic2", schema)); @@ -164,7 +193,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { void producerEvictedFromCache() throws PulsarClientException { CachingPulsarProducerFactory producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, Collections.emptyMap(), Duration.ofSeconds(3L), 10L, 2); - ProducerCacheKey cacheKey = new ProducerCacheKey<>(schema, "topic1", null); + ProducerCacheKey cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null); Producer actualProducer = actualProducerFrom(producerFactory.createProducer("topic1", schema)); @@ -188,8 +217,9 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Override protected void assertProducerHasTopicSchemaAndRouter(Producer producer, String topic, Schema schema, - MessageRouter router) { - super.assertProducerHasTopicSchemaAndRouter(actualProducerFrom(producer), topic, schema, router); + MessageRouter router, List producerInterceptors) { + super.assertProducerHasTopicSchemaAndRouter(actualProducerFrom(producer), topic, schema, router, + producerInterceptors); } @SuppressWarnings("unchecked") @@ -218,7 +248,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { protected CachingPulsarProducerFactory producerFactory(PulsarClient pulsarClient, Map producerConfig) { CachingPulsarProducerFactory producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, - producerConfig, Duration.ofMinutes(5L), 10L, 2); + producerConfig, Duration.ofMinutes(5L), 30L, 2); producerFactories.add(producerFactory); return producerFactory; } @@ -228,13 +258,13 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { @Test void nullSchemaIsNotAllowed() { - assertThatThrownBy(() -> new ProducerCacheKey<>(null, "topic1", null)) + assertThatThrownBy(() -> new ProducerCacheKey<>(null, "topic1", null, null)) .isInstanceOf(IllegalArgumentException.class).hasMessage("'schema' must be non-null"); } @Test void nullTopicIsNotAllowed() { - assertThatThrownBy(() -> new ProducerCacheKey<>(schema, null, null)) + assertThatThrownBy(() -> new ProducerCacheKey<>(schema, null, null, null)) .isInstanceOf(IllegalArgumentException.class).hasMessage("'topic' must be non-null"); } @@ -249,32 +279,49 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests { static Stream equalsAndHashCodeTestProvider() { MessageRouter router1 = mock(MessageRouter.class); - ProducerCacheKey key1 = new ProducerCacheKey<>(Schema.STRING, "topic1", router1); + List interceptors1 = Collections.singletonList(mock(ProducerInterceptor.class)); + ProducerCacheKey key1 = new ProducerCacheKey<>(Schema.STRING, "topic1", router1, interceptors1); return Stream.of(arguments(Named.of("differentClass", key1), "someStrangeObject", false), arguments(Named.of("null", key1), null, false), arguments(Named.of("sameInstance", key1), key1, true), arguments( - Named.of("sameSchemaSameTopicSameNullRouter", - new ProducerCacheKey<>(Schema.STRING, "topic1", null)), - new ProducerCacheKey<>(Schema.STRING, "topic1", null), true), + Named.of("sameSchemaSameTopicSameNullRouterSameNullInterceptors", + new ProducerCacheKey<>(Schema.STRING, "topic1", null, null)), + new ProducerCacheKey<>(Schema.STRING, "topic1", null, null), true), arguments( - Named.of("sameSchemaSameTopicSameNonNullRouter", - new ProducerCacheKey<>(Schema.STRING, "topic1", router1)), - new ProducerCacheKey<>(Schema.STRING, "topic1", router1), true), + Named.of("sameSchemaSameTopicSameNonNullRouterSameNullInterceptors", + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, null)), + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, null), true), arguments( Named.of("differentSchemaInstanceSameSchemaType", - new ProducerCacheKey<>(new StringSchema(), "topic1", router1)), - new ProducerCacheKey<>(new StringSchema(), "topic1", router1), true), - arguments(Named.of("differentSchemaType", new ProducerCacheKey<>(Schema.STRING, "topic1", router1)), - new ProducerCacheKey<>(Schema.INT64, "topic1", router1), false), - arguments(Named.of("differentTopic", new ProducerCacheKey<>(Schema.STRING, "topic1", router1)), - new ProducerCacheKey<>(Schema.STRING, "topic2", router1), false), + new ProducerCacheKey<>(new StringSchema(), "topic1", router1, null)), + new ProducerCacheKey<>(new StringSchema(), "topic1", router1, null), true), + arguments( + Named.of("differentSchemaType", + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, interceptors1)), + new ProducerCacheKey<>(Schema.INT64, "topic1", router1, interceptors1), false), + arguments( + Named.of("differentTopic", + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, interceptors1)), + new ProducerCacheKey<>(Schema.STRING, "topic2", router1, interceptors1), false), arguments( Named.of("differentNonNullRouter", - new ProducerCacheKey<>(Schema.STRING, "topic1", router1)), - new ProducerCacheKey<>(Schema.STRING, "topic1", mock(MessageRouter.class)), false), - arguments(Named.of("differentNullRouter", new ProducerCacheKey<>(Schema.STRING, "topic1", router1)), - new ProducerCacheKey<>(Schema.STRING, "topic1", null), false)); + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, null)), + new ProducerCacheKey<>(Schema.STRING, "topic1", mock(MessageRouter.class), null), false), + arguments( + Named.of("differentNullRouter", + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, null)), + new ProducerCacheKey<>(Schema.STRING, "topic1", null, null), false), + arguments( + Named.of("differentNonNullInterceptors", + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, interceptors1)), + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, + Collections.singletonList(mock(ProducerInterceptor.class))), + false), + arguments( + Named.of("differentNullInterceptor", + new ProducerCacheKey<>(Schema.STRING, "topic1", router1, interceptors1)), + new ProducerCacheKey<>(Schema.STRING, "topic1", null, null), false)); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java index 1c37c9f8..9f7fe698 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.Mockito.mock; import java.util.Collections; +import java.util.List; import java.util.Map; import org.apache.pulsar.client.api.MessageRouter; @@ -28,6 +29,8 @@ import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; +import org.apache.pulsar.client.impl.ProducerInterceptors; import org.apache.pulsar.client.impl.conf.ProducerConfigurationData; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.AfterEach; @@ -39,6 +42,7 @@ import org.junit.jupiter.api.Test; * {@link CachingPulsarProducerFactory}. * * @author Chris Bono + * @author Alexander Preuß */ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { @@ -62,7 +66,7 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { void createProducerWithSpecificTopic() throws PulsarClientException { PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); try (Producer producer = producerFactory.createProducer("topic1", schema)) { - assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, null); + assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, null, null); } } @@ -71,7 +75,7 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.emptyMap()); MessageRouter router = mock(MessageRouter.class); try (Producer producer = producerFactory.createProducer("topic1", schema, router)) { - assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, router); + assertProducerHasTopicSchemaAndRouter(producer, "topic1", schema, router, null); } } @@ -80,7 +84,7 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { PulsarProducerFactory producerFactory = producerFactory(pulsarClient, Collections.singletonMap("topicName", "topic0")); try (Producer producer = producerFactory.createProducer(null, schema)) { - assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, null); + assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, null, null); } } @@ -90,7 +94,17 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { Collections.singletonMap("topicName", "topic0")); MessageRouter router = mock(MessageRouter.class); try (Producer producer = producerFactory.createProducer(null, schema, router)) { - assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, router); + assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, router, null); + } + } + + @Test + void createProducerWithDefaultTopicAndInterceptor() throws PulsarClientException { + PulsarProducerFactory producerFactory = producerFactory(pulsarClient, + Collections.singletonMap("topicName", "topic0")); + List interceptors = Collections.singletonList(mock(ProducerInterceptor.class)); + try (Producer producer = producerFactory.createProducer(null, schema, null, interceptors)) { + assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, null, interceptors); } } @@ -103,12 +117,21 @@ abstract class PulsarProducerFactoryTests extends AbstractContainerBaseTests { } protected void assertProducerHasTopicSchemaAndRouter(Producer producer, String topic, Schema schema, - MessageRouter router) { + MessageRouter router, List producerInterceptors) { assertThat(producer.getTopic()).isEqualTo(topic); assertThat(producer).hasFieldOrPropertyWithValue("schema", schema); assertThat(producer).extracting("conf") .asInstanceOf(InstanceOfAssertFactories.type(ProducerConfigurationData.class)) .extracting(ProducerConfigurationData::getCustomMessageRouter).isSameAs(router); + if (producerInterceptors == null) { + assertThat(producer).extracting("interceptors").isNull(); + } + else { + assertThat(producer).extracting("interceptors") + .asInstanceOf(InstanceOfAssertFactories.type(ProducerInterceptors.class)).extracting("interceptors") + .asInstanceOf(InstanceOfAssertFactories.type(List.class)).isEqualTo(producerInterceptors); + } + } /** 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 2bd31804..69ff53f3 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 @@ -20,11 +20,14 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; import static org.junit.jupiter.params.provider.Arguments.arguments; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import java.time.Duration; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; @@ -39,6 +42,7 @@ import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.api.TopicMetadata; +import org.apache.pulsar.client.api.interceptor.ProducerInterceptor; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.Named; import org.junit.jupiter.params.ParameterizedTest; @@ -100,7 +104,21 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { } } - static Stream sendMessageTestProvider() { + @ParameterizedTest(name = "{0}") + @MethodSource("interceptorInvocationTestProvider") + void interceptorInvocationTest(String topic, List interceptors) throws Exception { + try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) { + PulsarProducerFactory producerFactory = new DefaultPulsarProducerFactory<>(client, + Collections.singletonMap("topicName", topic)); + PulsarTemplate pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors); + pulsarTemplate.send("test-interceptor"); + for (ProducerInterceptor interceptor : interceptors) { + verify(interceptor, atLeastOnce()).eligible(any(Message.class)); + } + } + } + + private static Stream sendMessageTestProvider() { return Stream.of( arguments(Named.of("sendMessageToDefaultTopic", "smt-topic-1"), @@ -175,6 +193,14 @@ class PulsarTemplateTests extends AbstractContainerBaseTests { sampleMessageKeyCustomizer, mockRouter())); } + private static Stream interceptorInvocationTestProvider() { + return Stream.of( + arguments(Named.of("testSingleInterceptor", "iit-topic-1"), + Collections.singletonList(mock(ProducerInterceptor.class))), + arguments(Named.of("testMultipleInterceptors", "iit-topic-2"), + List.of(mock(ProducerInterceptor.class), mock(ProducerInterceptor.class)))); + } + private static MessageRouter mockRouter() { MessageRouter router = mock(MessageRouter.class); when(router.choosePartition(any(Message.class), any(TopicMetadata.class))).thenReturn(0);