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

Added configurable acknowledge modes for JmsMessageDrivenSourceAdapter (INT-124).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-22 00:16:34 +00:00
parent aafa8391d4
commit 96ef57d36f
3 changed files with 88 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
View File

@@ -60,6 +60,22 @@
<xsd:attribute name="channel" type="xsd:string" use="required"/>
<xsd:attribute name="poll-period" type="xsd:int"/>
<xsd:attribute name="message-converter" type="xsd:string"/>
<xsd:attribute name="acknowledge" default="auto">
<xsd:annotation>
<xsd:documentation><![CDATA[
The native JMS acknowledge mode: "auto", "client", "dups-ok" or "transacted".
The latter effectively activates a locally transacted Session.
]]></xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:NMTOKEN">
<xsd:enumeration value="auto"/>
<xsd:enumeration value="client"/>
<xsd:enumeration value="dups-ok"/>
<xsd:enumeration value="transacted"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

40
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.adapter.jms;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.Session;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.Lifecycle;
@@ -37,29 +38,33 @@ import org.springframework.util.Assert;
*/
public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object> implements Lifecycle, DisposableBean {
private AbstractJmsListeningContainer container;
private volatile AbstractJmsListeningContainer container;
private ConnectionFactory connectionFactory;
private volatile ConnectionFactory connectionFactory;
private Destination destination;
private volatile Destination destination;
private String destinationName;
private volatile String destinationName;
private MessageConverter messageConverter = new SimpleMessageConverter();
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private TaskExecutor taskExecutor;
private volatile TaskExecutor taskExecutor;
private long receiveTimeout = 1000;
private volatile boolean sessionTransacted;
private int concurrentConsumers = 1;
private volatile int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
private int maxConcurrentConsumers = 1;
private volatile long receiveTimeout = 1000;
private int maxMessagesPerTask = Integer.MIN_VALUE;
private volatile int concurrentConsumers = 1;
private int idleTaskExecutionLimit = 1;
private volatile int maxConcurrentConsumers = 1;
private long sendTimeout = -1;
private volatile int maxMessagesPerTask = Integer.MIN_VALUE;
private volatile int idleTaskExecutionLimit = 1;
private volatile long sendTimeout = -1;
public void setContainer(AbstractJmsListeningContainer container) {
@@ -91,6 +96,14 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
this.taskExecutor = taskExecutor;
}
public void setSessionTransacted(boolean sessionTransacted) {
this.sessionTransacted = sessionTransacted;
}
public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) {
this.sessionAcknowledgeMode = sessionAcknowledgeMode;
}
@Override
public void initialize() {
if (this.container == null) {
@@ -116,7 +129,8 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask);
dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit);
dmlc.setSessionTransacted(true);
dmlc.setSessionTransacted(this.sessionTransacted);
dmlc.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
dmlc.setAutoStartup(false);
ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel());
listener.setMessageConverter(this.messageConverter);

45
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.adapter.jms.config;
import javax.jms.Session;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanCreationException;
@@ -58,6 +60,16 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
private static final String MESSAGE_CONVERTER_PROPERTY = "messageConverter";
private static final String ACKNOWLEDGE_ATTRIBUTE = "acknowledge";
private static final String ACKNOWLEDGE_AUTO = "auto";
private static final String ACKNOWLEDGE_CLIENT = "client";
private static final String ACKNOWLEDGE_DUPS_OK = "dups-ok";
private static final String ACKNOWLEDGE_TRANSACTED = "transacted";
protected Class<?> getBeanClass(Element element) {
if (StringUtils.hasText(element.getAttribute(POLL_PERIOD_ATTRIBUTE))) {
@@ -153,6 +165,39 @@ public class JmsSourceAdapterParser extends AbstractSingleBeanDefinitionParser {
if (StringUtils.hasText(messageConverter)) {
builder.addPropertyReference(MESSAGE_CONVERTER_PROPERTY, messageConverter);
}
Integer acknowledgeMode = parseAcknowledgeMode(element);
if (acknowledgeMode != null) {
if (acknowledgeMode.intValue() == Session.SESSION_TRANSACTED) {
builder.addPropertyValue("sessionTransacted", Boolean.TRUE);
}
else {
builder.addPropertyValue("sessionAcknowledgeMode", acknowledgeMode);
}
}
}
private Integer parseAcknowledgeMode(Element element) {
String acknowledge = element.getAttribute(ACKNOWLEDGE_ATTRIBUTE);
if (StringUtils.hasText(acknowledge)) {
int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
if (ACKNOWLEDGE_TRANSACTED.equals(acknowledge)) {
acknowledgeMode = Session.SESSION_TRANSACTED;
}
else if (ACKNOWLEDGE_DUPS_OK.equals(acknowledge)) {
acknowledgeMode = Session.DUPS_OK_ACKNOWLEDGE;
}
else if (ACKNOWLEDGE_CLIENT.equals(acknowledge)) {
acknowledgeMode = Session.CLIENT_ACKNOWLEDGE;
}
else if (!ACKNOWLEDGE_AUTO.equals(acknowledge)) {
throw new BeanCreationException("Invalid jms-source 'acknowledge' setting: " +
"only \"auto\", \"client\", \"dups-ok\" and \"transacted\" supported.");
}
return acknowledgeMode;
}
else {
return null;
}
}
}