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

Refactored JmsOutboundGateway so that a 'replyDestination' may be set. If none is provided, it will create a TemporaryQueue per invocation (INT-466).

Browse Source
This commit is contained in:
Mark Fisher
2008-11-12 23:58:01 +00:00
parent 482465ff95
commit 34f8cee5b0
5 changed files with 162 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsInboundGateway.java
View File

@@ -148,6 +148,9 @@ public class JmsInboundGateway extends SimpleMessagingGateway implements Disposa
Message<?> replyMessage = JmsInboundGateway.this.sendAndReceiveMessage(object);
if (replyMessage != null) {
javax.jms.Message jmsReply = messageConverter.toMessage(replyMessage, session);
if (jmsReply.getJMSCorrelationID() == null) {
jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID());
}
MessageProducer producer = session.createProducer(jmsMessage.getJMSReplyTo());
producer.send(jmsMessage.getJMSReplyTo(), jmsReply);
}

180
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/JmsOutboundGateway.java
View File

@@ -18,12 +18,17 @@ package org.springframework.integration.jms;
import java.io.Serializable;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueRequestor;
import javax.jms.QueueSession;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.Session;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.core.Message;
@@ -31,8 +36,9 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ReplyMessageHolder;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.SessionCallback;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.support.JmsUtils;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.util.Assert;
@@ -41,56 +47,172 @@ import org.springframework.util.Assert;
* An outbound Messaging Gateway for request/reply JMS.
*
* @author Mark Fisher
* @author Arjen Poutsma
* @author Juergen Hoeller
*/
public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler implements InitializingBean {
private volatile Queue jmsQueue;
private volatile Destination requestDestination;
private volatile Destination replyDestination;
private volatile long receiveTimeout = 5000;
private volatile int deliveryMode = javax.jms.Message.DEFAULT_DELIVERY_MODE;
private volatile long timeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
private volatile int priority = javax.jms.Message.DEFAULT_PRIORITY;
private ConnectionFactory connectionFactory;
private volatile MessageConverter messageConverter = new HeaderMappingMessageConverter(new SimpleMessageConverter());
private final JmsTemplate jmsTemplate = new JmsTemplate();
public void setRequestDestination(Destination requestDestination) {
this.requestDestination = requestDestination;
}
public void setJmsQueue(Queue jmsQueue) {
this.jmsQueue = jmsQueue;
public void setReplyDestination(Destination replyDestination) {
this.replyDestination = replyDestination;
}
public void setConnectionFactory(ConnectionFactory connectionFactory) {
this.jmsTemplate.setConnectionFactory(connectionFactory);
this.connectionFactory = connectionFactory;
}
public void setReplyChannel(MessageChannel replyChannel) {
this.setOutputChannel(replyChannel);
public void setDeliveryMode(int deliveryMode) {
this.deliveryMode = deliveryMode;
}
public void setPriority(int priority) {
this.priority = priority;
}
public void setTimeToLive(long timeToLive) {
this.timeToLive = timeToLive;
}
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null");
this.messageConverter = messageConverter;
}
public void setReplyChannel(MessageChannel replyChannel) {
this.setOutputChannel(replyChannel);
}
public void afterPropertiesSet() {
this.jmsTemplate.afterPropertiesSet();
Assert.notNull(this.jmsQueue, "jmsQueue must not be null");
Assert.notNull(this.connectionFactory, "connectionFactory must not be null");
Assert.notNull(this.requestDestination, "requestDestination must not be null");
}
@Override
protected void handleRequestMessage(final Message<?> message, final ReplyMessageHolder replyMessageHolder) {
final Message<?> requestMessage = MessageBuilder.fromMessage(message).build();
this.jmsTemplate.execute(new SessionCallback() {
public Object doInJms(Session session) throws JMSException {
Assert.state(session instanceof QueueSession,
"QueueSession is required for the outbound JMS Gateway");
javax.jms.Message jmsRequest = (messageConverter != null)
? messageConverter.toMessage(requestMessage, session)
: session.createObjectMessage((Serializable) requestMessage);
QueueRequestor requestor = new QueueRequestor((QueueSession) session, jmsQueue);
javax.jms.Message jmsReply = requestor.request(jmsRequest);
Object result = (messageConverter != null)
? messageConverter.fromMessage(jmsReply) : jmsReply;
replyMessageHolder.set(result);
return null;
try {
javax.jms.Message jmsReply = JmsOutboundGateway.this.sendAndReceive(requestMessage);
Object result = (messageConverter != null) ? messageConverter.fromMessage(jmsReply) : jmsReply;
replyMessageHolder.set(result);
}
catch (JMSException e) {
throw new MessageHandlingException(requestMessage, e);
}
}
private javax.jms.Message sendAndReceive(Message<?> requestMessage) throws JMSException {
Connection connection = createConnection();
Session session = null;
MessageProducer messageProducer = null;
MessageConsumer messageConsumer = null;
Destination replyTo = null;
try {
session = createSession(connection);
javax.jms.Message jmsRequest = (messageConverter != null)
? messageConverter.toMessage(requestMessage, session)
: session.createObjectMessage((Serializable) requestMessage);
messageProducer = session.createProducer(this.requestDestination);
messageProducer.setDeliveryMode(this.deliveryMode);
messageProducer.setPriority(this.priority);
messageProducer.setTimeToLive(this.timeToLive);
replyTo = (this.replyDestination != null)
? this.replyDestination : session.createTemporaryQueue();
jmsRequest.setJMSReplyTo(replyTo);
connection.start();
messageProducer.send(jmsRequest);
if (replyTo instanceof TemporaryQueue || replyTo instanceof TemporaryTopic) {
messageConsumer = session.createConsumer(replyTo);
}
}, true);
else {
String messageId = jmsRequest.getJMSMessageID().replaceAll("'", "''");
String messageSelector = "JMSCorrelationID = '" + messageId + "'";
messageConsumer = session.createConsumer(replyTo, messageSelector);
}
return (this.receiveTimeout >= 0) ? messageConsumer.receive(receiveTimeout) : messageConsumer.receive();
}
finally {
JmsUtils.closeMessageProducer(messageProducer);
JmsUtils.closeMessageConsumer(messageConsumer);
JmsUtils.closeSession(session);
this.deleteDestinationIfTemporary(replyTo);
ConnectionFactoryUtils.releaseConnection(connection, this.connectionFactory, true);
}
}
/**
* Deletes either a {@link TemporaryQueue} or {@link TemporaryTopic}.
* Ignores any other {@link Destination} type and also ignores any
* {@link JMSException}s that may be thrown when attempting to delete.
*/
private void deleteDestinationIfTemporary(Destination destination) {
try {
if (destination instanceof TemporaryQueue) {
((TemporaryQueue) destination).delete();
}
else if (destination instanceof TemporaryTopic) {
((TemporaryTopic) destination).delete();
}
}
catch (JMSException e) {
// ignore
}
}
/**
* Create a new JMS Connection for this JMS gateway, ideally a
* <code>javax.jms.QueueConnection</code>.
* <p>The default implementation uses the
* <code>javax.jms.QueueConnectionFactory</code> API if available,
* falling back to a standard JMS 1.1 ConnectionFactory otherwise.
* This is necessary for working with generic JMS 1.1 connection pools
* (such as ActiveMQ's <code>org.apache.activemq.pool.PooledConnectionFactory</code>).
*/
protected Connection createConnection() throws JMSException {
ConnectionFactory cf = this.connectionFactory;
if (cf instanceof QueueConnectionFactory) {
return ((QueueConnectionFactory) cf).createQueueConnection();
}
else {
return cf.createConnection();
}
}
/**
* Create a new JMS Session for this JMS gateway, ideally a
* <code>javax.jms.QueueSession</code>.
* <p>The default implementation uses the
* <code>javax.jms.QueueConnection</code> API if available,
* falling back to a standard JMS 1.1 Connection otherwise.
* This is necessary for working with generic JMS 1.1 connection pools
* (such as ActiveMQ's <code>org.apache.activemq.pool.PooledConnectionFactory</code>).
*/
protected Session createSession(Connection connection) throws JMSException {
if (connection instanceof QueueConnection) {
return ((QueueConnection) connection).createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
}
else {
return connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
}
}
}

4
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/JmsOutboundGatewayParser.java
View File

@@ -23,7 +23,6 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;outbound-gateway&gt; element of the integration 'jms' namespace.
@@ -41,7 +40,8 @@ public class JmsOutboundGatewayParser extends AbstractConsumerEndpointParser {
protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsOutboundGateway.class);
builder.addPropertyReference("connectionFactory", element.getAttribute("connection-factory"));
builder.addPropertyReference("jmsQueue", element.getAttribute("jms-queue"));
builder.addPropertyReference("requestDestination", element.getAttribute("request-destination"));
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-destination");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
return builder;

3
org.springframework.integration.jms/src/main/java/org/springframework/integration/jms/config/spring-integration-jms-1.0.xsd
View File

@@ -74,7 +74,8 @@
<xsd:attribute name="reply-channel" type="xsd:string"/>
<xsd:attribute name="request-timeout" type="xsd:string"/>
<xsd:attribute name="reply-timeout" type="xsd:string"/>
<xsd:attribute name="jms-queue" type="xsd:string" use="required"/>
<xsd:attribute name="request-destination" type="xsd:string" use="required"/>
<xsd:attribute name="reply-destination" type="xsd:string"/>
<xsd:attribute name="connection-factory" type="xsd:string" default="connectionFactory"/>
<xsd:attribute name="message-converter" type="xsd:string"/>
</xsd:complexType>

8
org.springframework.integration.jms/src/test/java/org/springframework/integration/jms/config/jmsOutboundGatewayWithConverter.xml
View File

@@ -15,9 +15,9 @@
</si:channel>
<jms:outbound-gateway id="jmsGateway"
jms-queue="testQueue"
request-channel="requestChannel"
message-converter="converter"/>
request-destination="requestQueue"
request-channel="requestChannel"
message-converter="converter"/>
@@ -29,7 +29,7 @@
</constructor-arg>
</bean>
<bean id="testQueue" class="org.springframework.integration.jms.StubQueue"/>
<bean id="requestQueue" class="org.springframework.integration.jms.StubQueue"/>
<bean id="converter" class="org.springframework.integration.jms.StubMessageConverter" />