diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index d3854787..180feec0 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -531,6 +531,36 @@ public void listen(Foo foo) { On the producer side also, for the Java primitive types, the framework can infer the Schema, but for any other types, you need set that on the `PulsarTemmplate`. + +=== Intercepting messages + +==== Intercept messages on the Producer +Adding a `ProducerInterceptor` allows you to intercept and mutate messages received by the producer before being published to the brokers. +To do so, you can pass a list of interceptors into the `PulsarTemplate` constructor. +When using multiple interceptors, the order they are applied in will be the order they appear in the list. + +If you are using Spring Boot auto-configuration, you can simply specify the interceptors as Beans. +They will be passed automatically to the `PulsarTemplate`. +Ordering of the interceptors is achieved by using the `@Order` annotation as seen below. + +==== +[source, java] +---- +@Bean +@Order(100) +ProducerInterceptor firstInterceptor() { + ... +} + +@Bean +@Order(200) +ProducerInterceptor secondInterceptor() { + ... +} +---- +==== + + ==== Appendix The reference documentation has the following appendices: 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 e60652b6..343ebd7a 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 @@ -30,6 +30,9 @@ import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.FilteredClassLoader; import org.springframework.boot.test.context.assertj.AssertableApplicationContext; import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.annotation.Order; import org.springframework.pulsar.annotation.EnablePulsar; import org.springframework.pulsar.annotation.PulsarBootstrapConfiguration; import org.springframework.pulsar.annotation.PulsarListenerAnnotationBeanPostProcessor; @@ -167,6 +170,16 @@ class PulsarAutoConfigurationTests { .contains(interceptor))); } + @Test + void customProducerInterceptorsOrderedProperly() { + this.contextRunner.withUserConfiguration(InterceptorTestConfiguration.class) + .run((context -> assertThat(context).hasNotFailed().getBean(PulsarTemplate.class) + .extracting("interceptors") + .asInstanceOf(InstanceOfAssertFactories.list(ProducerInterceptor.class)) + .containsExactly(InterceptorTestConfiguration.interceptorBar, + InterceptorTestConfiguration.interceptorFoo))); + } + @Nested class ProducerFactoryAutoConfigurationTests { @@ -207,4 +220,24 @@ class PulsarAutoConfigurationTests { } + @Configuration(proxyBeanMethods = false) + static class InterceptorTestConfiguration { + + static ProducerInterceptor interceptorFoo = mock(ProducerInterceptor.class); + static ProducerInterceptor interceptorBar = mock(ProducerInterceptor.class); + + @Bean + @Order(200) + ProducerInterceptor interceptorFoo() { + return interceptorFoo; + } + + @Bean + @Order(100) + ProducerInterceptor interceptorBar() { + return interceptorBar; + } + + } + }