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

Add ProducerInterceptor support

Browse Source 
See #35
This commit is contained in:
Alexander Preuß
2022-08-08 13:56:56 +02:00
committed by Chris Bono
parent 01017cb020
commit e32ed00750
9 changed files with 226 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<ProducerInterceptor> interceptors) {
return new PulsarTemplate<>(pulsarProducerFactory, interceptors.orderedStream().toList());
}
@Bean

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

@@ -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 {

30
spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java
View File

@@ -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 <T> producer type.
* @author Chris Bono
* @author Alexander Preuß
*/
public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactory<T> implements DisposableBean {
@@ -89,12 +92,14 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter) {
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors) {
final String topicName = ProducerUtils.resolveTopicName(topic, this);
ProducerCacheKey<T> producerCacheKey = new ProducerCacheKey<>(schema, topicName, messageRouter);
ProducerCacheKey<T> 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<T> extends DefaultPulsarProducerFactor
}
@Override
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter)
throws PulsarClientException {
Producer<T> producer = super.doCreateProducer(topic, schema, messageRouter);
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
Producer<T> 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<T> extends DefaultPulsarProducerFactor
private final MessageRouter router;
private final List<ProducerInterceptor> 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<T> schema, String topic, @Nullable MessageRouter router) {
ProducerCacheKey(Schema<T> schema, String topic, @Nullable MessageRouter router,
@Nullable List<ProducerInterceptor> 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<T> 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);
}
}

20
spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarProducerFactory.java
View File

@@ -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 <T> producer type.
* @author Soby Chacko
* @author Chris Bono
* @author Alexander Preuß
*/
public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T> {
@@ -54,17 +57,23 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
@Override
public Producer<T> createProducer(String topic, Schema<T> schema) throws PulsarClientException {
return createProducer(topic, schema, null);
return createProducer(topic, schema, null, null);
}
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter)
throws PulsarClientException {
return doCreateProducer(topic, schema, messageRouter);
return createProducer(topic, schema, messageRouter, null);
}
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter)
throws PulsarClientException {
@Override
public Producer<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
return doCreateProducer(topic, schema, messageRouter, producerInterceptors);
}
protected Producer<T> doCreateProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors) throws PulsarClientException {
final String resolvedTopic = ProducerUtils.resolveTopicName(topic, this);
this.logger.trace(() -> String.format("Creating producer for '%s' topic", resolvedTopic));
final ProducerBuilder<T> producerBuilder = this.pulsarClient.newProducer(schema);
@@ -75,6 +84,9 @@ public class DefaultPulsarProducerFactory<T> implements PulsarProducerFactory<T>
if (messageRouter != null) {
producerBuilder.messageRouter(messageRouter);
}
if (!CollectionUtils.isEmpty(producerInterceptors)) {
producerBuilder.intercept(producerInterceptors.toArray(new ProducerInterceptor[0]));
}
return producerBuilder.create();
}

16
spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarProducerFactory.java
View File

