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

GH-1359: After Receive Post Processor Invoked x2

Browse Source 
Resolves https://github.com/spring-projects/spring-amqp/issues/1359

When the `RabbitTemplate` is configured with `afterReceivePostProcessors`
and uses the default internal `DirectReplyToMessageListenerContainer`, the
postprocessors are applied twice, once by the container and once by the
template.

The template should not propagate the post processors into the container.

**cherry-pick to 2.3.x, 2.2.x**
This commit is contained in:
Gary Russell
2021-06-25 09:50:36 -04:00
committed by Artem Bilan
parent b7fb6991ae
commit 10929b7b40
2 changed files with 13 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java
View File

@@ -1980,10 +1980,6 @@ public class RabbitTemplate extends RabbitAccessor // NOSONAR type line count
if (this.taskExecutor != null) {
container.setTaskExecutor(this.taskExecutor);
}
if (this.afterReceivePostProcessors != null) {
container.setAfterReceivePostProcessors(this.afterReceivePostProcessors
.toArray(new MessagePostProcessor[this.afterReceivePostProcessors.size()]));
}
container.setNoLocal(this.noLocalReplyConsumer);
if (this.replyErrorHandler != null) {
container.setErrorHandler(this.replyErrorHandler);

23
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2021 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -59,43 +59,46 @@ public class RabbitTemplateMPPIntegrationTests {
@Autowired
private Config config;
@Test // 2.0.x only
@Test
public void testMPPsAppliedDirectReplyToContainerTests() {
this.config.afterMppCalled = 0;
this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties()));
assertThat(this.config.beforeMppCalled).as("before MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isEqualTo(1);
}
@Test
public void testMPPsAppliedDirectReplyToTests() {
this.config.afterMppCalled = 0;
this.template.setUseDirectReplyToContainer(false);
this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties()));
assertThat(this.config.beforeMppCalled).as("before MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isEqualTo(1);
}
@Test
public void testMPPsAppliedTemporaryReplyQueueTests() {
this.config.afterMppCalled = 0;
this.template.setUseDirectReplyToContainer(false);
this.template.setUseTemporaryReplyQueues(true);
this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties()));
assertThat(this.config.beforeMppCalled).as("before MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isEqualTo(1);
}
@Test
public void testMPPsAppliedReplyContainerTests() {
this.config.afterMppCalled = 0;
this.template.setReplyAddress(REPLIES);
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(this.config.cf());
try {
container.setQueueNames(REPLIES);
container.setMessageListener(this.template);
container.setAfterReceivePostProcessors(this.config.afterMPP());
container.afterPropertiesSet();
container.start();
this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties()));
assertThat(this.config.beforeMppCalled).as("before MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isTrue();
assertThat(this.config.afterMppCalled).as("after MPP not called").isEqualTo(1);
}
finally {
container.stop();
@@ -106,9 +109,9 @@ public class RabbitTemplateMPPIntegrationTests {
@EnableRabbit
public static class Config {
private boolean beforeMppCalled;
boolean beforeMppCalled;
private boolean afterMppCalled;
int afterMppCalled;
@Bean
public CachingConnectionFactory cf() {
@@ -131,7 +134,7 @@ public class RabbitTemplateMPPIntegrationTests {
@Bean
public MessagePostProcessor afterMPP() {
return m -> {
this.afterMppCalled = true;
this.afterMppCalled++;
return m;
};
}