From ca32f3f464b489a85eac827675b7569e5f9c3576 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 5 Dec 2017 13:48:13 -0800 Subject: [PATCH] AMQP-790: Fix after receive MPPs with send/receive JIRA: https://jira.spring.io/browse/AMQP-790 Previously, `afterReceivePostProcessors` were not called on `sendAndReceive()` operations. # Conflicts: # spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java --- .../amqp/rabbit/junit/BrokerRunning.java | 1 + .../amqp/rabbit/core/RabbitTemplate.java | 33 ++-- .../RabbitTemplateMPPIntegrationTests.java | 157 ++++++++++++++++++ 3 files changed, 178 insertions(+), 13 deletions(-) create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java diff --git a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java index 07c6064e..0448b40d 100644 --- a/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java +++ b/spring-rabbit-junit/src/main/java/org/springframework/amqp/rabbit/junit/BrokerRunning.java @@ -551,6 +551,7 @@ public final class BrokerRunning extends TestWatcher { this.connectionFactory.setPort(this.port); this.connectionFactory.setUsername(this.user); this.connectionFactory.setPassword(this.password); + this.connectionFactory.setAutomaticRecoveryEnabled(false); } return this.connectionFactory; } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java index e9d5f629..9568c63b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/core/RabbitTemplate.java @@ -792,7 +792,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, } public void convertAndSend(String routingKey, Object message, MessagePostProcessor messagePostProcessor, - CorrelationData correlationData) + CorrelationData correlationData) throws AmqpException { convertAndSend(this.exchange, routingKey, message, messagePostProcessor, correlationData); } @@ -808,7 +808,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, Message messageToSend = convertMessageIfNecessary(message); messageToSend = messagePostProcessor instanceof CorrelationAwareMessagePostProcessor ? ((CorrelationAwareMessagePostProcessor) messagePostProcessor) - .postProcessMessage(messageToSend, correlationData) + .postProcessMessage(messageToSend, correlationData) : messagePostProcessor.postProcessMessage(messageToSend); send(exchange, routingKey, messageToSend, correlationData); } @@ -954,7 +954,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, @Override public boolean receiveAndReply(final String queueName, ReceiveAndReplyCallback callback, final String replyExchange, - final String replyRoutingKey) throws AmqpException { + final String replyRoutingKey) throws AmqpException { return this.receiveAndReply(queueName, callback, new ReplyToAddressCallback() { @Override @@ -973,13 +973,13 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, @Override public boolean receiveAndReply(String queueName, ReceiveAndReplyCallback callback, - ReplyToAddressCallback replyToAddressCallback) throws AmqpException { + ReplyToAddressCallback replyToAddressCallback) throws AmqpException { return doReceiveAndReply(queueName, callback, replyToAddressCallback); } @SuppressWarnings("unchecked") private boolean doReceiveAndReply(final String queueName, final ReceiveAndReplyCallback callback, - final ReplyToAddressCallback replyToAddressCallback) throws AmqpException { + final ReplyToAddressCallback replyToAddressCallback) throws AmqpException { return this.execute(new ChannelCallback() { @SuppressWarnings("deprecation") @@ -1212,7 +1212,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, if (messagePostProcessor != null) { requestMessage = messagePostProcessor instanceof CorrelationAwareMessagePostProcessor ? ((CorrelationAwareMessagePostProcessor) messagePostProcessor) - .postProcessMessage(requestMessage, correlationData) + .postProcessMessage(requestMessage, correlationData) : messagePostProcessor.postProcessMessage(requestMessage); } Message replyMessage = doSendAndReceive(exchange, routingKey, requestMessage, correlationData); @@ -1279,13 +1279,18 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, - byte[] body) throws IOException { + byte[] body) throws IOException { MessageProperties messageProperties = RabbitTemplate.this.messagePropertiesConverter .toMessageProperties(properties, envelope, RabbitTemplate.this.encoding); Message reply = new Message(body, messageProperties); if (logger.isTraceEnabled()) { logger.trace("Message received " + reply); } + if (RabbitTemplate.this.afterReceivePostProcessors != null) { + for (MessagePostProcessor processor : RabbitTemplate.this.afterReceivePostProcessors) { + reply = processor.postProcessMessage(reply); + } + } pendingReply.reply(reply); } }; @@ -1300,7 +1305,8 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, try { channel.basicCancel(consumerTag); } - catch (Exception e) { } + catch (Exception e) { + } } return reply; } @@ -1310,7 +1316,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, protected Message doSendAndReceiveWithFixed(final String exchange, final String routingKey, final Message message, final CorrelationData correlationData) { Assert.state(this.isListener, "RabbitTemplate is not configured as MessageListener - " - + "cannot use a 'replyAddress': " + this.replyAddress); + + "cannot use a 'replyAddress': " + this.replyAddress); return this.execute(new ChannelCallback() { @SuppressWarnings("deprecation") @@ -1516,7 +1522,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, for (MessagePostProcessor processor : this.beforePublishPostProcessors) { messageToUse = processor instanceof CorrelationAwareMessagePostProcessor ? ((CorrelationAwareMessagePostProcessor) processor) - .postProcessMessage(messageToUse, correlationData) + .postProcessMessage(messageToUse, correlationData) : processor.postProcessMessage(messageToUse); } } @@ -1564,6 +1570,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, private Message buildMessageFromDelivery(com.rabbitmq.client.QueueingConsumer.Delivery delivery) { return buildMessage(delivery.getEnvelope(), delivery.getProperties(), delivery.getBody(), -1); } + private Message buildMessageFromResponse(GetResponse response) { return buildMessage(response.getEnvelope(), response.getProps(), response.getBody(), response.getMessageCount()); } @@ -1640,7 +1647,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, else { throw new IllegalStateException( "Channel does not support confirms or returns; " + - "is the connection factory configured for confirms or returns?"); + "is the connection factory configured for confirms or returns?"); } } @@ -1663,7 +1670,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, String routingKey, BasicProperties properties, byte[] body) - throws IOException { + throws IOException { ReturnCallback returnCallback = this.returnCallback; if (returnCallback == null) { @@ -1766,7 +1773,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, else { if (savedCorrelation != null) { message.getMessageProperties().setHeader(this.correlationKey, - savedCorrelation); + savedCorrelation); } else { message.getMessageProperties().getHeaders().remove(this.correlationKey); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java new file mode 100644 index 00000000..0fffab63 --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateMPPIntegrationTests.java @@ -0,0 +1,157 @@ +/* + * Copyright 2017 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.core; + +import static org.junit.Assert.assertTrue; + +import org.junit.AfterClass; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.annotation.EnableRabbit; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.junit.BrokerRunning; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.annotation.DirtiesContext.ClassMode; +import org.springframework.test.context.junit4.SpringRunner; + +/** + * @author Gary Russell + * @since 1.7.6 + * + */ +@RunWith(SpringRunner.class) +@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD) +public class RabbitTemplateMPPIntegrationTests { + + private static final String QUEUE = "mpp.tests"; + + private static final String REPLIES = "mpp.tests.replies"; + + @ClassRule + public static BrokerRunning brokerIsRunning = BrokerRunning.isRunningWithEmptyQueues(QUEUE, REPLIES); + + @Autowired + private RabbitTemplate template; + + @Autowired + private Config config; + + @AfterClass + public static void tearDown() { + brokerIsRunning.removeTestQueues(); + } + + @Test + public void testMPPsAppliedDirectReplyToContainerTests() { + this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties())); + assertTrue("before MPP not called", this.config.beforeMppCalled); + assertTrue("after MPP not called", this.config.afterMppCalled); + } + + @Test + public void testMPPsAppliedDirectReplyToTests() { + this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties())); + assertTrue("before MPP not called", this.config.beforeMppCalled); + assertTrue("after MPP not called", this.config.afterMppCalled); + } + + @Test + public void testMPPsAppliedTemporaryReplyQueueTests() { + this.template.setUseTemporaryReplyQueues(true); + this.template.sendAndReceive(new Message("foo".getBytes(), new MessageProperties())); + assertTrue("before MPP not called", this.config.beforeMppCalled); + assertTrue("after MPP not called", this.config.afterMppCalled); + } + + @Test + public void testMPPsAppliedReplyContainerTests() { + 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())); + assertTrue("before MPP not called", this.config.beforeMppCalled); + assertTrue("after MPP not called", this.config.afterMppCalled); + } + finally { + container.stop(); + } + } + + @Configuration + @EnableRabbit + public static class Config { + + private boolean beforeMppCalled; + + private boolean afterMppCalled; + + @Bean + public CachingConnectionFactory cf() { + return new CachingConnectionFactory(brokerIsRunning.getConnectionFactory()); + } + + @Bean + public RabbitTemplate template() { + RabbitTemplate rabbitTemplate = new RabbitTemplate(cf()); + rabbitTemplate.setRoutingKey(QUEUE); + rabbitTemplate.setBeforePublishPostProcessors(m -> { + this.beforeMppCalled = true; + return m; + }); + rabbitTemplate.setAfterReceivePostProcessors(afterMPP()); + return rabbitTemplate; + } + + @Bean + public MessagePostProcessor afterMPP() { + return m -> { + this.afterMppCalled = true; + return m; + }; + } + + @Bean + public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() { + SimpleRabbitListenerContainerFactory cf = new SimpleRabbitListenerContainerFactory(); + cf.setConnectionFactory(cf()); + return cf; + } + + @RabbitListener(queues = QUEUE) + public byte[] foo(byte[] in) { + return in; + } + + } + +}