Refactored the JmsInboundGateway into the simpler JmsMessageDrivenEndpoint. The JmsInboundGatewayParser now handles a 'container' reference, else it creates a DefaultMessageListenerContainer (INT-482). The JmsMessageDrivenEndpoint will also provide the necessary implementation to support an event-driven inbound-channel-adapter for JMS (INT-477).
This commit is contained in:
Mark Fisher
@@ -28,7 +28,7 @@ import org.springframework.jms.support.converter.MessageConverter;
|
||||
/**
|
||||
* A source for receiving JMS Messages with a polling listener. This source is
|
||||
* only recommended for very low message volume. Otherwise, the
|
||||
* {@link JmsInboundGateway} that uses Spring's MessageListener container
|
||||
* {@link JmsMessageDrivenEndpoint} that uses Spring's MessageListener container
|
||||
* support is a better option.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
|
||||
@@ -1,203 +0,0 @@
|
||||
/*
|
||||
* Copyright 2002-2008 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.jms;
|
||||
|
||||
import javax.jms.ConnectionFactory;
|
||||
import javax.jms.Destination;
|
||||
import javax.jms.MessageListener;
|
||||
import javax.jms.Session;
|
||||
import javax.jms.Topic;
|
||||
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.integration.endpoint.AbstractEndpoint;
|
||||
import org.springframework.jms.listener.AbstractMessageListenerContainer;
|
||||
import org.springframework.jms.listener.DefaultMessageListenerContainer;
|
||||
import org.springframework.jms.listener.SessionAwareMessageListener;
|
||||
import org.springframework.transaction.PlatformTransactionManager;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* A message-driven adapter for receiving JMS messages and sending to a channel.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class JmsInboundGateway extends AbstractEndpoint implements DisposableBean {
|
||||
|
||||
private final Object listener;
|
||||
|
||||
private volatile AbstractMessageListenerContainer container;
|
||||
|
||||
private volatile ConnectionFactory connectionFactory;
|
||||
|
||||
private volatile Destination destination;
|
||||
|
||||
private volatile String destinationName;
|
||||
|
||||
private volatile boolean pubSubDomain;
|
||||
|
||||
private volatile TaskExecutor taskExecutor;
|
||||
|
||||
private volatile PlatformTransactionManager transactionManager;
|
||||
|
||||
private volatile boolean sessionTransacted;
|
||||
|
||||
private volatile int sessionAcknowledgeMode = Session.AUTO_ACKNOWLEDGE;
|
||||
|
||||
private volatile int concurrentConsumers = 1;
|
||||
|
||||
private volatile int maxConcurrentConsumers = 1;
|
||||
|
||||
private volatile int maxMessagesPerTask = Integer.MIN_VALUE;
|
||||
|
||||
private volatile int idleTaskExecutionLimit = 1;
|
||||
|
||||
|
||||
public JmsInboundGateway(Object listener) {
|
||||
Assert.notNull(listener, "listener must not be null");
|
||||
Assert.isTrue(listener instanceof MessageListener || listener instanceof SessionAwareMessageListener,
|
||||
"listener must implement either [" + MessageListener.class.getName()
|
||||
+ "] or [" + SessionAwareMessageListener.class.getName() + "]");
|
||||
this.listener = listener;
|
||||
}
|
||||
|
||||
|
||||
public void setContainer(AbstractMessageListenerContainer container) {
|
||||
this.container = container;
|
||||
}
|
||||
|
||||
public void setConnectionFactory(ConnectionFactory connectionFactory) {
|
||||
this.connectionFactory = connectionFactory;
|
||||
}
|
||||
|
||||
public void setDestination(Destination destination) {
|
||||
if (destination instanceof Topic) {
|
||||
this.pubSubDomain = true;
|
||||
}
|
||||
this.destination = destination;
|
||||
}
|
||||
|
||||
public void setDestinationName(String destinationName) {
|
||||
this.destinationName = destinationName;
|
||||
}
|
||||
|
||||
/**
|
||||
* Specify whether the request destination is a Topic. This value is
|
||||
* necessary when providing a destination name for a Topic rather than
|
||||
* a destination reference.
|
||||
*
|
||||
* @param pubSubDomain true if the request destination is a Topic
|
||||
*/
|
||||
public void setPubSubDomain(boolean pubSubDomain) {
|
||||
this.pubSubDomain = pubSubDomain;
|
||||
}
|
||||
|
||||
public void setTaskExecutor(TaskExecutor taskExecutor) {
|
||||
this.taskExecutor = taskExecutor;
|
||||
}
|
||||
|
||||
public void setTransactionManager(PlatformTransactionManager transactionManager) {
|
||||
this.transactionManager = transactionManager;
|
||||
}
|
||||
|
||||
public void setSessionTransacted(boolean sessionTransacted) {
|
||||
this.sessionTransacted = sessionTransacted;
|
||||
}
|
||||
|
||||
public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) {
|
||||
this.sessionAcknowledgeMode = sessionAcknowledgeMode;
|
||||
}
|
||||
|
||||
public void setConcurrentConsumers(int concurrentConsumers) {
|
||||
this.concurrentConsumers = concurrentConsumers;
|
||||
}
|
||||
|
||||
public void setMaxConcurrentConsumers(int maxConcurrentConsumers) {
|
||||
this.maxConcurrentConsumers = maxConcurrentConsumers;
|
||||
}
|
||||
|
||||
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
|
||||
this.maxMessagesPerTask = maxMessagesPerTask;
|
||||
}
|
||||
|
||||
public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) {
|
||||
this.idleTaskExecutionLimit = idleTaskExecutionLimit;
|
||||
}
|
||||
|
||||
|
||||
private void initialize() {
|
||||
if (this.container == null) {
|
||||
this.container = createDefaultContainer();
|
||||
}
|
||||
this.container.setMessageListener(this.listener);
|
||||
if (!this.container.isActive()) {
|
||||
this.container.afterPropertiesSet();
|
||||
}
|
||||
}
|
||||
|
||||
private AbstractMessageListenerContainer createDefaultContainer() {
|
||||
Assert.isTrue(this.connectionFactory != null
|
||||
&& (this.destination != null || this.destinationName != null),
|
||||
"If a 'container' reference is not provided, then 'connectionFactory'"
|
||||
+ " and 'destination' (or 'destinationName') are required.");
|
||||
DefaultMessageListenerContainer dmlc = new DefaultMessageListenerContainer();
|
||||
dmlc.setConnectionFactory(this.connectionFactory);
|
||||
if (this.destination != null) {
|
||||
dmlc.setDestination(this.destination);
|
||||
}
|
||||
if (this.destinationName != null) {
|
||||
dmlc.setDestinationName(this.destinationName);
|
||||
}
|
||||
dmlc.setPubSubDomain(this.pubSubDomain);
|
||||
dmlc.setConcurrentConsumers(this.concurrentConsumers);
|
||||
dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
|
||||
dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask);
|
||||
dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit);
|
||||
dmlc.setTransactionManager(this.transactionManager);
|
||||
dmlc.setSessionTransacted(this.sessionTransacted);
|
||||
dmlc.setSessionAcknowledgeMode(this.sessionAcknowledgeMode);
|
||||
dmlc.setAutoStartup(false);
|
||||
if (this.taskExecutor != null) {
|
||||
dmlc.setTaskExecutor(this.taskExecutor);
|
||||
}
|
||||
return dmlc;
|
||||
}
|
||||
|
||||
// Lifecycle implementation
|
||||
|
||||
@Override
|
||||
protected void doStart() {
|
||||
this.initialize();
|
||||
this.container.start();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() {
|
||||
if (this.container != null) {
|
||||
this.container.stop();
|
||||
}
|
||||
}
|
||||
|
||||
// DisposableBean implementation
|
||||
|
||||
public void destroy() {
|
||||
if (this.container != null) {
|
||||
this.container.destroy();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,73 @@
|
||||
/*
|
||||
* Copyright 2002-2008 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.jms;
|
||||
|
||||
import org.springframework.beans.factory.DisposableBean;
|
||||
import org.springframework.integration.endpoint.AbstractEndpoint;
|
||||
import org.springframework.jms.listener.AbstractMessageListenerContainer;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
/**
|
||||
* A message-driven endpoint that receive JMS messages, converts them into
|
||||
* Spring Integration Messages, and then sends the result to a channel.
|
||||
*
|
||||
* @author Mark Fisher
|
||||
*/
|
||||
public class JmsMessageDrivenEndpoint extends AbstractEndpoint implements DisposableBean {
|
||||
|
||||
private final AbstractMessageListenerContainer listenerContainer;
|
||||
|
||||
private final ChannelPublishingJmsMessageListener listener;
|
||||
|
||||
|
||||
public JmsMessageDrivenEndpoint(AbstractMessageListenerContainer listenerContainer, ChannelPublishingJmsMessageListener listener) {
|
||||
Assert.notNull(listenerContainer, "listener container must not be null");
|
||||
Assert.notNull(listener, "listener must not be null");
|
||||
listenerContainer.setMessageListener(listener);
|
||||
this.listener = listener;
|
||||
this.listenerContainer = listenerContainer;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
protected void onInit() throws Exception {
|
||||
this.listener.afterPropertiesSet();
|
||||
if (!this.listenerContainer.isActive()) {
|
||||
this.listenerContainer.afterPropertiesSet();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStart() {
|
||||
if (!this.listenerContainer.isRunning()) {
|
||||
this.listenerContainer.start();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void doStop() {
|
||||
this.listenerContainer.stop();
|
||||
}
|
||||
|
||||
public void destroy() throws Exception {
|
||||
if (this.isRunning()) {
|
||||
this.stop();
|
||||
}
|
||||
this.listenerContainer.destroy();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -20,14 +20,15 @@ import javax.jms.Session;
|
||||
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
import org.springframework.beans.factory.BeanCreationException;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
|
||||
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
|
||||
import org.springframework.integration.jms.ChannelPublishingJmsMessageListener;
|
||||
import org.springframework.integration.jms.JmsInboundGateway;
|
||||
import org.springframework.integration.jms.JmsMessageDrivenEndpoint;
|
||||
import org.springframework.jms.listener.DefaultMessageListenerContainer;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
@@ -37,9 +38,19 @@ import org.springframework.util.StringUtils;
|
||||
*/
|
||||
public class JmsInboundGatewayParser extends AbstractSingleBeanDefinitionParser {
|
||||
|
||||
private static String[] containerAttributes = new String[] {
|
||||
JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
|
||||
JmsAdapterParserUtils.DESTINATION_ATTRIBUTE,
|
||||
JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE,
|
||||
"transaction-manager", "pub-sub-domain",
|
||||
"concurrent-consumers", "max-concurrent-consumers",
|
||||
"max-messages-per-task", "idle-task-execution-limit"
|
||||
};
|
||||
|
||||
|
||||
@Override
|
||||
protected Class<?> getBeanClass(Element element) {
|
||||
return JmsInboundGateway.class;
|
||||
return JmsMessageDrivenEndpoint.class;
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -54,23 +65,35 @@ public class JmsInboundGatewayParser extends AbstractSingleBeanDefinitionParser
|
||||
|
||||
@Override
|
||||
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
|
||||
String containerBeanName = this.parseMessageListenerContainer(element, parserContext);
|
||||
String listenerBeanName = this.parseMessageListener(element, parserContext);
|
||||
builder.addConstructorArgReference(containerBeanName);
|
||||
builder.addConstructorArgReference(listenerBeanName);
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
|
||||
}
|
||||
|
||||
private String parseMessageListenerContainer(Element element, ParserContext parserContext) {
|
||||
if (element.hasAttribute("container")) {
|
||||
for (String containerAttribute : containerAttributes) {
|
||||
Assert.isTrue(!element.hasAttribute(containerAttribute), "The '" + containerAttribute +
|
||||
"' attribute should not be provided when specifying a 'container' reference.");
|
||||
}
|
||||
return element.getAttribute("container");
|
||||
}
|
||||
// otherwise, we build a DefaultMessageListenerContainer instance
|
||||
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(DefaultMessageListenerContainer.class);
|
||||
String destination = element.getAttribute(JmsAdapterParserUtils.DESTINATION_ATTRIBUTE);
|
||||
String destinationName = element.getAttribute(JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE);
|
||||
if (StringUtils.hasText(destination) || StringUtils.hasText(destinationName)) {
|
||||
builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
|
||||
JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
|
||||
if (StringUtils.hasText(destination)) {
|
||||
builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination);
|
||||
}
|
||||
else {
|
||||
builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName);
|
||||
}
|
||||
Assert.isTrue(StringUtils.hasText(destination) ^ StringUtils.hasText(destinationName),
|
||||
"Exactly one of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE +
|
||||
"' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' is required.");
|
||||
builder.addPropertyReference(JmsAdapterParserUtils.CONNECTION_FACTORY_PROPERTY,
|
||||
JmsAdapterParserUtils.determineConnectionFactoryBeanName(element));
|
||||
if (StringUtils.hasText(destination)) {
|
||||
builder.addPropertyReference(JmsAdapterParserUtils.DESTINATION_PROPERTY, destination);
|
||||
}
|
||||
else {
|
||||
throw new BeanCreationException("One of '" + JmsAdapterParserUtils.DESTINATION_ATTRIBUTE +
|
||||
"' or '" + JmsAdapterParserUtils.DESTINATION_NAME_ATTRIBUTE + "' must be provided.");
|
||||
builder.addPropertyValue(JmsAdapterParserUtils.DESTINATION_NAME_PROPERTY, destinationName);
|
||||
}
|
||||
Integer acknowledgeMode = JmsAdapterParserUtils.parseAcknowledgeMode(element);
|
||||
if (acknowledgeMode != null) {
|
||||
@@ -87,6 +110,8 @@ public class JmsInboundGatewayParser extends AbstractSingleBeanDefinitionParser
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-concurrent-consumers");
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "max-messages-per-task");
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "idle-task-execution-limit");
|
||||
builder.addPropertyValue("autoStartup", false);
|
||||
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
|
||||
}
|
||||
|
||||
private String parseMessageListener(Element element, ParserContext parserContext) {
|
||||
|
||||
Reference in New Issue
Block a user