diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/NamespaceUtils.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/NamespaceUtils.java index 99bcedbe..1b0f7354 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/NamespaceUtils.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/NamespaceUtils.java @@ -17,6 +17,7 @@ import java.util.List; import org.w3c.dom.Element; +import org.springframework.amqp.rabbit.support.ExpressionFactoryBean; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.BeanDefinitionHolder; import org.springframework.beans.factory.config.BeanReference; @@ -26,9 +27,11 @@ import org.springframework.beans.factory.parsing.BeanComponentDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.ManagedList; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.core.Conventions; +import org.springframework.expression.common.LiteralExpression; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; @@ -289,4 +292,51 @@ public abstract class NamespaceUtils { } } + public static BeanDefinition createExpressionDefinitionFromValueOrExpression(String valueElementName, + String expressionElementName, ParserContext parserContext, Element element, boolean oneRequired) { + + Assert.hasText(valueElementName, "'valueElementName' must not be empty"); + Assert.hasText(expressionElementName, "'expressionElementName' must not be empty"); + + String valueElementValue = element.getAttribute(valueElementName); + String expressionElementValue = element.getAttribute(expressionElementName); + + boolean hasAttributeValue = StringUtils.hasText(valueElementValue); + boolean hasAttributeExpression = StringUtils.hasText(expressionElementValue); + + if (hasAttributeValue && hasAttributeExpression){ + parserContext.getReaderContext().error("Only one of '" + valueElementName + "' or '" + + expressionElementName + "' is allowed", element); + } + + if (oneRequired && (!hasAttributeValue && !hasAttributeExpression)){ + parserContext.getReaderContext().error("One of '" + valueElementName + "' or '" + + expressionElementName + "' is required", element); + } + BeanDefinition expressionDef = null; + if (hasAttributeValue) { + expressionDef = new RootBeanDefinition(LiteralExpression.class); + expressionDef.getConstructorArgumentValues().addGenericArgumentValue(valueElementValue); + } + else { + expressionDef = createExpressionDefIfAttributeDefined(expressionElementName, element); + } + return expressionDef; + } + + public static BeanDefinition createExpressionDefIfAttributeDefined(String expressionElementName, Element element) { + + Assert.hasText(expressionElementName, "'expressionElementName' must no be empty"); + + String expressionElementValue = element.getAttribute(expressionElementName); + + if (StringUtils.hasText(expressionElementValue)){ + BeanDefinitionBuilder expressionDefBuilder = + BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class); + expressionDefBuilder.addConstructorArgValue(expressionElementValue); + return expressionDefBuilder.getRawBeanDefinition(); + } + return null; + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java index 086267df..b07449a6 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/TemplateParser.java @@ -29,6 +29,7 @@ import org.springframework.util.xml.DomUtils; /** * @author Dave Syer * @author Gary Russell + * @author Artem Bilan */ class TemplateParser extends AbstractSingleBeanDefinitionParser { @@ -54,8 +55,6 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { private static final String MANDATORY_ATTRIBUTE = "mandatory"; - private static final String IMMEDIATE_ATTRIBUTE = "immediate"; - private static final String RETURN_CALLBACK_ATTRIBUTE = "return-callback"; private static final String CONFIRM_CALLBACK_ATTRIBUTE = "confirm-callback"; @@ -101,13 +100,18 @@ class TemplateParser extends AbstractSingleBeanDefinitionParser { NamespaceUtils.setValueIfAttributeDefined(builder, element, ENCODING_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, MESSAGE_CONVERTER_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, REPLY_QUEUE_ATTRIBUTE); - NamespaceUtils.setValueIfAttributeDefined(builder, element, MANDATORY_ATTRIBUTE); - NamespaceUtils.setValueIfAttributeDefined(builder, element, IMMEDIATE_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, RETURN_CALLBACK_ATTRIBUTE); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, CONFIRM_CALLBACK_ATTRIBUTE); NamespaceUtils.setValueIfAttributeDefined(builder, element, CORRELATION_KEY); NamespaceUtils.setReferenceIfAttributeDefined(builder, element, RETRY_TEMPLATE); + BeanDefinition expressionDef = + NamespaceUtils.createExpressionDefinitionFromValueOrExpression(MANDATORY_ATTRIBUTE, + "mandatory-expression", parserContext, element, false); + if (expressionDef != null) { + builder.addPropertyValue("mandatoryExpression", expressionDef); + } + BeanDefinition replyContainer = null; Element childElement = null; List childElements = DomUtils.getChildElementsByTagName(element, LISTENER_ELEMENT); 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 608d6c20..292cdd6a 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 @@ -52,8 +52,16 @@ import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.rabbit.support.PendingConfirm; import org.springframework.amqp.rabbit.support.PublisherCallbackChannel; import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator; +import org.springframework.amqp.rabbit.support.ValueExpression; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.context.expression.BeanFactoryResolver; +import org.springframework.context.expression.MapAccessor; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.support.StandardEvaluationContext; import org.springframework.retry.RetryCallback; import org.springframework.retry.RetryContext; import org.springframework.retry.support.RetryTemplate; @@ -104,7 +112,7 @@ import com.rabbitmq.client.GetResponse; * @author Artem Bilan * @since 1.0 */ -public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, MessageListener, +public class RabbitTemplate extends RabbitAccessor implements BeanFactoryAware, RabbitOperations, MessageListener, PublisherCallbackChannel.Listener { /** Alias for amq.direct default exchange */ @@ -139,9 +147,10 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, private volatile ReturnCallback returnCallback; - private final ConcurrentHashMap> pendingConfirms = new ConcurrentHashMap>(); + private final ConcurrentHashMap> pendingConfirms = + new ConcurrentHashMap>(); - private volatile boolean mandatory; + private volatile Expression mandatoryExpression = new ValueExpression(false); private final String uuid = UUID.randomUUID().toString(); @@ -149,6 +158,7 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, private volatile RetryTemplate retryTemplate; + private final StandardEvaluationContext evaluationContext = new StandardEvaluationContext(); private final ReplyToAddressCallback defaultReplyToAddressCallback = new ReplyToAddressCallback() { @@ -159,7 +169,6 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, }; - /** * Convenient constructor for use with setter injection. Don't forget to set the connection factory. */ @@ -303,7 +312,18 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, } public void setMandatory(boolean mandatory) { - this.mandatory = mandatory; + this.mandatoryExpression = new ValueExpression(mandatory); + } + + /** + * @param mandatoryExpression a SpEL {@link Expression} to evaluate against each request + * message, if a {@link #returnCallback} has been provided. The result of expression must be + * a {@code boolean} value. + * @since 1.4 + */ + public void setMandatoryExpression(Expression mandatoryExpression) { + Assert.notNull(mandatoryExpression, "'mandatoryExpression' must not be null"); + this.mandatoryExpression = mandatoryExpression; } /** @@ -320,13 +340,18 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, /** * Add a {@link RetryTemplate} which will be used for all rabbit operations. - * * @param retryTemplate The retry template. */ public void setRetryTemplate(RetryTemplate retryTemplate) { this.retryTemplate = retryTemplate; } + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory)); + this.evaluationContext.addPropertyAccessor(new MapAccessor()); + } + /** * Gets unconfirmed correlatiom data older than age and removes them. * @param age in millseconds @@ -877,7 +902,8 @@ public class RabbitTemplate extends RabbitAccessor implements RabbitOperations, publisherCallbackChannel.addPendingConfirm(this, channel.getNextPublishSeqNo(), new PendingConfirm(correlationData, System.currentTimeMillis())); } - boolean mandatory = this.returnCallback != null && this.mandatory; + boolean mandatory = this.returnCallback != null && + this.mandatoryExpression.getValue(this.evaluationContext, message, Boolean.class); MessageProperties messageProperties = message.getMessageProperties(); if (mandatory) { messageProperties.getHeaders().put(PublisherCallbackChannel.RETURN_CORRELATION, this.uuid); diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java new file mode 100644 index 00000000..c12c7bb0 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ExpressionFactoryBean.java @@ -0,0 +1,64 @@ +/* + * Copyright 2002-2010 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.support; + +import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.expression.Expression; +import org.springframework.expression.ExpressionParser; +import org.springframework.expression.spel.SpelParserConfiguration; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.util.Assert; + +/** + * FactoryBean for creating Expression instances. + * + * @author Mark Fisher + * @since 1.4 + */ +public class ExpressionFactoryBean extends AbstractFactoryBean { + + private final static ExpressionParser DEFAULT_PARSER = new SpelExpressionParser(); + + + private final String expressionString; + + private volatile ExpressionParser parser = DEFAULT_PARSER; + + + public ExpressionFactoryBean(String expressionString) { + Assert.hasText(expressionString, "expressionString must not be empty or null"); + this.expressionString = expressionString; + } + + + public void setParserConfiguration(SpelParserConfiguration parserConfiguration) { + Assert.notNull(parserConfiguration, "parserConfiguration must not be null"); + this.parser = new SpelExpressionParser(parserConfiguration); + } + + + @Override + public Class getObjectType() { + return Expression.class; + } + + @Override + protected Expression createInstance() throws Exception { + return this.parser.parseExpression(this.expressionString); + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ValueExpression.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ValueExpression.java new file mode 100644 index 00000000..7447ee99 --- /dev/null +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/ValueExpression.java @@ -0,0 +1,175 @@ +/* + * 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.rabbit.support; + +import org.springframework.core.convert.TypeDescriptor; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.EvaluationException; +import org.springframework.expression.Expression; +import org.springframework.expression.TypedValue; +import org.springframework.util.Assert; + +/** + * A very simple hardcoded implementation of the {@link org.springframework.expression.Expression} + * interface that represents an immutable value. + * It is used as value holder in the context of expression evaluation. + * + * @param - The expected value type. + * + * @author Artem Bilan + * @since 1.4 + */ +public class ValueExpression implements Expression { + + /** Fixed value of this expression */ + private final V value; + + private final Class aClass; + + private final TypedValue typedResultValue; + + private final TypeDescriptor typeDescriptor; + + @SuppressWarnings("unchecked") + public ValueExpression(V value) { + Assert.notNull(value); + this.value = value; + this.aClass = (Class) this.value.getClass(); + this.typedResultValue = new TypedValue(this.value); + this.typeDescriptor = this.typedResultValue.getTypeDescriptor(); + } + + @Override + public V getValue() throws EvaluationException { + return this.value; + } + + @Override + public V getValue(Object rootObject) throws EvaluationException { + return this.value; + } + + @Override + public V getValue(EvaluationContext context) throws EvaluationException { + return this.value; + } + + @Override + public V getValue(EvaluationContext context, Object rootObject) throws EvaluationException { + return this.value; + } + + @Override + public T getValue(Object rootObject, Class desiredResultType) throws EvaluationException { + return getValue(desiredResultType); + } + + @Override + public T getValue(Class desiredResultType) throws EvaluationException { + return org.springframework.expression.common.ExpressionUtils + .convertTypedValue(null, this.typedResultValue, desiredResultType); + } + + @Override + public T getValue(EvaluationContext context, Object rootObject, Class desiredResultType) + throws EvaluationException { + return getValue(context, desiredResultType); + } + + @Override + public T getValue(EvaluationContext context, Class desiredResultType) throws EvaluationException { + return org.springframework.expression.common.ExpressionUtils + .convertTypedValue(context, this.typedResultValue, desiredResultType); + } + + @Override + public Class getValueType() throws EvaluationException { + return this.aClass; + } + + @Override + public Class getValueType(Object rootObject) throws EvaluationException { + return this.aClass; + } + + @Override + public Class getValueType(EvaluationContext context) throws EvaluationException { + return this.aClass; + } + + @Override + public Class getValueType(EvaluationContext context, Object rootObject) throws EvaluationException { + return this.aClass; + } + + @Override + public TypeDescriptor getValueTypeDescriptor() throws EvaluationException { + return this.typeDescriptor; + } + + @Override + public TypeDescriptor getValueTypeDescriptor(Object rootObject) throws EvaluationException { + return this.typeDescriptor; + } + + @Override + public TypeDescriptor getValueTypeDescriptor(EvaluationContext context) throws EvaluationException { + return this.typeDescriptor; + } + + @Override + public TypeDescriptor getValueTypeDescriptor(EvaluationContext context, Object rootObject) + throws EvaluationException { + return this.typeDescriptor; + } + + @Override + public boolean isWritable(EvaluationContext context) throws EvaluationException { + return false; + } + + @Override + public boolean isWritable(EvaluationContext context, Object rootObject) throws EvaluationException { + return false; + } + + @Override + public boolean isWritable(Object rootObject) throws EvaluationException { + return false; + } + + @Override + public void setValue(EvaluationContext context, Object value) throws EvaluationException { + setValue(context, null, value); + } + + @Override + public void setValue(Object rootObject, Object value) throws EvaluationException { + setValue(null, rootObject, value); + } + + @Override + public void setValue(EvaluationContext context, Object rootObject, Object value) throws EvaluationException { + throw new EvaluationException(this.value.toString(), "Cannot call setValue() on a ValueExpression"); + } + + @Override + public String getExpressionString() { + return this.value.toString(); + } + +} diff --git a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd index 45b2f565..8e7912f9 100644 --- a/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd +++ b/spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd @@ -976,11 +976,25 @@ - + + + + + + + + + diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java index 704e62bf..e8e77ed3 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/TemplateParserTests.java @@ -29,6 +29,7 @@ import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.converter.SerializerMessageConverter; +import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.beans.factory.xml.XmlBeanDefinitionReader; @@ -39,7 +40,7 @@ import org.springframework.retry.support.RetryTemplate; * * @author Dave Syer * @author Gary Russell - * + * @author Artem Bilan */ public final class TemplateParserTests { @@ -56,20 +57,25 @@ public final class TemplateParserTests { public void testTemplate() throws Exception { AmqpTemplate template = beanFactory.getBean("template", AmqpTemplate.class); assertNotNull(template); - DirectFieldAccessor dfa = new DirectFieldAccessor(template); - assertEquals(Boolean.FALSE, dfa.getPropertyValue("mandatory")); - assertNull(dfa.getPropertyValue("returnCallback")); - assertNull(dfa.getPropertyValue("confirmCallback")); + assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(template, "mandatoryExpression.value")); + assertNull(TestUtils.getPropertyValue(template, "returnCallback")); + assertNull(TestUtils.getPropertyValue(template, "confirmCallback")); } @Test public void testTemplateWithCallbacks() throws Exception { AmqpTemplate template = beanFactory.getBean("withCallbacks", AmqpTemplate.class); assertNotNull(template); - DirectFieldAccessor dfa = new DirectFieldAccessor(template); - assertEquals(Boolean.TRUE, dfa.getPropertyValue("mandatory")); - assertNotNull(dfa.getPropertyValue("returnCallback")); - assertNotNull(dfa.getPropertyValue("confirmCallback")); + assertEquals("true", TestUtils.getPropertyValue(template, "mandatoryExpression.literalValue")); + assertNotNull(TestUtils.getPropertyValue(template, "returnCallback")); + assertNotNull(TestUtils.getPropertyValue(template, "confirmCallback")); + } + + @Test + public void testTemplateWithMandatoryExpression() throws Exception { + AmqpTemplate template = beanFactory.getBean("withMandatoryExpression", AmqpTemplate.class); + assertNotNull(template); + assertEquals("'true'", TestUtils.getPropertyValue(template, "mandatoryExpression.expression")); } @Test 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 0a90cbb5..16bde278 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 @@ -68,6 +68,8 @@ import org.springframework.amqp.rabbit.test.BrokerTestUtils; import org.springframework.amqp.support.converter.SimpleMessageConverter; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; @@ -153,8 +155,8 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { @Override public Void doInRabbit(Channel channel) throws Exception { - Map listenerMap = TestUtils.getPropertyValue(((ChannelProxy) channel).getTargetChannel(), "listenerForSeq", - Map.class); + Map listenerMap = TestUtils.getPropertyValue(((ChannelProxy) channel).getTargetChannel(), "listenerForSeq", + Map.class); int n = 0; while (n++ < 100 && listenerMap.size() > 0) { Thread.sleep(100); @@ -193,12 +195,13 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { public Object doInRabbit(Channel channel) throws Exception { try { threadLatch.await(10, TimeUnit.SECONDS); - } catch (InterruptedException e) { + } + catch (InterruptedException e) { Thread.currentThread().interrupt(); } templateWithConfirmsEnabled.doSend(channel, "", ROUTE, - new SimpleMessageConverter().toMessage("message", new MessageProperties()), - new CorrelationData("def")); + new SimpleMessageConverter().toMessage("message", new MessageProperties()), + new CorrelationData("def")); return null; } }); @@ -259,6 +262,28 @@ public class RabbitTemplatePublisherCallbacksIntegrationTests { assertEquals("message", new String(message.getBody(), "utf-8")); } + @Test + public void testPublisherReturnsWithMandatoryExpression() throws Exception { + final CountDownLatch latch = new CountDownLatch(1); + final List returns = new ArrayList(); + templateWithReturnsEnabled.setReturnCallback(new ReturnCallback() { + @Override + public void returnedMessage(Message message, int replyCode, + String replyText, String exchange, String routingKey) { + returns.add(message); + latch.countDown(); + } + }); + Expression mandatoryExpression = new SpelExpressionParser().parseExpression("'message'.bytes == body"); + templateWithReturnsEnabled.setMandatoryExpression(mandatoryExpression); + templateWithReturnsEnabled.convertAndSend(ROUTE + "junk", (Object) "message", new CorrelationData("abc")); + templateWithReturnsEnabled.convertAndSend(ROUTE + "junk", (Object) "foo", new CorrelationData("abc")); + assertTrue(latch.await(1000, TimeUnit.MILLISECONDS)); + assertEquals(1, returns.size()); + Message message = returns.get(0); + assertEquals("message", new String(message.getBody(), "utf-8")); + } + @Test public void testPublisherConfirmNotReceived() throws Exception { ConnectionFactory mockConnectionFactory = mock(ConnectionFactory.class); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/Tester.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/Tester.java new file mode 100644 index 00000000..7905dc3a --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/Tester.java @@ -0,0 +1,92 @@ +/* + * Copyright 2013 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 java.io.IOException; +import java.util.Collections; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.concurrent.Executors; + +import org.junit.Test; + +import org.springframework.amqp.AmqpIOException; + +import com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.ConfirmListener; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +public class Tester { + + @Test + public void testDealLockOnConfirmChannelClose() throws Exception { + ConnectionFactory factory = new ConnectionFactory(); + Connection conn = factory.newConnection(); + final Channel channel = conn.createChannel(); + + final SortedSet unconfirmedSet = Collections.synchronizedSortedSet(new TreeSet()); + + channel.addConfirmListener(new ConfirmListener() { + public void handleAck(long seqNo, boolean multiple) { + System.out.println(seqNo + " " + multiple); + if (multiple) { + unconfirmedSet.headSet(seqNo + 1).clear(); + } + else { + unconfirmedSet.remove(seqNo); + } + } + + public void handleNack(long seqNo, boolean multiple) { + // handle the lost messages somehow + } + }); + + channel.confirmSelect(); + for (long i = 0; i < 10; ++i) { + unconfirmedSet.add(channel.getNextPublishSeqNo()); + channel.basicPublish("", "test.queue", new AMQP.BasicProperties.Builder().build(), "nop".getBytes()); + } + + while (unconfirmedSet.size() > 0) + Thread.sleep(100); + + Executors.newSingleThreadExecutor().execute(new Runnable() { + @Override + public void run() { + try { + channel.close(); + } + catch (IOException e) { + throw new AmqpIOException(e); + } + } + }); + + assertTrue(unconfirmedSet.isEmpty()); + conn.close(); + } + +} diff --git a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml index 4a740cb7..c94d207e 100644 --- a/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml +++ b/spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml @@ -27,6 +27,9 @@ + + diff --git a/src/reference/docbook/amqp.xml b/src/reference/docbook/amqp.xml index bdfa5be7..bac265a7 100644 --- a/src/reference/docbook/amqp.xml +++ b/src/reference/docbook/amqp.xml @@ -774,6 +774,12 @@ public static MessagePropertiesBuilder fromClonedProperties(MessageProperties pr When the template's mandatory property is 'true' returned messages are provided by the callback described in . + + + Starting with version 1.4 the RabbitTemplate supports + the SpEL mandatoryExpression property, which is evaluated against each request message, as the root + evaluation object, resolving to a boolean value. Bean references, such as + "@myBean.isMandatory(#root)" can be used in the expression. diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 6fc6cabe..503ffdf0 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -88,6 +88,15 @@ See . +
+ RabbitTemplate: mandatoryExpression + + The mandatoryExpression SpEL Expression property + has been added to the RabbitTemplate to evaluate a mandatory + boolean value against each request message, when a ReturnCallback is in use. + See . + +