From eca2314ea99db33bde71ad80e5204e3eb96f68e7 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 3 Oct 2014 19:05:35 +0300 Subject: [PATCH] AMQP-429: Change `@SendTo` Logic to `Address` JIRA: https://jira.spring.io/browse/AMQP-429 Since `@SendTo` can apply only `String` values, but from other side `Address` can parse that `String`, there is no stops to use `Address` object as a `replyToAddress`. Now to send a result of `@RabbitListener` method to some specific `replyTo` you can specify it like this: ``` @RabbitListener(queues = "test.sendTo") @SendTo("direct:///test.sendTo.reply") public String capitalizeAndSendTo(String foo) { return foo.toUpperCase(); } ``` * Add integration test * In addition remove redundant `@RabbitListener.responseRoutingKey()` AMQP-429: Revert a legacy code Note: `@SendTo("")` is a valid use case, when we just want to send message to the Default Exchange with default `routingKey`. Hence fix `MethodRabbitListenerEndpointTests#emptySendTo()` to expect valid result instead of `Exception` Fix `Address` to treat empty String as `null` AMQP-429: Pushed the change for empty `@SendTo` AMQP-429: Fix `Address` overhead with `ExchangeType` * Provide new `Address` algorithm to parse String argument * Fix tests * Add note to Docs regarding `@SendTo` AMQP-429: Provide backward compatibility for `Address` Remove redundant test for `returns` Polishing --- .../springframework/amqp/core/Address.java | 65 +++++++++++-------- .../amqp/core/AddressTests.java | 35 +++++++--- .../amqp/core/MessagePropertiesTests.java | 2 +- .../amqp/remoting/RemotingTest.java | 7 +- .../rabbit/annotation/RabbitListener.java | 10 --- ...itListenerAnnotationBeanPostProcessor.java | 4 +- .../AbstractRabbitListenerEndpoint.java | 22 ------- .../config/MethodRabbitListenerEndpoint.java | 20 +++--- .../amqp/rabbit/core/RabbitTemplate.java | 4 +- .../AbstractAdaptableMessageListener.java | 2 +- .../AbstractRabbitAnnotationDrivenTests.java | 17 +++-- .../EnableRabbitIntegrationTests.java | 32 +++++++-- .../MethodRabbitListenerEndpointTests.java | 43 +++++++----- .../core/RabbitTemplateIntegrationTests.java | 2 +- ...atePublisherCallbacksIntegrationTests.java | 31 --------- src/reference/docbook/amqp.xml | 22 +++++++ 16 files changed, 169 insertions(+), 149 deletions(-) diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java b/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java index 2490b994..e8b61875 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/Address.java @@ -19,28 +19,26 @@ import java.util.regex.Pattern; import org.springframework.util.StringUtils; /** - * Represents an address for publication of an AMQP message. The AMQP 0-8 and 0-9 specifications have an unstructured - * string that is used as a "reply to" address. There are however conventions in use and this class makes it easier to + * Represents an address for publication of an AMQP message. The AMQP 0-8 and 0-9 + * specifications have an unstructured string that is used as a "reply to" address. + * There are however conventions in use and this class makes it easier to * follow these conventions, which can be easily summarised as: * * 
- * (exchangeType)://(exchange)/(routingKey)
+ * (exchange)/(routingKey)
  *
