From 96ef57d36f88904d16bf2202a4bb6972ba71eeb1 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 22 Feb 2008 00:16:34 +0000 Subject: [PATCH] Added configurable acknowledge modes for JmsMessageDrivenSourceAdapter (INT-124). --- .../spring-integration-adapters-1.0.xsd | 16 +++++++ .../jms/JmsMessageDrivenSourceAdapter.java | 40 +++++++++++------ .../jms/config/JmsSourceAdapterParser.java | 45 +++++++++++++++++++ 3 files changed, 88 insertions(+), 13 deletions(-) diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd index 0bd0dce470..9a4603f7e1 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd @@ -60,6 +60,22 @@ + + + + + + + + + + + + + diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java index 2eeeaf1337..2030bb1dd3 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java @@ -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 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 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 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); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java index 0510175c15..4741abf880 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParser.java @@ -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; + } } }