@@ -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 <T> producer payload type
* @author Soby Chacko
* @author Chris Bono
* @author Alexander Preuß
*/
public interface PulsarProducerFactory<T> {
@@ -54,6 +57,19 @@ public interface PulsarProducerFactory<T> {
Producer<T> createProducer(String topic, Schema<T> 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<T> createProducer(String topic, Schema<T> schema, MessageRouter messageRouter,
List<ProducerInterceptor> producerInterceptors) throws PulsarClientException;
/**
* Return a map of configuration options to use when creating producers.
* @return the map of configuration options

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

@@ -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<T> implements PulsarOperations<T> {
private final PulsarProducerFactory<T> producerFactory;
private final List<ProducerInterceptor> interceptors;
/**
* Constructs a template instance.
* @param producerFactory the producer factory used to create the backing Pulsar
* producers.
*/
public PulsarTemplate(PulsarProducerFactory<T> 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<T> producerFactory, List<ProducerInterceptor> interceptors) {
this.producerFactory = producerFactory;
this.interceptors = interceptors;
}
@Override
@@ -89,7 +104,7 @@ public class PulsarTemplate<T> implements PulsarOperations<T> {
private Producer<T> prepareProducerForSend(String topic, T message, MessageRouter messageRouter)
throws PulsarClientException {
Schema<T> schema = SchemaUtils.getSchema(message);
return this.producerFactory.createProducer(topic, schema, messageRouter);
return this.producerFactory.createProducer(topic, schema, messageRouter, this.interceptors);
}
}

119
spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java
View File

@@ -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<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
ProducerCacheKey<String> cacheKey = new ProducerCacheKey<>(schema, "topic1", null);
ProducerCacheKey<String> cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null);
Producer<String> producer1 = producerFactory.createProducer("topic1", schema);
Producer<String> producer2 = producerFactory.createProducer("topic1", new StringSchema());
@@ -111,32 +113,59 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
Schema<String> schema2 = new StringSchema();
MessageRouter router1 = mock(MessageRouter.class);
MessageRouter router2 = mock(MessageRouter.class);
List<ProducerInterceptor> interceptors1 = List.of(mock(ProducerInterceptor.class));
List<ProducerInterceptor> interceptors2 = List.of(mock(ProducerInterceptor.class));
PulsarProducerFactory<String> 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<ProducerCacheKey<String>> 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<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
ProducerCacheKey<String> cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null);
ProducerCacheKey<String> cacheKey2 = new ProducerCacheKey<>(schema, "topic2", null);
ProducerCacheKey<String> cacheKey1 = new ProducerCacheKey<>(schema, "topic1", null, null);
ProducerCacheKey<String> cacheKey2 = new ProducerCacheKey<>(schema, "topic2", null, null);
Producer<String> actualProducer1 = actualProducerFrom(producerFactory.createProducer("topic1", schema));
Producer<String> actualProducer2 = actualProducerFrom(producerFactory.createProducer("topic2", schema));
@@ -164,7 +193,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
void producerEvictedFromCache() throws PulsarClientException {
CachingPulsarProducerFactory<String> producerFactory = new CachingPulsarProducerFactory<>(pulsarClient,
Collections.emptyMap(), Duration.ofSeconds(3L), 10L, 2);
ProducerCacheKey<String> cacheKey = new ProducerCacheKey<>(schema, "topic1", null);
ProducerCacheKey<String> cacheKey = new ProducerCacheKey<>(schema, "topic1", null, null);
Producer<String> actualProducer = actualProducerFrom(producerFactory.createProducer("topic1", schema));
@@ -188,8 +217,9 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
@Override
protected void assertProducerHasTopicSchemaAndRouter(Producer<String> producer, String topic, Schema<String> schema,
MessageRouter router) {
super.assertProducerHasTopicSchemaAndRouter(actualProducerFrom(producer), topic, schema, router);
MessageRouter router, List<ProducerInterceptor> producerInterceptors) {
super.assertProducerHasTopicSchemaAndRouter(actualProducerFrom(producer), topic, schema, router,
producerInterceptors);
}
@SuppressWarnings("unchecked")
@@ -218,7 +248,7 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
protected CachingPulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
Map<String, Object> producerConfig) {
CachingPulsarProducerFactory<String> 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<Arguments> equalsAndHashCodeTestProvider() {
MessageRouter router1 = mock(MessageRouter.class);
ProducerCacheKey<String> key1 = new ProducerCacheKey<>(Schema.STRING, "topic1", router1);
List<ProducerInterceptor> interceptors1 = Collections.singletonList(mock(ProducerInterceptor.class));
ProducerCacheKey<String> 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));
}
}

33
spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarProducerFactoryTests.java
View File

@@ -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<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
try (Producer<String> 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<String> producerFactory = producerFactory(pulsarClient, Collections.emptyMap());
MessageRouter router = mock(MessageRouter.class);
try (Producer<String> 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<String> producerFactory = producerFactory(pulsarClient,
Collections.singletonMap("topicName", "topic0"));
try (Producer<String> 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<String> producer = producerFactory.createProducer(null, schema, router)) {
assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, router);
assertProducerHasTopicSchemaAndRouter(producer, "topic0", schema, router, null);
}
}
@Test
void createProducerWithDefaultTopicAndInterceptor() throws PulsarClientException {
PulsarProducerFactory<String> producerFactory = producerFactory(pulsarClient,
Collections.singletonMap("topicName", "topic0"));
List<ProducerInterceptor> interceptors = Collections.singletonList(mock(ProducerInterceptor.class));
try (Producer<String> 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<String> producer, String topic, Schema<String> schema,
MessageRouter router) {
MessageRouter router, List<ProducerInterceptor> 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);
}
}
/**

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

@@ -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<Arguments> sendMessageTestProvider() {
@ParameterizedTest(name = "{0}")
@MethodSource("interceptorInvocationTestProvider")
void interceptorInvocationTest(String topic, List<ProducerInterceptor> interceptors) throws Exception {
try (PulsarClient client = PulsarClient.builder().serviceUrl(getPulsarBrokerUrl()).build()) {
PulsarProducerFactory<String> producerFactory = new DefaultPulsarProducerFactory<>(client,
Collections.singletonMap("topicName", topic));
PulsarTemplate<String> pulsarTemplate = new PulsarTemplate<>(producerFactory, interceptors);
pulsarTemplate.send("test-interceptor");
for (ProducerInterceptor interceptor : interceptors) {
verify(interceptor, atLeastOnce()).eligible(any(Message.class));
}
}
}
private static Stream<Arguments> sendMessageTestProvider() {
return Stream.of(
arguments(Named.of("sendMessageToDefaultTopic", "smt-topic-1"),
@@ -175,6 +193,14 @@ class PulsarTemplateTests extends AbstractContainerBaseTests {
sampleMessageKeyCustomizer, mockRouter()));
}
private static Stream<Arguments> 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);