AMQP-273: mandatoryExpression for RabbitTemplate

JIRA: https://jira.spring.io/browse/AMQP-273

Polishing

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

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<Element> childElements = DomUtils.getChildElementsByTagName(element, LISTENER_ELEMENT);

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<Object, SortedMap<Long, PendingConfirm>> pendingConfirms = new ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>>();
+	private final ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>> pendingConfirms =
+			new ConcurrentHashMap<Object, SortedMap<Long, PendingConfirm>>();
 
-	private volatile boolean mandatory;
+	private volatile Expression mandatoryExpression = new ValueExpression<Boolean>(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<Object>() {
 
@@ -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<Boolean>(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);

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<Expression> {
+
+	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);
+	}
+
+}

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 <V> - The expected value type.
+ *
+ * @author Artem Bilan
+ * @since 1.4
+ */
+public class ValueExpression<V> implements Expression {
+
+	/** Fixed value of this expression */
+	private final V value;
+
+	private final Class<V> aClass;
+
+	private final TypedValue typedResultValue;
+
+	private final TypeDescriptor typeDescriptor;
+
+	@SuppressWarnings("unchecked")
+	public ValueExpression(V value) {
+		Assert.notNull(value);
+		this.value = value;
+		this.aClass = (Class<V>) 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> T getValue(Object rootObject, Class<T> desiredResultType) throws EvaluationException {
+		return getValue(desiredResultType);
+	}
+
+	@Override
+	public <T> T getValue(Class<T> desiredResultType) throws EvaluationException {
+		return org.springframework.expression.common.ExpressionUtils
+				.convertTypedValue(null, this.typedResultValue, desiredResultType);
+	}
+
+	@Override
+	public <T> T getValue(EvaluationContext context, Object rootObject, Class<T> desiredResultType)
+			throws EvaluationException {
+		return getValue(context, desiredResultType);
+	}
+
+	@Override
+	public <T> T getValue(EvaluationContext context, Class<T> desiredResultType) throws EvaluationException {
+		return org.springframework.expression.common.ExpressionUtils
+				.convertTypedValue(context, this.typedResultValue, desiredResultType);
+	}
+
+	@Override
+	public Class<V> getValueType() throws EvaluationException {
+		return this.aClass;
+	}
+
+	@Override
+	public Class<V> getValueType(Object rootObject) throws EvaluationException {
+		return this.aClass;
+	}
+
+	@Override
+	public Class<V> getValueType(EvaluationContext context) throws EvaluationException {
+		return this.aClass;
+	}
+
+	@Override
+	public Class<V> 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();
+	}
+
+}

spring-rabbit/src/main/resources/org/springframework/amqp/rabbit/config/spring-rabbit-1.4.xsd

@@ -976,11 +976,25 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="mandatory" type="xsd:string" use="optional">
+			<xsd:attribute name="mandatory" use="optional">
 				<xsd:annotation>
 					<xsd:documentation><![CDATA[
 	When 'true' sets the mandatory flag on basic.publish; only applies if
-	a 'return-callback' is provided.
+	a 'return-callback' is provided. Mutually exclusive with 'mandatory-expression'.
+	Defaults to 'false'.
+					]]></xsd:documentation>
+				</xsd:annotation>
+				<xsd:simpleType>
+					<xsd:union memberTypes="xsd:boolean xsd:string" />
+				</xsd:simpleType>
+			</xsd:attribute>
+			<xsd:attribute name="mandatory-expression" use="optional" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation><![CDATA[
+	A SpEL expression to be evaluated against each request message to determine a 'mandatory' boolean value.
+	The BeanFactoryResolver is available too, if the RabbitTemplate is used from Spring Context,
+	allowing for expressions such as '@myBean.isMandatory(#root)`.
+	Only applies if a 'return-callback' is provided. Mutually exclusive with 'mandatory'.
 					]]></xsd:documentation>
 				</xsd:annotation>
 			</xsd:attribute>

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

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<Message> returns = new ArrayList<Message>();
+		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);

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<Long> unconfirmedSet = Collections.<Long>synchronizedSortedSet(new TreeSet<Long>());
+
+		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();
+	}
+
+}

spring-rabbit/src/test/resources/org/springframework/amqp/rabbit/config/TemplateParserTests-context.xml

@@ -27,6 +27,9 @@
 	<rabbit:template id="withCallbacks" connection-factory="connectionFactory"
 		mandatory="true" return-callback="rcb" confirm-callback="ccb" />
 
+	<rabbit:template id="withMandatoryExpression" connection-factory="connectionFactory"
+		mandatory-expression="'true'"/>
+
 	<beans:bean id="rcb" class="org.mockito.Mockito" factory-method="mock">
 		<beans:constructor-arg value="org.springframework.amqp.rabbit.core.RabbitTemplate$ReturnCallback" />
 	</beans:bean>

src/reference/docbook/amqp.xml

@@ -774,6 +774,12 @@ public static MessagePropertiesBuilder fromClonedProperties(MessageProperties pr
       <para>
         When the template's <code>mandatory</code> property is 'true' returned messages are provided
         by the callback described in <xref linkend="amqp-template"/>.
+	  </para>
+	  <para>
+		Starting with <emphasis>version 1.4</emphasis> the <classname>RabbitTemplate</classname> supports
+		the SpEL <code>mandatoryExpression</code> property, which is evaluated against each request message, as the root
+		evaluation object, resolving to a <code>boolean</code> value. Bean references, such as
+		<code>"@myBean.isMandatory(#root)"</code> can be used in the expression.
       </para>
     </section>
   </section>

src/reference/docbook/whats-new.xml

@@ -88,6 +88,15 @@
 				See <xref linkend="lc-auto-delete"/>.
 			</para>
 		</section>
+		<section>
+			<title>RabbitTemplate: mandatoryExpression</title>
+			<para>
+				The <code>mandatoryExpression</code> SpEL <interfacename>Expression</interfacename> property
+				has been added to the <classname>RabbitTemplate</classname> to evaluate a <code>mandatory</code>
+				boolean value against each request message, when a <classname>ReturnCallback</classname> is in use.
+				See <xref linkend="template-confirms"/>.
+			</para>
+		</section>
 	</section>
 
 	<section>