From f08df63febb22ee81b54a09ec05fb37f112d5a47 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 8 Dec 2014 19:14:14 +0200 Subject: [PATCH] AMQP-448 Support 3.4.x Direct reply-to JIRA: https://jira.spring.io/browse/AMQP-448 RabbitMQ now supports "Direct reply-to" where an RPC sender can specify `amq.rabbitmq.reply-to` and the broker provides special handling to return the reply to the sending channel. This is more efficient than creating a temporary reply queue each time. The `RabbitTemplate` has provided an alternative to creating new reply queues via a reply message listener container and correlation data. With this change, the template now supports "Direct reply-to" and will use it instead of temporary reply queues when no reply-listener is provided. Tested on a server with 3.4.1 RabbitMQ as well as one with 3.1.1 for backwards compatibility. __NOTE:__ On the receiving side, the message `reply_to` property has the form `amq.rabbitmq.reply-to.`. Because `/` is a valid `base64` character, we can't use normal reply-to address decoding `/`. The `MessageListenerAdapter` therefore detects that Direct reply-to is being used before decoding the address. Polishing Move reply address decoding to a static method in `Address` and invoke from all places where it is needed. Causes a minor leak of rabbitmq into spring-amqp (the name of the direct reply queue). AMQP-448 Polishing; PR Comments --- .../amqp/core/AddressUtils.java | 48 +++++++++++++++ .../service/AmqpInvokerServiceExporter.java | 4 +- .../amqp/rabbit/core/RabbitTemplate.java | 61 +++++++++++++++++-- .../AbstractAdaptableMessageListener.java | 3 +- .../core/RabbitTemplateIntegrationTests.java | 58 ++++++++++++++++++ src/reference/docbook/amqp.xml | 27 +++++++- 6 files changed, 191 insertions(+), 10 deletions(-) create mode 100644 spring-amqp/src/main/java/org/springframework/amqp/core/AddressUtils.java diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/AddressUtils.java b/spring-amqp/src/main/java/org/springframework/amqp/core/AddressUtils.java new file mode 100644 index 00000000..b5534151 --- /dev/null +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/AddressUtils.java @@ -0,0 +1,48 @@ +/* + * Copyright 2014 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.core; + +/** + * @author Gary Russell + * @since 1.4.1 + * + */ +public class AddressUtils { + + public static final String AMQ_RABBITMQ_REPLY_TO = "amq.rabbitmq.reply-to"; + + /** + * Decodes the reply-to {@link Address} into exchange/key. + * + * @param request the inbound message. + * @return the Address. + */ + public static Address decodeReplyToAddress(Message request) { + Address replyTo; + String replyToString = request.getMessageProperties().getReplyTo(); + if (replyToString == null) { + replyTo = null; + } + else if (replyToString.startsWith(AMQ_RABBITMQ_REPLY_TO)) { + replyTo = new Address("", replyToString); + } + else { + replyTo = new Address(replyToString); + } + return replyTo; + } + +} diff --git a/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java b/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java index 05c9a07c..1b6cfdcf 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java @@ -15,6 +15,7 @@ package org.springframework.amqp.remoting.service; import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AddressUtils; import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; @@ -48,6 +49,7 @@ import org.springframework.remoting.support.RemoteInvocationResult; * "http://static.springsource.org/spring-amqp/reference/html/amqp.html#request-reply" >here. * * @author David Bilge + * @author Gary Russell * @since 1.2 */ public class AmqpInvokerServiceExporter extends RemoteInvocationBasedExporter implements MessageListener { @@ -58,7 +60,7 @@ public class AmqpInvokerServiceExporter extends RemoteInvocationBasedExporter im @Override public void onMessage(Message message) { - Address replyToAddress = message.getMessageProperties().getReplyToAddress(); + Address replyToAddress = AddressUtils.decodeReplyToAddress(message); if (replyToAddress == null) { throw new AmqpRejectAndDontRequeueException("No replyToAddress in inbound AMQP Message"); } 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 1bafac5d..471bf900 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 @@ -32,6 +32,7 @@ import org.springframework.amqp.AmqpException; import org.springframework.amqp.AmqpIllegalStateException; import org.springframework.amqp.AmqpRejectAndDontRequeueException; import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AddressUtils; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessagePostProcessor; @@ -177,6 +178,10 @@ public class RabbitTemplate extends RabbitAccessor private volatile Expression receiveConnectionFactorySelectorExpression; + private volatile boolean usingFastReplyTo; + + private volatile boolean evaluatedFastReplyTo; + /** * Convenient constructor for use with setter injection. Don't forget to set the connection factory. */ @@ -244,12 +249,14 @@ public class RabbitTemplate extends RabbitAccessor /** * A queue for replies; if not provided, a temporary exclusive, auto-delete queue will - * be used for each reply. + * be used for each reply, unless RabbitMQ supports 'amq.rabbitmq.reply-to' - see + * http://www.rabbitmq.com/direct-reply-to.html * * @param replyQueue the replyQueue to set */ public void setReplyQueue(Queue replyQueue) { this.replyQueue = replyQueue; + this.evaluatedFastReplyTo = false; } /** @@ -449,6 +456,37 @@ public class RabbitTemplate extends RabbitAccessor return unconfirmed.size() > 0 ? unconfirmed : null; } + private void evaluateFastReplyTo() { + this.usingFastReplyTo = false; + if (this.replyQueue == null || AddressUtils.AMQ_RABBITMQ_REPLY_TO.equals(this.replyQueue.getName())) { + try { + execute(new ChannelCallback() { + + @Override + public Void doInRabbit(Channel channel) throws Exception { + channel.queueDeclarePassive(AddressUtils.AMQ_RABBITMQ_REPLY_TO); + return null; + } + }); + this.usingFastReplyTo = true; + } + catch (Exception e) { + if (replyQueue != null) { + logger.error("Broker does not support fast replies via 'amq.rabbitmq.reply-to', temporary " + + "queues will be used:" + e.getMessage() + "."); + } + else { + if (logger.isDebugEnabled()) { + logger.debug("Broker does not support fast replies via 'amq.rabbitmq.reply-to', temporary " + + "queues will be used:" + e.getMessage() + "."); + } + } + RabbitTemplate.this.replyQueue = null; + } + } + this.evaluatedFastReplyTo = true; + } + @Override public void send(Message message) throws AmqpException { send(this.exchange, this.routingKey, message); @@ -813,7 +851,14 @@ public class RabbitTemplate extends RabbitAccessor * @return the message that is received in reply */ protected Message doSendAndReceive(final String exchange, final String routingKey, final Message message) { - if (this.replyQueue == null) { + if (!this.evaluatedFastReplyTo) { + synchronized(this) { + if (!this.evaluatedFastReplyTo) { + evaluateFastReplyTo(); + } + } + } + if (this.replyQueue == null || this.usingFastReplyTo) { return doSendAndReceiveWithTemporary(exchange, routingKey, message); } else { @@ -830,8 +875,14 @@ public class RabbitTemplate extends RabbitAccessor Assert.isNull(message.getMessageProperties().getReplyTo(), "Send-and-receive methods can only be used if the Message does not already have a replyTo property."); - DeclareOk queueDeclaration = channel.queueDeclare(); - String replyTo = queueDeclaration.getQueue(); + String replyTo; + if (RabbitTemplate.this.usingFastReplyTo) { + replyTo = AddressUtils.AMQ_RABBITMQ_REPLY_TO; + } + else { + DeclareOk queueDeclaration = channel.queueDeclare(); + replyTo = queueDeclaration.getQueue(); + } message.getMessageProperties().setReplyTo(replyTo); String consumerTag = UUID.randomUUID().toString(); @@ -1069,7 +1120,7 @@ public class RabbitTemplate extends RabbitAccessor * @see org.springframework.amqp.core.MessageProperties#getReplyTo() */ private Address getReplyToAddress(Message request) throws AmqpException { - Address replyTo = request.getMessageProperties().getReplyToAddress(); + Address replyTo = AddressUtils.decodeReplyToAddress(request); if (replyTo == null) { if (this.exchange == null) { throw new AmqpException( diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java index 9fe11a2e..3a56b1fe 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java @@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AddressUtils; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessageProperties; @@ -272,7 +273,7 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene * @see org.springframework.amqp.core.MessageProperties#getReplyTo() */ protected Address getReplyToAddress(Message request) throws Exception { - Address replyTo = request.getMessageProperties().getReplyToAddress(); + Address replyTo = AddressUtils.decodeReplyToAddress(request); if (replyTo == null) { if (this.responseExchange == null) { throw new AmqpException( diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index d3f2f057..62863854 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -13,10 +13,13 @@ package org.springframework.amqp.rabbit.core; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.doAnswer; @@ -37,6 +40,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.junit.After; import org.junit.Before; import org.junit.Rule; @@ -47,6 +51,7 @@ import org.mockito.stubbing.Answer; import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Address; +import org.springframework.amqp.core.AddressUtils; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; @@ -57,6 +62,7 @@ import org.springframework.amqp.core.ReplyToAddressCallback; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.SingleConnectionFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.rabbit.test.BrokerRunning; @@ -91,6 +97,8 @@ import com.rabbitmq.client.GetResponse; */ public class RabbitTemplateIntegrationTests { + private static final Log logger = LogFactory.getLog(RabbitTemplateIntegrationTests.class); + private static final String ROUTE = "test.queue"; private static final Queue REPLY_QUEUE = new Queue("test.reply.queue"); @@ -1046,6 +1054,56 @@ public class RabbitTemplateIntegrationTests { assertEquals(messageId, new String(result.getMessageProperties().getCorrelationId())); } + @Test + public void testSendAndReceiveFastImplicit() { + sendAndReceiveFastGuts(); + } + + @Test + public void testSendAndReceiveFastExplicit() { + this.template.setReplyQueue(new Queue(AddressUtils.AMQ_RABBITMQ_REPLY_TO)); + sendAndReceiveFastGuts(); + } + + private void sendAndReceiveFastGuts() { + try { + this.template.execute(new ChannelCallback() { + + @Override + public Void doInRabbit(Channel channel) throws Exception { + channel.queueDeclarePassive(AddressUtils.AMQ_RABBITMQ_REPLY_TO); + return null; + } + }); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(this.template.getConnectionFactory()); + container.setQueueNames(ROUTE); + final AtomicReference replyToWas = new AtomicReference(); + MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new Object() { + + @SuppressWarnings("unused") + public Message handleMessage(Message message) { + replyToWas.set(message.getMessageProperties().getReplyTo()); + return new Message(new String(message.getBody()).toUpperCase().getBytes(), + message.getMessageProperties()); + } + }); + messageListenerAdapter.setMessageConverter(null); + container.setMessageListener(messageListenerAdapter); + container.start(); + this.template.setQueue(ROUTE); + this.template.setRoutingKey(ROUTE); + Object result = this.template.convertSendAndReceive("foo"); + container.stop(); + assertEquals("FOO", result); + assertThat(replyToWas.get(), startsWith(AddressUtils.AMQ_RABBITMQ_REPLY_TO)); + } + catch (Exception e) { + assertThat(e.getCause().getCause().getMessage(), containsString("404")); + logger.info("Broker does not support fast replies; test skipped " + e.getMessage()); + } + } + @SuppressWarnings("serial") private class PlannedException extends RuntimeException { public PlannedException() { diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index 7f77de6f..bd7644f5 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -1591,7 +1591,8 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]>convertSendAndReceive. See the Javadoc of AmqpTemplate for more detail. - By default, a new temporary queue is used for each reply. However, a single reply queue can be configured on the template, + By default, a new temporary queue is used for each reply (but see ). + However, a single reply queue can be configured on the template, which can be more efficient, and also allows you to set arguments on that queue. In this case, however, you must also provide a <reply-listener/> sub element. This element provides a listener container for the reply queue, with the template being the listener. All of the attributes @@ -1607,10 +1608,29 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]> +
+ RabbitMQ Direct reply-to + + Starting the version 3.4.0, the RabbitMQ server now supports + Direct reply-to; this eliminates the main reason + for a fixed reply queue (to avoid the need to create a temporary queue for each request). + Starting with Spring AMQP version 1.4.1 Direct reply-to will be used by default + (if supported by the server) instead of creating temporary reply queues. + When no replyQueue is provided (or it is set with the name amqp.rabbitmq.reply-to), + the RabbitTemplate will automatically + detect whether Direct reply-to is supported and either use it or fall back to using a temporary reply queue. + When using Direct reply-to, a reply-listener is not required and should not be configured. + + + Reply listeners are still supported with named queues (other than amq.rabbitmq.reply-to), + allowing control of reply concurrency etc. + +
Message Correlation With A Reply Queue - When using a fixed reply queue, it is necessary to provide correlation data so that replies can be correlated + When using a fixed reply queue (other than amqp.rabbitmq.reply-to), + it is necessary to provide correlation data so that replies can be correlated to requests. See RabbitMQ Remote Procedure Call (RPC). By default, the standard correlationId property will be used to hold the correlation data. However, if you wish to use a custom propertry to hold correlation @@ -1632,7 +1652,8 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]> - When the template does not use a fixed replyQueue, a listener container is not needed. + When the template does not use a fixed replyQueue (or is using Direct reply-to - see + ) a listener container is not needed. If you define your RabbitTemplate as a <bean/>, or using an