request destination exprs for outbound-gateway

spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java

@@ -33,11 +33,14 @@ import javax.jms.TemporaryQueue;
 import javax.jms.TemporaryTopic;
 import javax.jms.Topic;
 
+import org.springframework.expression.Expression;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
+import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessageTimeoutException;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.jms.connection.ConnectionFactoryUtils;
 import org.springframework.jms.support.JmsUtils;
@@ -61,6 +64,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 
 	private volatile String requestDestinationName;
 
+	private volatile ExpressionEvaluatingMessageProcessor<?> requestDestinationExpressionProcessor;
+
 	private volatile Destination replyDestination;
 
 	private volatile String replyDestinationName;
@@ -121,7 +126,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 
 	/**
 	 * Set the JMS Destination to which request Messages should be sent.
-	 * Either this or the 'requestDestinationName' property is required.
+	 * Either this or one of 'requestDestinationName' or 'requestDestinationExpression' is required.
 	 */
 	public void setRequestDestination(Destination requestDestination) {
 		if (requestDestination instanceof Topic) {
@@ -131,13 +136,22 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 	}
 
 	/**
-	 * Set the name of the JMS Destination to which request Messages should be
-	 * sent. Either this or the 'requestDestination' property is required.
+	 * Set the name of the JMS Destination to which request Messages should be sent.
+	 * Either this or one of 'requestDestination' or 'requestDestinationExpression' is required.
 	 */
 	public void setRequestDestinationName(String requestDestinationName) {
 		this.requestDestinationName = requestDestinationName;
 	}
 
+	/**
+	 * Set the SpEL Expression to be used for determining the request Destination instance
+	 * or request destination name. Either this or one of 'requestDestination' or
+	 * 'requestDestinationName' is required.
+	 */
+	public void setRequestDestinationExpression(Expression requestDestinationExpression) {
+		this.requestDestinationExpressionProcessor = new ExpressionEvaluatingMessageProcessor<Object>(requestDestinationExpression);
+	}
+
 	/**
 	 * Set the JMS Destination from which reply Messages should be received.
 	 * If none is provided, this gateway will create a {@link TemporaryQueue} per invocation.
@@ -304,14 +318,33 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 		return "jms:outbound-gateway";
 	}
 
-	private Destination getRequestDestination(Session session) throws JMSException {
+	private Destination getRequestDestination(Message<?> message, Session session) throws JMSException {
 		if (this.requestDestination != null) {
 			return this.requestDestination;
 		}
+		if (this.requestDestinationName != null) {
+			return this.resolveRequestDestination(this.requestDestinationName, session);
+		}
+		if (this.requestDestinationExpressionProcessor != null) {
+			Object result = this.requestDestinationExpressionProcessor.processMessage(message);
+			if (result instanceof Destination) {
+				return (Destination) result;
+			}
+			if (result instanceof String) {
+				return this.resolveRequestDestination((String) result, session);
+			}
+			throw new MessageDeliveryException(message,
+					"Evaluation of requestDestinationExpression failed to produce a Destination or destination name. Result was: " + result);
+		}
+		throw new MessageDeliveryException(message,
+				"No requestDestination, requestDestinationName, or requestDestinationExpression has been configured.");
+	}
+
+	private Destination resolveRequestDestination(String requestDestinationName, Session session) throws JMSException {
 		Assert.notNull(this.destinationResolver,
 				"DestinationResolver is required when relying upon the 'requestDestinationName' property.");
 		return this.destinationResolver.resolveDestinationName(
-				session, this.requestDestinationName, this.requestPubSubDomain);
+				session, requestDestinationName, this.requestPubSubDomain);
 	}
 
 	private Destination getReplyDestination(Session session) throws JMSException {
@@ -334,8 +367,14 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 				return;
 			}
 			Assert.notNull(this.connectionFactory, "connectionFactory must not be null");
-			Assert.isTrue(this.requestDestination != null || this.requestDestinationName != null,
-					"Either a 'requestDestination' or 'requestDestinationName' is required.");
+			Assert.isTrue(this.requestDestination != null
+					^ this.requestDestinationName != null
+					^ this.requestDestinationExpressionProcessor != null,
+					"Exactly one of 'requestDestination', 'requestDestinationName', or 'requestDestinationExpression' is required.");
+			if (this.requestDestinationExpressionProcessor != null) {
+				this.requestDestinationExpressionProcessor.setBeanFactory(getBeanFactory());
+				this.requestDestinationExpressionProcessor.setConversionService(getConversionService());
+			}
 			this.initialized = true;
 		}
 	}
@@ -402,14 +441,15 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 				priority = this.priority;
 			}
 			javax.jms.Message replyMessage = null;
+			Destination requestDestination = this.getRequestDestination(requestMessage, session);
 			if (this.correlationKey != null) {
-				replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(jmsRequest, replyTo, session, priority);
+				replyMessage = this.doSendAndReceiveWithGeneratedCorrelationId(requestDestination, jmsRequest, replyTo, session, priority);
 			}
 			else if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) {
-				replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(jmsRequest, replyTo, session, priority);
+				replyMessage = this.doSendAndReceiveWithTemporaryReplyToDestination(requestDestination, jmsRequest, replyTo, session, priority);
 			}
 			else {
-				replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(jmsRequest, replyTo, session, priority);
+				replyMessage = this.doSendAndReceiveWithMessageIdCorrelation(requestDestination, jmsRequest, replyTo, session, priority);
 			}
 			return replyMessage;
 		}
@@ -423,11 +463,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 	/**
 	 * Creates the MessageConsumer before sending the request Message since we are generating our own correlationId value for the MessageSelector.
 	 */
-	private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
+	private javax.jms.Message doSendAndReceiveWithGeneratedCorrelationId(Destination requestDestination,
+			javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
 		MessageProducer messageProducer = null;
 		MessageConsumer messageConsumer = null;
 		try {
-			messageProducer = session.createProducer(this.getRequestDestination(session));
+			messageProducer = session.createProducer(requestDestination);
 			String correlationId = UUID.randomUUID().toString().replaceAll("'", "''");
 			Assert.state(this.correlationKey != null, "correlationKey must not be null");
 			String messageSelector = null;
@@ -452,11 +493,12 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 	/**
 	 * Creates the MessageConsumer before sending the request Message since we do not need any correlation.
 	 */
-	private javax.jms.Message doSendAndReceiveWithTemporaryReplyToDestination(javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
+	private javax.jms.Message doSendAndReceiveWithTemporaryReplyToDestination(Destination requestDestination,
+			javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
 		MessageProducer messageProducer = null;
 		MessageConsumer messageConsumer = null;
 		try {
-			messageProducer = session.createProducer(this.getRequestDestination(session));
+			messageProducer = session.createProducer(requestDestination);
 			messageConsumer = session.createConsumer(replyTo);
 			this.sendRequestMessage(jmsRequest, messageProducer, priority);
 			return this.receiveReplyMessage(messageConsumer);
@@ -470,7 +512,8 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 	/**
 	 * Creates the MessageConsumer after sending the request Message since we need the MessageID for correlation with a MessageSelector.
 	 */
-	private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
+	private javax.jms.Message doSendAndReceiveWithMessageIdCorrelation(Destination requestDestination,
+			javax.jms.Message jmsRequest, Destination replyTo, Session session, int priority) throws JMSException {
 		if (replyTo instanceof Topic && logger.isWarnEnabled()) {
 			logger.warn("Relying on the MessageID for correlation is not recommended when using a Topic as the replyTo Destination " +
 					"because that ID can only be provided to a MessageSelector after the reuqest Message has been sent thereby " +
@@ -481,7 +524,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler {
 		MessageProducer messageProducer = null;
 		MessageConsumer messageConsumer = null;
 		try {
-			messageProducer = session.createProducer(this.getRequestDestination(session));
+			messageProducer = session.createProducer(requestDestination);
 			this.sendRequestMessage(jmsRequest, messageProducer, priority);
 			String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''");
 			String messageSelector = "JMSCorrelationID = '" + messageId + "'";

spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java

@@ -20,6 +20,7 @@ import org.w3c.dom.Element;
 
 import org.springframework.beans.factory.support.BeanDefinitionBuilder;
 import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.config.ExpressionFactoryBean;
 import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
 import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
 import org.springframework.util.StringUtils;
@@ -44,16 +45,25 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
 		builder.addPropertyReference("connectionFactory", element.getAttribute("connection-factory"));
 		String requestDestination = element.getAttribute("request-destination");
 		String requestDestinationName = element.getAttribute("request-destination-name");
-		if (!(StringUtils.hasText(requestDestination) ^ StringUtils.hasText(requestDestinationName))) {
-			parserContext.getReaderContext().error(
-					"Exactly one of the 'request-destination' or 'request-destination-name' attributes is required.", element);
+		String requestDestinationExpression = element.getAttribute("request-destination-expression");
+		boolean hasRequestDestination = StringUtils.hasText(requestDestination);
+		boolean hasRequestDestinationName = StringUtils.hasText(requestDestinationName);
+		boolean hasRequestDestinationExpression = StringUtils.hasText(requestDestinationExpression);
+		if (!(hasRequestDestination ^ hasRequestDestinationName ^ hasRequestDestinationExpression)) {
+			parserContext.getReaderContext().error("Exactly one of the 'request-destination', " +
+					"'request-destination-name', or 'request-destination-expression' attributes is required.", element);
 		}
-		if (StringUtils.hasText(requestDestination)) {
+		if (hasRequestDestination) {
 			builder.addPropertyReference("requestDestination", requestDestination);
 		}
-		else if (StringUtils.hasText(requestDestinationName)) {
+		else if (hasRequestDestinationName) {
 			builder.addPropertyValue("requestDestinationName", requestDestinationName);
 		}
+		else if (hasRequestDestinationExpression) {
+			BeanDefinitionBuilder expressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionFactoryBean.class);
+			expressionBuilder.addConstructorArgValue(requestDestinationExpression);
+			builder.addPropertyValue("requestDestinationExpression", expressionBuilder.getBeanDefinition());
+		}
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-destination");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-destination-name");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");

spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.1.xsd

@@ -732,7 +732,9 @@
 					<xsd:documentation>
 	A reference to a javax.jms.Destination by bean name. As an alternative to a bean
 	reference, use 'request-destination-name' and 'request-pub-sub-domain' which will rely
-	upon the DestinationResolver strategy (DynamicDestinationResolver by default). 
+	upon the DestinationResolver strategy (DynamicDestinationResolver by default). This
+	attribute is mutually exclusive with 'request-destination-name' and
+	'request-destination-expression'.
 					</xsd:documentation>
 					<xsd:appinfo>
 						<tool:annotation kind="ref">
@@ -741,8 +743,34 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
-			<xsd:attribute name="request-destination-name" type="xsd:string"/>
-			<xsd:attribute name="request-pub-sub-domain" type="xsd:string"/>
+			<xsd:attribute name="request-destination-name" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+	Name of a destination to which request messages will be sent. This name will be handled
+	by this gateway's DestinationResolver. This attribute is mutually exclusive with
+	'request-destination' and 'request-destination-expression'.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
+			<xsd:attribute name="request-destination-expression" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+	A SpEL expression to be evaluated at runtime against each Spring Integration request Message as
+	the root object. The result should be either a Destination instance or a String representing
+	the destination name. In the latter case, it will be passed to this adapter's DestinationResolver.
+	This attribute is mutually exclusive with 'request-destination' and 'request-destination-name'.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>			
+			<xsd:attribute name="request-pub-sub-domain" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+	When resolving a request destination name (rather than having a 'request-destination' reference),
+	a true value here specifies that the DestinationResolver should resolve Topics rather than Queues.
+	Default is false.  
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="reply-destination" type="xsd:string">
 				<xsd:annotation>
 							<xsd:documentation>

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests-context.xml

@@ -15,6 +15,10 @@
 
 	<int-jms:message-driven-channel-adapter channel="channelAdapterResults2" destination-name="queue.test.dynamic.adapter.2" extract-payload="false"/>
 
+	<int:channel id="gatewayChannel"/>
+
+	<int-jms:outbound-gateway request-channel="gatewayChannel" request-destination-expression="'queue.test.dynamic.gateway.' + headers.destinationNumber"/>
+
 	<int:channel id="channelAdapterResults1">
 		<int:queue capacity="1"/>
 	</int:channel>
@@ -33,4 +37,11 @@
 		<property name="cacheProducers" value="false"/>
 	</bean>
 
+	<jms:listener-container>
+		<jms:listener destination="queue.test.dynamic.gateway.1" ref="responder" method="one"/>
+		<jms:listener destination="queue.test.dynamic.gateway.2" ref="responder" method="two"/>
+	</jms:listener-container>
+
+	<bean id="responder" class="org.springframework.integration.jms.config.JmsDynamicDestinationTests$Responder"/>
+
 </beans>

spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsDynamicDestinationTests.java

@@ -28,6 +28,7 @@ import org.junit.runner.RunWith;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageChannel;
+import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.core.PollableChannel;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.test.context.ContextConfiguration;
@@ -44,6 +45,9 @@ public class JmsDynamicDestinationTests {
 	@Autowired
 	private MessageChannel channelAdapterChannel;
 
+	@Autowired
+	private MessageChannel gatewayChannel;
+
 	@Autowired
 	private PollableChannel channelAdapterResults1;
 
@@ -74,4 +78,29 @@ public class JmsDynamicDestinationTests {
 		assertEquals("queue://queue.test.dynamic.adapter.2", jmsResult2.getJMSDestination().toString());		
 	}
 
+	@Test
+	public void gateway() throws Exception {
+		Message<?> message1 = MessageBuilder.withPayload("test-1").setHeader("destinationNumber", 1).build();
+		Message<?> message2 = MessageBuilder.withPayload("test-2").setHeader("destinationNumber", 2).build();
+		MessagingTemplate template = new MessagingTemplate();
+		Message<?> result1 = template.sendAndReceive(gatewayChannel, message1);
+		Message<?> result2 = template.sendAndReceive(gatewayChannel, message2);
+		assertNotNull(result1);
+		assertNotNull(result2);
+		assertEquals("test-1!", result1.getPayload());
+		assertEquals("test-2!!", result2.getPayload());
+	}
+
+
+	public static class Responder {
+
+		public String one(String message) {
+			return message + "!";
+		}
+
+		public String two(String message) {
+			return message + "!!";
+		}
+	}
+
 }