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

AMQP-448 Support 3.4.x Direct reply-to

Browse Source 
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.<base64string>`. Because `/` is a
valid `base64` character, we can't use normal reply-to address decoding
`<exchange>/<routingKey>`. 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
This commit is contained in:
Gary Russell
2014-12-08 19:14:14 +02:00
committed by Artem Bilan
parent c501d214d5
commit f08df63feb
6 changed files with 191 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-amqp/src/main/java/org/springframework/amqp/core/AddressUtils.java Normal file
View File

@@ -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;
}
}

4
spring-amqp/src/main/java/org/springframework/amqp/remoting/service/AmqpInvokerServiceExporter.java
View File

@@ -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</a>.
*
* @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");
}

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

@@ -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<Void>() {
@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(

3
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
View File

@@ -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(

58
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java
View File

@@ -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<Void>() {
@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<String> replyToWas = new AtomicReference<String>();
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() {

27
src/reference/docbook/amqp.xml
View File

@@ -1591,7 +1591,8 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]></programlist
is applied to both the request and reply. Those methods are named <methodname>convertSendAndReceive</methodname>.
See the Javadoc of <classname>AmqpTemplate</classname> for more detail.</para>
<para>
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 <xref linkend="direct-reply-to" />).
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 &lt;reply-listener/&gt; sub element. This element provides a listener container for the
reply queue, with the template being the listener. All of the <xref linkend="containerAttributes" /> attributes
@@ -1607,10 +1608,29 @@ Object receiveAndConvert(String queueName) throws AmqpException;]]></programlist
While the container and template share a connection factory, they do not share a channel and therefore requests
and replies are not performed within the same transaction (if transactional).
</para>
<section id="direct-reply-to">
<title>RabbitMQ Direct reply-to</title>
<important>
Starting the <emphasis>version 3.4.0</emphasis>, the RabbitMQ server now supports
<ulink url="http://www.rabbitmq.com/direct-reply-to.html">Direct reply-to</ulink>; this eliminates the main reason
for a fixed reply queue (to avoid the need to create a temporary queue for each request).
Starting with <emphasis>Spring AMQP version 1.4.1</emphasis> Direct reply-to will be used by default
(if supported by the server) instead of creating temporary reply queues.
When no <code>replyQueue</code> is provided (or it is set with the name <code>amqp.rabbitmq.reply-to</code>),
the <classname>RabbitTemplate</classname> 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 <code>reply-listener</code> is not required and should not be configured.
</important>
<para>
Reply listeners are still supported with named queues (other than <code>amq.rabbitmq.reply-to</code>),
allowing control of reply concurrency etc.
</para>
</section>
<section>
<title>Message Correlation With A Reply Queue</title>
<para>
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 <code>amqp.rabbitmq.reply-to</code>),
it is necessary to provide correlation data so that replies can be correlated
to requests. See <ulink url="http://www.rabbitmq.com/tutorials/tutorial-six-java.html">
RabbitMQ Remote Procedure Call (RPC)</ulink>. By default, the standard <code>correlationId</code> 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;]]></programlist
the parser defines the container and wires in the template as the listener.
</para>
<note>
When the template does not use a fixed <code>replyQueue</code>, a listener container is not needed.
When the template does not use a fixed <code>replyQueue</code> (or is using Direct reply-to - see
<xref linkend="direct-reply-to" />) a listener container is not needed.
</note>
<para>
If you define your <classname>RabbitTemplate</classname> as a <code>&lt;bean/&gt;</code>, or using an