* - * Here we also allow the exchange type to default to direct, and the exchange name to default to empty (so just a - * routing key will work if you know the queue name). - * - * @see ExchangeTypes + * Here we also the exchange name to default to empty + * (so just a routing key will work if you know the queue name). * * @author Mark Pollack * @author Mark Fisher * @author Dave Syer + * @author Artem Bilan */ public class Address { - private static final Pattern pattern = Pattern.compile("^([^:]+)://([^/]*)/?(.*)$"); - - private final String exchangeType; + private static final Pattern pattern = Pattern.compile("^(?:.*://)?([^/]*)/?(.*)$"); private final String exchangeName; @@ -50,28 +48,28 @@ public class Address { * Create an Address instance from a structured String in the form * * 
-	 * (exchangeType)://(exchange)/(routingKey)
+	 * (exchange)/(routingKey)
 	 *
* - * where examples of valid exchangeType values can be found in the {@link ExchangeTypes} static - * constants. - * * @param address a structured string. */ public Address(String address) { - if (address == null) { - this.exchangeType = ExchangeTypes.DIRECT; + if (!StringUtils.hasText(address)) { this.exchangeName = ""; this.routingKey = ""; - } else { + } + else if (address.lastIndexOf('/') <= 0) { + this.routingKey = address.replaceFirst("/", ""); + this.exchangeName = ""; + } + else { Matcher matcher = pattern.matcher(address); boolean matchFound = matcher.find(); if (matchFound) { - this.exchangeType = matcher.group(1); - this.exchangeName = matcher.group(2); - this.routingKey = matcher.group(3); - } else { - this.exchangeType = ExchangeTypes.DIRECT; + this.exchangeName = matcher.group(1); + this.routingKey = matcher.group(2); + } + else { this.exchangeName = ""; this.routingKey = address; } @@ -79,21 +77,32 @@ public class Address { } /*** - * Create an Address given the exchange type, exchange name and routing key. This will set the exchange type, name - * and the routing key explicitly. - * + * Create an Address given the exchange type, exchange name and routing key. + * This will set the exchange type, name and the routing key explicitly. * @param exchangeType The exchange type. * @param exchangeName The exchange name. * @param routingKey The routing key. */ + @Deprecated public Address(String exchangeType, String exchangeName, String routingKey) { - this.exchangeType = exchangeType; this.exchangeName = exchangeName; this.routingKey = routingKey; } + /*** + * Create an Address given the exchange name and routing key. + * This will set the exchange type, name and the routing key explicitly. + * @param exchangeName The exchange name. + * @param routingKey The routing key. + */ + public Address(String exchangeName, String routingKey) { + this.exchangeName = exchangeName; + this.routingKey = routingKey; + } + + @Deprecated public String getExchangeType() { - return this.exchangeType; + return null; } public String getExchangeName() { @@ -106,7 +115,7 @@ public class Address { @Override public String toString() { - StringBuilder sb = new StringBuilder(this.exchangeType + "://" + this.exchangeName + "/"); + StringBuilder sb = new StringBuilder(this.exchangeName + "/"); if (StringUtils.hasText(this.routingKey)) { sb.append(this.routingKey); } diff --git a/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java b/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java index 316687a5..07578b51 100644 --- a/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java +++ b/spring-amqp/src/test/java/org/springframework/amqp/core/AddressTests.java @@ -24,13 +24,14 @@ import org.junit.Test; /** * @author Mark Pollack * @author Mark Fisher + * @author Artem Bilan */ public class AddressTests { @Test public void toStringCheck() { - Address address = new Address(ExchangeTypes.DIRECT, "my-exchange", "routing-key"); - String replyToUri = "direct://my-exchange/routing-key"; + Address address = new Address("my-exchange", "routing-key"); + String replyToUri = "my-exchange/routing-key"; Assert.assertEquals(replyToUri, address.toString()); } @@ -38,34 +39,48 @@ public class AddressTests { public void parse() { String replyToUri = "direct://my-exchange/routing-key"; Address address = new Address(replyToUri); - assertEquals(address.getExchangeType(), ExchangeTypes.DIRECT); - assertEquals(address.getExchangeName(), "my-exchange"); - assertEquals(address.getRoutingKey(), "routing-key"); + assertEquals("my-exchange", address.getExchangeName()); + assertEquals("routing-key", address.getRoutingKey()); } @Test public void parseUnstructuredWithRoutingKeyOnly() { Address address = new Address("my-routing-key"); assertEquals("my-routing-key", address.getRoutingKey()); - assertEquals("direct:///my-routing-key", address.toString()); + assertEquals("/my-routing-key", address.toString()); + + address = new Address("/foo"); + assertEquals("foo", address.getRoutingKey()); + assertEquals("/foo", address.toString()); + + address = new Address("bar/baz"); + assertEquals("bar", address.getExchangeName()); + assertEquals("baz", address.getRoutingKey()); + assertEquals("bar/baz", address.toString()); } @Test public void parseWithoutRoutingKey() { Address address = new Address("fanout://my-exchange"); - assertEquals(ExchangeTypes.FANOUT, address.getExchangeType()); assertEquals("my-exchange", address.getExchangeName()); assertEquals("", address.getRoutingKey()); - assertEquals("fanout://my-exchange/", address.toString()); + assertEquals("my-exchange/", address.toString()); } @Test public void parseWithDefaultExchangeAndRoutingKey() { Address address = new Address("direct:///routing-key"); - assertEquals(ExchangeTypes.DIRECT, address.getExchangeType()); assertEquals("", address.getExchangeName()); assertEquals("routing-key", address.getRoutingKey()); - assertEquals("direct:///routing-key", address.toString()); + assertEquals("/routing-key", address.toString()); + } + + @Test + public void testEmpty() { + Address address = new Address("/"); + assertEquals("", address.getExchangeName()); + assertEquals("", address.getRoutingKey()); + assertEquals("/", address.toString()); } } diff --git a/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java b/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java index bf078ace..3604434b 100644 --- a/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java +++ b/spring-amqp/src/test/java/org/springframework/amqp/core/MessagePropertiesTests.java @@ -29,7 +29,7 @@ public class MessagePropertiesTests { @Test public void testReplyTo() throws Exception { MessageProperties properties = new MessageProperties(); - properties.setReplyTo("fanout://foo/bar"); + properties.setReplyTo("foo/bar"); assertEquals("bar", properties.getReplyToAddress().getRoutingKey()); } diff --git a/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java b/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java index e3228089..8c200413 100644 --- a/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java +++ b/spring-amqp/src/test/java/org/springframework/amqp/remoting/RemotingTest.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-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 @@ -61,7 +61,7 @@ public class RemotingTest { public Object convertSendAndReceive(Object payload) throws AmqpException { MessageConverter messageConverter = serviceExporter.getMessageConverter(); - Address replyTo = new Address("fakeExchange", "fakeExchangeName", "fakeRoutingKey"); + Address replyTo = new Address("fakeExchangeName", "fakeRoutingKey"); MessageProperties messageProperties = new MessageProperties(); messageProperties.setReplyToAddress(replyTo); Message message = messageConverter.toMessage(payload, messageProperties); @@ -89,6 +89,7 @@ public class RemotingTest { } @Test(expected = GeneralException.class) + @SuppressWarnings("ThrowableResultOfMethodCallIgnored") public void testExceptionReturningMethod() { riggedProxy.notReallyExceptionReturningMethod(); } @@ -96,8 +97,6 @@ public class RemotingTest { @Test public void testActuallyExceptionReturningMethod() { SpecialException returnedException = riggedProxy.actuallyExceptionReturningMethod(); - Assert.assertNotNull(returnedException); - Assert.assertTrue(returnedException instanceof SpecialException); } } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java index 984ebeac..29f3581d 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListener.java @@ -118,16 +118,6 @@ public @interface RabbitListener { */ String priority() default ""; - /** - * The routing key to send along with a response message. - *

This will be applied in case of a request message that does not carry - * a "replyTo" property. Note: This only applies to a listener method with - * a return value, for which each result object will be converted into a - * response message. - * @return the response routing key. - */ - String responseRoutingKey() default ""; - /** * Reference to a {@link org.springframework.amqp.rabbit.core.RabbitAdmin * RabbitAdmin}. Required if the listener is using auto-delete diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java index 5a998e7c..76bd7834 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java @@ -237,9 +237,7 @@ public class RabbitListenerAnnotationBeanPostProcessor rabbitListener + " (must be an integer)", ex); } } - if (StringUtils.hasText(rabbitListener.responseRoutingKey())) { - endpoint.setResponseRoutingKey(resolve(rabbitListener.responseRoutingKey())); - } + String rabbitAdmin = resolve(rabbitListener.admin()); if (StringUtils.hasText(rabbitAdmin)) { Assert.state(this.beanFactory != null, "BeanFactory must be set to resolve RabbitAdmin by bean name"); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java index 0e45995c..421c6ec5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerEndpoint.java @@ -27,7 +27,6 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; -import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener; import org.springframework.util.Assert; /** @@ -50,8 +49,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn private Integer priority; - private String responseRoutingKey; - private RabbitAdmin admin; @@ -135,21 +132,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn return priority; } - /** - * Set the routing key to send along with a response message. - * @param responseRoutingKey the response routing key value. - */ - public void setResponseRoutingKey(String responseRoutingKey) { - this.responseRoutingKey = responseRoutingKey; - } - - /** - * @return the routing key to send along with a response message. - */ - public String getResponseRoutingKey() { - return responseRoutingKey; - } - /** * Set the {@link RabbitAdmin} instance to use. * @param admin the {@link RabbitAdmin} instance. @@ -207,9 +189,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn private void setupMessageListener(MessageListenerContainer container) { MessageListener messageListener = createMessageListener(container); - if (getResponseRoutingKey() != null && messageListener instanceof AbstractAdaptableMessageListener) { - ((AbstractAdaptableMessageListener) messageListener).setResponseRoutingKey(getResponseRoutingKey()); - } Assert.state(messageListener != null, "Endpoint [" + this + "] must provide a non null message listener"); container.setupMessageListener(messageListener); } @@ -225,7 +204,6 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn append("' | queueNames='").append(this.queueNames). append("' | exclusive='").append(this.exclusive). append("' | priority='").append(this.priority). - append("' | responseRoutingKey='").append(this.responseRoutingKey). append("' | admin='").append(this.admin).append("'"); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java index d3f2fe4d..196d073e 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * 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. @@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.config; import java.lang.reflect.Method; import java.util.Arrays; +import org.springframework.amqp.core.Address; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter; import org.springframework.amqp.support.converter.MessageConverter; @@ -27,13 +28,13 @@ import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.messaging.handler.invocation.InvocableHandlerMethod; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * A {@link RabbitListenerEndpoint} providing the method to invoke to process * an incoming message for this endpoint. * * @author Stephane Nicoll + * @author Artem Bilan * @since 1.4 */ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint { @@ -88,9 +89,10 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint InvocableHandlerMethod invocableHandlerMethod = this.messageHandlerMethodFactory.createInvocableHandlerMethod(getBean(), getMethod()); messageListener.setHandlerMethod(invocableHandlerMethod); - String responseExchange = getDefaultResponseExchange(); - if (StringUtils.hasText(responseExchange)) { - messageListener.setResponseExchange(responseExchange); + Address replyToAddress = getDefaultReplyToAddress(); + if (replyToAddress != null) { + messageListener.setResponseExchange(replyToAddress.getExchangeName()); + messageListener.setResponseRoutingKey(replyToAddress.getRoutingKey()); } MessageConverter messageConverter = container.getMessageConverter(); if (messageConverter != null) { @@ -107,15 +109,15 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint return new MessagingMessageListenerAdapter(); } - private String getDefaultResponseExchange() { + private Address getDefaultReplyToAddress() { SendTo ann = AnnotationUtils.getAnnotation(getMethod(), SendTo.class); if (ann != null) { - Object[] destinations = ann.value(); - if (destinations.length != 1) { + String[] destinations = ann.value(); + if (destinations.length > 1) { throw new IllegalStateException("Invalid @" + SendTo.class.getSimpleName() + " annotation on '" + getMethod() + "' one destination must be set (got " + Arrays.toString(destinations) + ")"); } - return (String) destinations[0]; + return destinations.length == 1 ? new Address(destinations[0]) : new Address(null); } return null; } 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 292cdd6a..1ba602b1 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 @@ -544,7 +544,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, @Override public Address getReplyToAddress(Message request, S reply) { - return new Address(null, replyExchange, replyRoutingKey); + return new Address(replyExchange, replyRoutingKey); } }); @@ -976,7 +976,7 @@ public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, "Cannot determine ReplyTo message property value: " + "Request message does not contain reply-to property, and no default Exchange was set."); } - replyTo = new Address(null, this.exchange, this.routingKey); + replyTo = new Address(this.exchange, this.routingKey); } return replyTo; } 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 99d39f48..9fe11a2e 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 @@ -280,7 +280,7 @@ public abstract class AbstractAdaptableMessageListener implements MessageListene "Request message does not contain reply-to property, " + "and no default response Exchange was set."); } - replyTo = new Address(null, this.responseExchange, this.responseRoutingKey); + replyTo = new Address(this.responseExchange, this.responseRoutingKey); } return replyTo; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java index f9c0fc5b..0c234409 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/AbstractRabbitAnnotationDrivenTests.java @@ -16,9 +16,15 @@ package org.springframework.amqp.rabbit.annotation; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; + import java.util.Collection; -import com.rabbitmq.client.Channel; import org.junit.Rule; import org.junit.Test; import org.junit.rules.ExpectedException; @@ -40,8 +46,7 @@ import org.springframework.validation.Errors; import org.springframework.validation.Validator; import org.springframework.validation.annotation.Validated; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; +import com.rabbitmq.client.Channel; /** * @@ -117,7 +122,6 @@ public abstract class AbstractRabbitAnnotationDrivenTests { assertTrue("No queue instances should be set", endpoint.getQueues().isEmpty()); assertEquals(true, endpoint.isExclusive()); assertEquals(new Integer(34), endpoint.getPriority()); - assertEquals("routing-123", endpoint.getResponseRoutingKey()); assertSame(context.getBean("rabbitAdmin"), endpoint.getAdmin()); // Resolve the container and invoke a message on it @@ -142,7 +146,7 @@ public abstract class AbstractRabbitAnnotationDrivenTests { static class FullBean { @RabbitListener(id = "listener1", containerFactory = "simpleFactory", queues = {"queue1", "queue2"}, - exclusive = true, priority = "34", responseRoutingKey = "routing-123", admin = "rabbitAdmin") + exclusive = true, priority = "34", admin = "rabbitAdmin") public void fullHandle(String msg) { } @@ -153,8 +157,7 @@ public abstract class AbstractRabbitAnnotationDrivenTests { @RabbitListener(id = "${rabbit.listener.id}", containerFactory = "${rabbit.listener.containerFactory}", queues = {"${rabbit.listener.queue}", "queue2"}, exclusive = true, - priority = "${rabbit.listener.priority}", responseRoutingKey = "${rabbit.listener.responseRoutingKey}", - admin = "${rabbit.listener.admin}") + priority = "${rabbit.listener.priority}", admin = "${rabbit.listener.admin}") public void fullHandle(String msg) { } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index f74c6744..c37bd0c1 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * 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. @@ -16,6 +16,10 @@ package org.springframework.amqp.rabbit.annotation; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + import org.junit.ClassRule; import org.junit.Test; import org.junit.runner.RunWith; @@ -33,16 +37,17 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.handler.annotation.Header; import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import static org.junit.Assert.*; - /** * * @author Stephane Nicoll + * @author Artem Bilan + * @since 1.4 */ @ContextConfiguration(classes = EnableRabbitIntegrationTests.EnableRabbitConfig.class) @RunWith(SpringJUnit4ClassRunner.class) @@ -51,7 +56,7 @@ public class EnableRabbitIntegrationTests { @ClassRule public static final BrokerRunning brokerRunning = BrokerRunning.isRunningWithEmptyQueues( - "test.simple", "test.header", "test.message", "test.reply"); + "test.simple", "test.header", "test.message", "test.reply", "test.sendTo", "test.sendTo.reply"); @Autowired private RabbitTemplate rabbitTemplate; @@ -90,6 +95,19 @@ public class EnableRabbitIntegrationTests { assertEquals("Wrong bar header", "barValue", reply.getMessageProperties().getHeaders().get("bar")); } + @Test + public void simpleEndpointWithSendTo() throws InterruptedException { + rabbitTemplate.convertAndSend("test.sendTo", "bar"); + int n = 0; + Object result = null; + while ((result = rabbitTemplate.receiveAndConvert("test.sendTo.reply")) == null && n++ < 10) { + Thread.sleep(100); + } + assertTrue(n < 10); + assertNotNull(result); + assertEquals("BAR", result); + } + public static class MyService { @RabbitListener(queues = "test.simple") @@ -112,6 +130,12 @@ public class EnableRabbitIntegrationTests { return MessageBuilder.withPayload(payload) .setHeader("foo", foo).setHeader("bar", "barValue").build(); } + + @RabbitListener(queues = "test.sendTo") + @SendTo("test.sendTo.reply") + public String capitalizeAndSendTo(String foo) { + return foo.toUpperCase(); + } } @Configuration diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java index 75592cab..78a677ef 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MethodRabbitListenerEndpointTests.java @@ -16,14 +16,22 @@ package org.springframework.amqp.rabbit.config; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.AdditionalMatchers.aryEq; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.springframework.amqp.rabbit.test.MessageTestUtils.createTextMessage; + import java.io.Serializable; import java.lang.reflect.Method; import java.util.Arrays; import java.util.HashMap; import java.util.Map; -import com.rabbitmq.client.AMQP; -import com.rabbitmq.client.Channel; import org.hamcrest.Matchers; import org.junit.Before; import org.junit.Rule; @@ -32,7 +40,6 @@ import org.junit.rules.ExpectedException; import org.junit.rules.TestName; import org.mockito.ArgumentCaptor; -import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.listener.ListenerExecutionFailedException; @@ -59,11 +66,8 @@ import org.springframework.validation.Errors; import org.springframework.validation.Validator; import org.springframework.validation.annotation.Validated; -import static org.junit.Assert.*; -import static org.mockito.AdditionalMatchers.*; -import static org.mockito.Mockito.eq; -import static org.mockito.Mockito.*; -import static org.springframework.amqp.rabbit.test.MessageTestUtils.*; +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; /** @@ -275,7 +279,7 @@ public class MethodRabbitListenerEndpointTests { MessagingMessageListenerAdapter listener = createDefaultInstance(String.class); listener.setMandatoryPublish(true); String body = "echo text"; - Address replyTo = new Address(null, "replyToQueue", "myRouting"); + Address replyTo = new Address("replyToQueue", "myRouting"); MessageProperties properties = new MessageProperties(); properties.setReplyToAddress(replyTo); @@ -319,11 +323,13 @@ public class MethodRabbitListenerEndpointTests { public void emptySendTo() throws Exception { MessagingMessageListenerAdapter listener = createDefaultInstance(String.class); - Channel channel = mock(Channel.class); + processAndReply(listener, createTextMessage("content"), "", "", false, null); + assertDefaultListenerMethodInvocation(); + } - thrown.expect(ReplyFailureException.class); - thrown.expectCause(Matchers.isA(AmqpException.class)); - listener.onMessage(createTextMessage("content"), channel); + @Test + public void noSendToValue() throws Exception { + emptySendTo(); } @Test @@ -392,8 +398,7 @@ public class MethodRabbitListenerEndpointTests { endpoint.setBean(sample); endpoint.setMethod(method); endpoint.setMessageHandlerMethodFactory(factory); - MessagingMessageListenerAdapter messageListener = endpoint.createMessageListener(container); - return messageListener; + return endpoint.createMessageListener(container); } private MessagingMessageListenerAdapter createInstance( @@ -533,7 +538,7 @@ public class MethodRabbitListenerEndpointTests { return content; } - @SendTo("replyDestination") + @SendTo("replyDestination/") public String processAndReplyWithSendTo(String content) { invocations.put("processAndReplyWithSendTo", true); return content; @@ -545,6 +550,12 @@ public class MethodRabbitListenerEndpointTests { return content; } + @SendTo + public String noSendToValue(String content) { + invocations.put("noSendToValue", true); + return content; + } + @SendTo({"firstDestination", "secondDestination"}) public String invalidSendTo(String content) { invocations.put("invalidSendTo", true); 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 bf67c424..d311610e 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 @@ -856,7 +856,7 @@ public class RabbitTemplateIntegrationTests { public Message handle(Message message) { MessageProperties messageProperties = new MessageProperties(); messageProperties.setContentType(message.getMessageProperties().getContentType()); - messageProperties.setHeader("testReplyTo", new Address("", "", ROUTE)); + messageProperties.setHeader("testReplyTo", new Address("", ROUTE)); return new Message(message.getBody(), messageProperties); } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java index 2a78daca..4d3b1f0c 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplatePublisherCallbacksIntegrationTests.java @@ -743,35 +743,4 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { assertNull(templateWithConfirmsEnabled.getUnconfirmed(0)); } - @Test - public void testReturnNotReceivedAfterPublisherCallbackChannelClose() throws Exception { - final CountDownLatch latch = new CountDownLatch(20); - templateWithReturnsEnabled.setMandatory(true); - templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() { - - @Override - public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { - latch.countDown(); - } - - }); - - ExecutorService executorService = Executors.newCachedThreadPool(); - for (int i = 0; i < 20; i++) { - executorService.execute(new Runnable() { - - @Override - public void run() { - templateWithReturnsEnabled.convertAndSend("BAD_ROUTE", (Object) "bad", new CorrelationData("cba")); - } - - }); - } - - executorService.shutdown(); - assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS)); - Thread.sleep(100); - assertFalse(latch.getCount() == 0); - } - } diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index ec2a8dc9..58fa75ec 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -1250,6 +1250,28 @@ public Message processOrder(Order order) { .setHeader("code", 1234) .build(); }]]> + + The @SendTo value is assumed as a reply exchange and + routingKey pair following the pattern exchange/routingKey, where one of those + parts can be omitted. The valid values are: + + + foo/bar - the replyTo exchange and routingKey. + + + foo/ - the replyTo exchange and default (empty) routingKey. + + + bar or /bar - the replyTo routingKey and default (empty) exchange. + + + / or empty - the replyTo default exchange and default routingKey. + + + Also @SendTo can be used without a value attribute. This case is equal to an empty + sendTo pattern. @SendTo is only used if the inbound message does not have a + replyToAddress property. +