Polish "Add ProducerInterceptor support"
* Add producer interceptor docs (author added in previous commit but were lost during rebase) * Add test for ordering of multiple configured interceptors Closes #35
This commit is contained in:
@@ -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:
|
||||
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user