GH-1425: Configure ReplyPostProcessor via Factory

Resolves https://github.com/spring-projects/spring-amqp/issues/1425
This commit is contained in:
Gary Russell
2022-10-26 14:20:28 -04:00
committed by Artem Bilan
parent 0459c9907c
commit f2fc13b6e1
4 changed files with 54 additions and 2 deletions

View File

@@ -17,6 +17,7 @@
package org.springframework.amqp.rabbit.config;
import java.util.Arrays;
import java.util.function.Function;
import org.aopalliance.aop.Advice;
@@ -26,6 +27,7 @@ import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener;
import org.springframework.amqp.rabbit.listener.adapter.ReplyPostProcessor;
import org.springframework.amqp.utils.JavaUtils;
import org.springframework.lang.Nullable;
import org.springframework.retry.RecoveryCallback;
@@ -54,6 +56,8 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
private Advice[] adviceChain;
private Function<String, ReplyPostProcessor> replyPostProcessorProvider;
@Override
public abstract C createListenerContainer(RabbitListenerEndpoint endpoint);
@@ -108,6 +112,17 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
this.recoveryCallback = recoveryCallback;
}
/**
* Set a function to provide a reply post processor; it will be used if there is no
* replyPostProcessor on the rabbit listener annotation. The input parameter is the
* listener id.
* @param replyPostProcessorProvider the post processor.
* @since 3.0
*/
public void setReplyPostProcessorProvider(Function<String, ReplyPostProcessor> replyPostProcessorProvider) {
this.replyPostProcessorProvider = replyPostProcessorProvider;
}
protected void applyCommonOverrides(@Nullable RabbitListenerEndpoint endpoint, C instance) {
if (endpoint != null) { // endpoint settings overriding default factory settings
JavaUtils.INSTANCE
@@ -130,6 +145,11 @@ public abstract class BaseRabbitListenerContainerFactory<C extends MessageListen
.acceptIfNotNull(endpoint.getReplyPostProcessor(), messageListener::setReplyPostProcessor)
.acceptIfNotNull(endpoint.getReplyContentType(), messageListener::setReplyContentType);
messageListener.setConverterWinsContentType(endpoint.isConverterWinsContentType());
if (endpoint.getReplyPostProcessor() == null && this.replyPostProcessorProvider != null) {
JavaUtils.INSTANCE
.acceptIfNotNull(this.replyPostProcessorProvider.apply(endpoint.getId()),
messageListener::setReplyPostProcessor);
}
}
}
}

View File

@@ -237,6 +237,9 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
@Autowired
private MultiListenerValidatedJsonBean multiValidated;
@Autowired
private ReplyPostProcessor rpp;
@BeforeAll
public static void setUp() {
System.setProperty(RabbitListenerAnnotationBeanPostProcessor.RABBIT_EMPTY_STRING_ARGUMENTS_PROPERTY,
@@ -310,6 +313,8 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
this.registry.start();
assertThat(listenerContainer.isRunning()).isTrue();
listenerContainer.stop();
assertThat(listenerContainer.getMessageListener()).extracting("replyPostProcessor")
.isSameAs(this.rpp);
}
@Test
@@ -1690,14 +1695,22 @@ public class EnableRabbitIntegrationTests extends NeedsManagementTests {
}
@Bean
public SimpleRabbitListenerContainerFactory rabbitAutoStartFalseListenerContainerFactory() {
public SimpleRabbitListenerContainerFactory rabbitAutoStartFalseListenerContainerFactory(
ReplyPostProcessor rpp) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(rabbitConnectionFactory());
factory.setReceiveTimeout(10L);
factory.setAutoStartup(false);
factory.setReplyPostProcessorProvider(id -> rpp);
return factory;
}
@Bean
ReplyPostProcessor rpp() {
return (in, out) -> out;
}
@Bean
public SimpleRabbitListenerContainerFactory jsonListenerContainerFactory() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();

View File

@@ -3144,6 +3144,22 @@ public ReplyPostProcessor echoCustomHeader() {
----
====
Starting with version 3.0, you can configure the post processor on the container factory instead of on the annotation.
====
[source, java]
----
factory.setReplyPostProcessorProvider(id -> (req, resp) -> {
resp.getMessageProperties().setHeader("myHeader", req.getMessageProperties().getHeader("myHeader"));
return resp;
});
----
====
The `id` parameter is the listener id.
A setting on the annotation will supersede the factory setting.
The `@SendTo` value is assumed as a reply `exchange` and `routingKey` pair that follows the `exchange/routingKey` pattern,
where one of those parts can be omitted.
The valid values are as follows:

View File

@@ -37,7 +37,10 @@ When setting the container factory `consumerBatchEnabled` to `true`, the `batchL
See <<receiving-batch>> for more infoprmation.
`MessageConverter` s can now return `Optional.empty()` for a null value; this is currently implemented by the `Jackson2JsonMessageConverter`.
See <<Jackson2JsonMessageConverter-from-message>> for more information.
See <<Jackson2JsonMessageConverter-from-message>> for more information
You can now configure a `ReplyPostProcessor` via the container factory rather than via a property on `@RabbitListener`.
See <<async-annotation-driven-reply>> for more information.
==== Connection Factory Changes