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

INT-1672 added the 'correlationKey' property to the ChannelPublishingJmsMessageListener as well as a 'correlation-key' attribute for the JMS <inbound-gateway> element

Browse Source
This commit is contained in:
Mark Fisher
2010-12-09 16:40:26 -05:00
parent 9c344687a3
commit dddeabd717
3 changed files with 75 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java
View File

@@ -30,6 +30,7 @@ import org.springframework.integration.Message;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.JmsUtils;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.converter.SimpleMessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
@@ -59,6 +60,8 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
private volatile Object defaultReplyDestination;
private volatile String correlationKey;
private volatile long replyTimeToLive = javax.jms.Message.DEFAULT_TIME_TO_LIVE;
private volatile int replyPriority = javax.jms.Message.DEFAULT_PRIORITY;
@@ -145,6 +148,21 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
this.replyDeliveryMode = replyDeliveryPersistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
}
/**
* Provide the name of a JMS property that should be copied from the request
* Message to the reply Message. If this value is NULL (the default) then the
* JMSMessageID from the request will be copied into the JMSCorrelationID of the reply
* unless there is already a value in the JMSCorrelationID property of the newly created
* reply Message in which case nothing will be copied. If the JMSCorrelationID of the
* request Message should be copied into the JMSCorrelationID of the reply Message
* instead, then this value should be set to "JMSCorrelationID".
* Any other value will be treated as a JMS String Property to be copied as-is
* from the request Message into the reply Message with the same property name.
*/
public void setCorrelationKey(String correlationKey) {
this.correlationKey = correlationKey;
}
/**
* Specify whether explicit QoS should be enabled for replies
* (for timeToLive, priority, and deliveryMode settings).
@@ -228,34 +246,18 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
Message<?> replyMessage = this.sendAndReceiveMessage(requestMessage);
if (replyMessage != null) {
Destination destination = this.getReplyDestination(jmsMessage, session);
if (destination != null){
if (destination != null) {
// convert SI Message to JMS Message
Object replyResult = replyMessage;
if (this.extractReplyPayload){
if (this.extractReplyPayload) {
replyResult = replyMessage.getPayload();
}
try {
javax.jms.Message jmsReply = this.messageConverter.toMessage(replyResult, session);
// map SI Message Headers to JMS Message Properties/Headers
headerMapper.fromHeaders(replyMessage.getHeaders(), jmsReply);
if (jmsReply.getJMSCorrelationID() == null) {
jmsReply.setJMSCorrelationID(jmsMessage.getJMSMessageID());
}
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {
producer.send(jmsReply,
this.replyDeliveryMode, this.replyPriority, this.replyTimeToLive);
}
else {
producer.send(jmsReply);
}
}
finally {
producer.close();
}
this.copyCorrelationIdFromRequestToReply(jmsMessage, jmsReply);
this.sendReply(jmsReply, destination, session);
}
catch (RuntimeException e) {
logger.error("Failed to generate JMS Reply Message from: " + replyResult, e);
@@ -263,6 +265,29 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
}
}
}
else if (logger.isDebugEnabled()) {
logger.debug("expected a reply but none was received");
}
}
}
private void copyCorrelationIdFromRequestToReply(javax.jms.Message requestMessage, javax.jms.Message replyMessage) throws JMSException {
if (this.correlationKey != null) {
if (this.correlationKey.equals("JMSCorrelationID")) {
replyMessage.setJMSCorrelationID(requestMessage.getJMSCorrelationID());
}
else {
String value = requestMessage.getStringProperty(this.correlationKey);
if (value != null) {
replyMessage.setStringProperty(this.correlationKey, value);
}
else if (logger.isWarnEnabled()) {
logger.warn("No property value available on request Message for correlationKey '" + this.correlationKey + "'");
}
}
}
else if (replyMessage.getJMSCorrelationID() == null) {
replyMessage.setJMSCorrelationID(requestMessage.getJMSMessageID());
}
}
@@ -315,6 +340,20 @@ public class ChannelPublishingJmsMessageListener extends MessagingGatewaySupport
return null;
}
private void sendReply(javax.jms.Message replyMessage, Destination destination, Session session) throws JMSException {
MessageProducer producer = session.createProducer(destination);
try {
if (this.explicitQosEnabledForReplies) {
producer.send(replyMessage, this.replyDeliveryMode, this.replyPriority, this.replyTimeToLive);
}
else {
producer.send(replyMessage);
}
}
finally {
JmsUtils.closeMessageProducer(producer);
}
}
/**
* Internal class combining a destination name

1
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
View File

@@ -161,6 +161,7 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "reply-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-request-payload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-reply-payload");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "correlation-key");
int defaults = 0;
if (StringUtils.hasText(element.getAttribute(DEFAULT_REPLY_DESTINATION_ATTRIB))) {
defaults++;

15
spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd
View File

@@ -528,6 +528,21 @@
</xsd:attribute>
<xsd:attribute name="default-reply-queue-name" type="xsd:string"/>
<xsd:attribute name="default-reply-topic-name" type="xsd:string"/>
<xsd:attribute name="correlation-key" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Provide the name of a JMS property that should be copied from the request
Message to the reply Message. If NO value is provided for this attribute then the
JMSMessageID from the request will be copied into the JMSCorrelationID of the reply
unless there is already a value in the JMSCorrelationID property of the newly created
reply Message in which case nothing will be copied. If the JMSCorrelationID of the
request Message should be copied into the JMSCorrelationID of the reply Message
instead, then this value should be set to "JMSCorrelationID".
Any other value will be treated as a JMS String Property to be copied as-is
from the request Message into the reply Message with the same property name.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>