diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java new file mode 100644 index 0000000000..3f70198790 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/AbstractJmsChannel.java @@ -0,0 +1,49 @@ +/* + * Copyright 2002-2010 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.integration.Message; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class AbstractJmsChannel extends AbstractMessageChannel { + + private final JmsTemplate jmsTemplate; + + + public AbstractJmsChannel(JmsTemplate jmsTemplate) { + Assert.notNull(jmsTemplate, "jmsTemplate must not be null"); + this.jmsTemplate = jmsTemplate; + } + + + JmsTemplate getJmsTemplate() { + return this.jmsTemplate; + } + + @Override + protected boolean doSend(Message message, long timeout) { + this.jmsTemplate.convertAndSend(message); + return true; + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java deleted file mode 100644 index 34b91888c6..0000000000 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannel.java +++ /dev/null @@ -1,223 +0,0 @@ -/* - * Copyright 2002-2010 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 org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.beans.factory.InitializingBean; -import org.springframework.context.SmartLifecycle; -import org.springframework.integration.Message; -import org.springframework.integration.MessageChannel; -import org.springframework.integration.MessagingException; -import org.springframework.integration.core.MessageHandler; -import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.dispatcher.MessageDispatcher; -import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; -import org.springframework.integration.dispatcher.UnicastingDispatcher; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.jms.core.JmsTemplate; -import org.springframework.jms.listener.AbstractMessageListenerContainer; -import org.springframework.jms.support.converter.MessageConverter; -import org.springframework.jms.support.destination.DestinationResolver; - -/** - * A {@link MessageChannel} implementation that is actually backed by a JMS - * Destination. This class is useful as a drop-in replacement for any - * Spring Integration channel. The benefit of using this channel is that - * the full power of any JMS provider is available with only minimal - * configuration changes and without requiring any code changes. The most - * obvious benefit is the ability to delegate message persistence to the - * JMS provider. - * - * @author Mark Fisher - * @since 2.0 - */ -public class JmsDestinationBackedMessageChannel extends MessageListenerContainerConfigurationSupport - implements SubscribableChannel, MessageListener, BeanNameAware, SmartLifecycle, InitializingBean { - - private final Log logger = LogFactory.getLog(this.getClass()); - - private final JmsTemplate jmsTemplate = new JmsTemplate(); - - private volatile MessageDispatcher dispatcher; - - private volatile String name; - - - public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, Destination destination) { - this.setConnectionFactory(connectionFactory); - this.setDestination(destination); - } - - public JmsDestinationBackedMessageChannel(ConnectionFactory connectionFactory, String destinationName, boolean isPubSub) { - this.setConnectionFactory(connectionFactory); - this.setDestinationName(destinationName); - this.setPubSubDomain(isPubSub); - } - - - @Override - public void setConnectionFactory(ConnectionFactory connectionFactory) { - super.setConnectionFactory(connectionFactory); - this.jmsTemplate.setConnectionFactory(connectionFactory); - } - - @Override - public void setDestination(Destination destination) { - super.setDestination(destination); - this.jmsTemplate.setDefaultDestination(destination); - } - - @Override - public void setDestinationName(String destinationName) { - super.setDestinationName(destinationName); - this.jmsTemplate.setDefaultDestinationName(destinationName); - } - - @Override - public void setDestinationResolver(DestinationResolver destinationResolver) { - super.setDestinationResolver(destinationResolver); - this.jmsTemplate.setDestinationResolver(destinationResolver); - } - - @Override - public void setPubSubDomain(boolean pubSubDomain) { - super.setPubSubDomain(pubSubDomain); - this.jmsTemplate.setPubSubDomain(pubSubDomain); - } - - public void setDeliveryPersistent(boolean deliveryPersistent) { - this.jmsTemplate.setDeliveryPersistent(deliveryPersistent); - } - - public void setExplicitQosEnabled(boolean explicitQosEnabled) { - this.jmsTemplate.setExplicitQosEnabled(explicitQosEnabled); - } - - public void setMessageConverter(MessageConverter messageConverter) { - this.jmsTemplate.setMessageConverter(messageConverter); - } - - public void setMessageIdEnabled(boolean messageIdEnabled) { - this.jmsTemplate.setMessageIdEnabled(messageIdEnabled); - } - - public void setMessageTimestampEnabled(boolean messageTimestampEnabled) { - this.jmsTemplate.setMessageTimestampEnabled(messageTimestampEnabled); - } - - public void setPriority(int priority) { - this.jmsTemplate.setPriority(priority); - } - - @Override - public void setPubSubNoLocal(boolean pubSubNoLocal) { - super.setPubSubNoLocal(pubSubNoLocal); - this.jmsTemplate.setPubSubNoLocal(pubSubNoLocal); - } - - @Override - public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) { - super.setSessionAcknowledgeMode(sessionAcknowledgeMode); - this.jmsTemplate.setSessionAcknowledgeMode(sessionAcknowledgeMode); - } - - @Override - public void setSessionTransacted(boolean sessionTransacted) { - super.setSessionTransacted(sessionTransacted); - this.jmsTemplate.setSessionTransacted(sessionTransacted); - } - - public void setTimeToLive(long timeToLive) { - this.jmsTemplate.setTimeToLive(timeToLive); - } - - public void setBeanName(String beanName) { - this.name = beanName; - } - - @Override - public void afterPropertiesSet() throws Exception { - super.afterPropertiesSet(); - AbstractMessageListenerContainer container = this.getListenerContainer(); - this.configureDispatcher(container.isPubSubDomain()); - container.setMessageListener(this); - if (!container.isActive()) { - container.afterPropertiesSet(); - } - } - - private void configureDispatcher(boolean isPubSub) { - if (isPubSub) { - this.dispatcher = new BroadcastingDispatcher(); - } - else { - UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); - unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); - this.dispatcher = unicastingDispatcher; - } - } - - public String getName() { - return this.name; - } - - public boolean subscribe(MessageHandler handler) { - return this.dispatcher.addHandler(handler); - } - - public boolean unsubscribe(MessageHandler handler) { - return this.dispatcher.removeHandler(handler); - } - - public boolean send(Message message) { - this.jmsTemplate.convertAndSend(message); - return true; - } - - public boolean send(Message message, long timeout) { - return this.send(message); - } - - - // MessageListener implementation - - public void onMessage(javax.jms.Message message) { - try { - Object converted = this.jmsTemplate.getMessageConverter().fromMessage(message); - if (converted != null) { - Message messageToSend = (converted instanceof Message) ? (Message) converted - : MessageBuilder.withPayload(converted).build(); - this.dispatcher.dispatch(messageToSend); - } - else if (this.logger.isWarnEnabled()) { - logger.warn("MessageConverter returned null, no Message to dispatch"); - } - } - catch (Exception e) { - throw new MessagingException("failed to handle incoming JMS Message", e); - } - } - -} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java deleted file mode 100644 index 5e1576a2eb..0000000000 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/MessageListenerContainerConfigurationSupport.java +++ /dev/null @@ -1,399 +0,0 @@ -/* - * Copyright 2002-2010 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 java.util.concurrent.Executor; - -import javax.jms.ConnectionFactory; -import javax.jms.Destination; -import javax.jms.ExceptionListener; -import javax.jms.Session; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.jms.listener.AbstractMessageListenerContainer; -import org.springframework.jms.listener.DefaultMessageListenerContainer; -import org.springframework.jms.listener.SimpleMessageListenerContainer; -import org.springframework.jms.support.destination.DestinationResolver; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.util.ErrorHandler; - -/** - * A base class for managing configurable properties of a MessageListenerContainer. - * - * @author Mark Fisher - * @since 2.0 - */ -abstract class MessageListenerContainerConfigurationSupport implements InitializingBean { - - private volatile AbstractMessageListenerContainer container; - - private volatile Class containerType; - - private volatile boolean acceptMessagesWhileStopping; - - private volatile boolean autoStartup = true; - - private volatile String cacheLevelName; - - private volatile String clientId; - - private volatile Integer concurrentConsumers; - - private volatile ConnectionFactory connectionFactory; - - private volatile Destination destination; - - private volatile String destinationName; - - private volatile DestinationResolver destinationResolver; - - private volatile String durableSubscriptionName; - - private volatile ErrorHandler errorHandler; - - private volatile ExceptionListener exceptionListener; - - private volatile Boolean exposeListenerSession; - - private volatile Integer idleTaskExecutionLimit; - - private volatile Integer maxConcurrentConsumers; - - private volatile Integer maxMessagesPerTask; - - private volatile String messageSelector; - - private volatile Integer phase; - - private volatile Boolean pubSubDomain; - - private volatile boolean pubSubNoLocal; - - private volatile Long receiveTimeout; - - private volatile Long recoveryInterval; - - /** - * This value differs from the container implementations' default (which is AUTO_ACKNOWLEDGE) - */ - private volatile int sessionAcknowledgeMode = Session.SESSION_TRANSACTED; - - /** - * This value differs from the container implementations' default (which is false). - */ - private volatile boolean sessionTransacted = true; - - private volatile boolean subscriptionDurable; - - private volatile Executor taskExecutor; - - private volatile PlatformTransactionManager transactionManager; - - private volatile String transactionName; - - private volatile Integer transactionTimeout; - - private volatile boolean initialized; - - private final Object initializationMonitor = new Object(); - - - public void setAcceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { - this.acceptMessagesWhileStopping = acceptMessagesWhileStopping; - } - - public void setAutoStartup(boolean autoStartup) { - this.autoStartup = autoStartup; - } - - public void setCacheLevelName(String cacheLevelName) { - this.cacheLevelName = cacheLevelName; - } - - public void setClientId(String clientId) { - this.clientId = clientId; - } - - public void setConcurrentConsumers(int concurrentConsumers) { - this.concurrentConsumers = concurrentConsumers; - } - - public void setConnectionFactory(ConnectionFactory connectionFactory) { - this.connectionFactory = connectionFactory; - } - - public void setContainerType(Class containerType) { - this.containerType = containerType; - } - - public void setDestination(Destination destination) { - this.destination = destination; - } - - public void setDestinationName(String destinationName) { - this.destinationName = destinationName; - } - - public void setDestinationResolver(DestinationResolver destinationResolver) { - this.destinationResolver = destinationResolver; - } - - public void setDurableSubscriptionName(String durableSubscriptionName) { - this.durableSubscriptionName = durableSubscriptionName; - } - - public void setErrorHandler(ErrorHandler errorHandler) { - this.errorHandler = errorHandler; - } - - public void setExceptionListener(ExceptionListener exceptionListener) { - this.exceptionListener = exceptionListener; - } - - public void setExposeListenerSession(boolean exposeListenerSession) { - this.exposeListenerSession = exposeListenerSession; - } - - public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) { - this.idleTaskExecutionLimit = idleTaskExecutionLimit; - } - - public void setMaxConcurrentConsumers(int maxConcurrentConsumers) { - this.maxConcurrentConsumers = maxConcurrentConsumers; - } - - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - this.maxMessagesPerTask = maxMessagesPerTask; - } - - public void setMessageSelector(String messageSelector) { - this.messageSelector = messageSelector; - } - - public void setPhase(int phase) { - this.phase = phase; - } - - public void setPubSubDomain(boolean pubSubDomain) { - this.pubSubDomain = pubSubDomain; - } - - public void setPubSubNoLocal(boolean pubSubNoLocal) { - this.pubSubNoLocal = pubSubNoLocal; - } - - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - - public void setRecoveryInterval(long recoveryInterval) { - this.recoveryInterval = recoveryInterval; - } - - public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) { - this.sessionAcknowledgeMode = sessionAcknowledgeMode; - } - - public void setSessionTransacted(boolean sessionTransacted) { - this.sessionTransacted = sessionTransacted; - } - - public void setSubscriptionDurable(boolean subscriptionDurable) { - this.subscriptionDurable = subscriptionDurable; - } - - public void setTaskExecutor(Executor taskExecutor) { - this.taskExecutor = taskExecutor; - } - - public void setTransactionManager(PlatformTransactionManager transactionManager) { - this.transactionManager = transactionManager; - } - - public void setTransactionName(String transactionName) { - this.transactionName = transactionName; - } - - public void setTransactionTimeout(int transactionTimeout) { - this.transactionTimeout = transactionTimeout; - } - - AbstractMessageListenerContainer getListenerContainer() { - if (!this.initialized) { - try { - this.initialize(); - } - catch (Exception e) { - throw new IllegalStateException("failed to initialize listener container", e); - } - } - return this.container; - } - - public void afterPropertiesSet() throws Exception { - this.initialize(); - } - - private void initialize() throws Exception { - synchronized (this.initializationMonitor) { - if (this.initialized) { - return; - } - if (this.containerType == null) { - this.containerType = DefaultMessageListenerContainer.class; - } - this.container = this.containerType.newInstance(); - this.container.setAcceptMessagesWhileStopping(this.acceptMessagesWhileStopping); - this.container.setAutoStartup(this.autoStartup); - this.container.setClientId(this.clientId); - this.container.setConnectionFactory(this.connectionFactory); - if (this.destination != null) { - this.container.setDestination(this.destination); - } - if (this.destinationName != null) { - this.container.setDestinationName(this.destinationName); - } - if (this.destinationResolver != null) { - this.container.setDestinationResolver(this.destinationResolver); - } - this.container.setDurableSubscriptionName(this.durableSubscriptionName); - this.container.setErrorHandler(this.errorHandler); - this.container.setExceptionListener(this.exceptionListener); - if (this.exposeListenerSession != null) { - this.container.setExposeListenerSession(this.exposeListenerSession); - } - this.container.setMessageSelector(this.messageSelector); - if (this.phase != null) { - this.container.setPhase(this.phase); - } - if (this.pubSubDomain != null) { - this.container.setPubSubDomain(this.pubSubDomain); - } - this.container.setSessionAcknowledgeMode(this.sessionAcknowledgeMode); - this.container.setSessionTransacted(this.sessionTransacted); - this.container.setSubscriptionDurable(this.subscriptionDurable); - if (this.container instanceof DefaultMessageListenerContainer) { - DefaultMessageListenerContainer dmlc = (DefaultMessageListenerContainer) this.container; - if (this.cacheLevelName != null) { - dmlc.setCacheLevelName(this.cacheLevelName); - } - if (this.concurrentConsumers != null) { - dmlc.setConcurrentConsumers(this.concurrentConsumers); - } - if (this.idleTaskExecutionLimit != null) { - dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit); - } - if (this.maxConcurrentConsumers != null) { - dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers); - } - if (this.maxMessagesPerTask != null) { - dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask); - } - dmlc.setPubSubNoLocal(this.pubSubNoLocal); - if (this.receiveTimeout != null) { - dmlc.setReceiveTimeout(this.receiveTimeout); - } - if (this.recoveryInterval != null) { - dmlc.setRecoveryInterval(this.recoveryInterval); - } - dmlc.setTaskExecutor(this.taskExecutor); - dmlc.setTransactionManager(this.transactionManager); - if (this.transactionName != null) { - dmlc.setTransactionName(this.transactionName); - } - if (this.transactionTimeout != null) { - dmlc.setTransactionTimeout(this.transactionTimeout); - } - } - else if (this.container instanceof SimpleMessageListenerContainer) { - SimpleMessageListenerContainer smlc = (SimpleMessageListenerContainer) this.container; - if (this.concurrentConsumers != null) { - smlc.setConcurrentConsumers(this.concurrentConsumers); - } - smlc.setPubSubNoLocal(this.pubSubNoLocal); - smlc.setTaskExecutor(this.taskExecutor); - } - this.initialized = true; - } - } - - // SmartLifecycle implementation (delegates to the MessageListener container) - - public int getPhase() { - return this.getListenerContainer().getPhase(); - } - - public boolean isAutoStartup() { - return this.autoStartup; - } - - public boolean isRunning() { - return this.initialized && this.container.isRunning(); - } - - /** - * Blocks until the listener container has subscribed; if the container does not support - * this test, or the caching mode is incompatible, true is returned. Otherwise blocks - * until timeout milliseconds have passed, or the consumer has registered. - * @see DefaultMessageListenerContainer#isRegisteredWithDestination() - * @param timeout Timeout in milliseconds. - * @return True if a subscriber has connected or the container/attributes does not support - * the test. False if a valid container does not have a registered consumer within - * timeout milliseconds. - */ - public boolean waitRegisteredWithDestination(long timeout) { - AbstractMessageListenerContainer container = this.getListenerContainer(); - if (container instanceof DefaultMessageListenerContainer) { - DefaultMessageListenerContainer listenerContainer = - (DefaultMessageListenerContainer) container; - if (listenerContainer.getCacheLevel() != DefaultMessageListenerContainer.CACHE_CONSUMER) { - return true; - } - while (timeout > 0) { - if (listenerContainer.isRegisteredWithDestination()) { - return true; - } - try { - Thread.sleep(100); - } catch (InterruptedException e) { } - timeout -= 100; - } - return false; - } - return true; - } - - public void start() { - this.getListenerContainer().start(); - } - - public void stop() { - if (this.isRunning()) { - this.container.stop(); - } - } - - public void stop(Runnable callback) { - if (this.isRunning()) { - this.container.stop(callback); - } - else { - callback.run(); - } - } - -} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java new file mode 100644 index 0000000000..5d85b15bb7 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java @@ -0,0 +1,68 @@ +/* + * Copyright 2002-2010 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.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.Message; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.jms.core.JmsTemplate; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel { + + private Log logger = LogFactory.getLog(this.getClass()); + + + public PollableJmsChannel(JmsTemplate jmsTemplate) { + super(jmsTemplate); + } + + + public Message receive() { + Object object = this.getJmsTemplate().receiveAndConvert(); + if (object == null) { + return null; + } + if (object instanceof Message) { + return (Message) object; + } + return MessageBuilder.withPayload(object).build(); + } + + public Message receive(long timeout) { + if (logger.isWarnEnabled() && this.timeoutConflictsWithTemplateValue(timeout)) { + logger.warn("The JmsTemplate's receiveTimeout value is always used for the JMS channel. " + + "Its current value is " + this.getJmsTemplate().getReceiveTimeout() + + ". The passed value of " + timeout + " will be ignored."); + } + return this.receive(); + } + + private boolean timeoutConflictsWithTemplateValue(long timeout) { + long templateTimeout = this.getJmsTemplate().getReceiveTimeout(); + return (timeout == -1 && templateTimeout != JmsTemplate.RECEIVE_TIMEOUT_INDEFINITE_WAIT) + || (timeout == 0 && templateTimeout != JmsTemplate.RECEIVE_TIMEOUT_NO_WAIT) + || (timeout != templateTimeout); + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java new file mode 100644 index 0000000000..3d5085df08 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java @@ -0,0 +1,162 @@ +/* + * Copyright 2002-2010 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.MessageListener; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.Message; +import org.springframework.integration.MessagingException; +import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.core.SubscribableChannel; +import org.springframework.integration.dispatcher.BroadcastingDispatcher; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy; +import org.springframework.integration.dispatcher.UnicastingDispatcher; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class SubscribableJmsChannel extends AbstractJmsChannel implements SubscribableChannel, SmartLifecycle, DisposableBean { + + private final AbstractMessageListenerContainer container; + + private volatile MessageDispatcher dispatcher; + + + public SubscribableJmsChannel(AbstractMessageListenerContainer container, JmsTemplate jmsTemplate) { + super(jmsTemplate); + Assert.notNull(container, "container must not be null"); + this.container = container; + } + + + public boolean subscribe(MessageHandler handler) { + return this.dispatcher.addHandler(handler); + } + + public boolean unsubscribe(MessageHandler handler) { + return this.dispatcher.removeHandler(handler); + } + + @Override + public void onInit() throws Exception { + super.onInit(); + this.configureDispatcher(this.container.isPubSubDomain()); + MessageListener listener = new DispatchingMessageListener(this.getJmsTemplate(), this.dispatcher); + this.container.setMessageListener(listener); + if (!this.container.isActive()) { + this.container.afterPropertiesSet(); + } + } + + private void configureDispatcher(boolean isPubSub) { + if (isPubSub) { + this.dispatcher = new BroadcastingDispatcher(); + } + else { + UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(); + unicastingDispatcher.setLoadBalancingStrategy(new RoundRobinLoadBalancingStrategy()); + this.dispatcher = unicastingDispatcher; + } + } + + + private static class DispatchingMessageListener implements MessageListener { + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final JmsTemplate jmsTemplate; + + private final MessageDispatcher dispatcher; + + + private DispatchingMessageListener(JmsTemplate jmsTemplate, MessageDispatcher dispatcher) { + this.jmsTemplate = jmsTemplate; + this.dispatcher = dispatcher; + } + + + public void onMessage(javax.jms.Message message) { + try { + Object converted = this.jmsTemplate.getMessageConverter().fromMessage(message); + if (converted != null) { + Message messageToSend = (converted instanceof Message) ? (Message) converted + : MessageBuilder.withPayload(converted).build(); + this.dispatcher.dispatch(messageToSend); + } + else if (this.logger.isWarnEnabled()) { + logger.warn("MessageConverter returned null, no Message to dispatch"); + } + } + catch (Exception e) { + throw new MessagingException("failed to handle incoming JMS Message", e); + } + } + } + + + /* + * SmartLifecycle implementation (delegates to the MessageListener container) + */ + + public boolean isAutoStartup() { + return (this.container != null) ? this.container.isAutoStartup() : false; + } + + public int getPhase() { + return (this.container != null) ? this.container.getPhase() : 0; + } + + public boolean isRunning() { + return (this.container != null) ? this.container.isRunning() : false; + } + + public void start() { + if (this.container != null) { + this.container.start(); + } + } + + public void stop() { + if (this.container != null) { + this.container.stop(); + } + } + + public void stop(Runnable callback) { + if (this.container != null) { + this.container.stop(callback); + } + } + + public void destroy() throws Exception { + if (this.container != null) { + this.container.destroy(); + } + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java new file mode 100644 index 0000000000..76d3296078 --- /dev/null +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java @@ -0,0 +1,442 @@ +/* + * Copyright 2002-2010 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.config; + +import java.util.concurrent.Executor; + +import javax.jms.ConnectionFactory; +import javax.jms.Destination; +import javax.jms.ExceptionListener; +import javax.jms.Session; + +import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.jms.AbstractJmsChannel; +import org.springframework.integration.jms.PollableJmsChannel; +import org.springframework.integration.jms.SubscribableJmsChannel; +import org.springframework.jms.core.JmsTemplate; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; +import org.springframework.jms.listener.SimpleMessageListenerContainer; +import org.springframework.jms.support.converter.MessageConverter; +import org.springframework.jms.support.destination.DestinationResolver; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.util.Assert; +import org.springframework.util.ErrorHandler; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class JmsChannelFactoryBean extends AbstractFactoryBean implements SmartLifecycle, DisposableBean { + + private volatile AbstractJmsChannel channel; + + private final boolean messageDriven; + + private final JmsTemplate jmsTemplate = new JmsTemplate(); + + private volatile AbstractMessageListenerContainer container; + + private volatile Class containerType; + + private volatile boolean acceptMessagesWhileStopping; + + private volatile boolean autoStartup = true; + + private volatile String cacheLevelName; + + private volatile String clientId; + + private volatile Integer concurrentConsumers; + + private volatile ConnectionFactory connectionFactory; + + private volatile Destination destination; + + private volatile String destinationName; + + private volatile DestinationResolver destinationResolver; + + private volatile String durableSubscriptionName; + + private volatile ErrorHandler errorHandler; + + private volatile ExceptionListener exceptionListener; + + private volatile Boolean exposeListenerSession; + + private volatile Integer idleTaskExecutionLimit; + + private volatile Integer maxConcurrentConsumers; + + private volatile Integer maxMessagesPerTask; + + private volatile String messageSelector; + + private volatile Integer phase; + + private volatile Boolean pubSubDomain; + + private volatile boolean pubSubNoLocal; + + private volatile Long receiveTimeout; + + private volatile Long recoveryInterval; + + /** + * This value differs from the container implementations' default (which is AUTO_ACKNOWLEDGE) + */ + private volatile int sessionAcknowledgeMode = Session.SESSION_TRANSACTED; + + /** + * This value differs from the container implementations' default (which is false). + */ + private volatile boolean sessionTransacted = true; + + private volatile boolean subscriptionDurable; + + private volatile Executor taskExecutor; + + private volatile PlatformTransactionManager transactionManager; + + private volatile String transactionName; + + private volatile Integer transactionTimeout; + + + public JmsChannelFactoryBean(boolean messageDriven) { + this.messageDriven = messageDriven; + } + + + /* + * Template properties + */ + + public void setDeliveryPersistent(boolean deliveryPersistent) { + this.jmsTemplate.setDeliveryPersistent(deliveryPersistent); + } + + public void setExplicitQosEnabled(boolean explicitQosEnabled) { + this.jmsTemplate.setExplicitQosEnabled(explicitQosEnabled); + } + + public void setMessageConverter(MessageConverter messageConverter) { + this.jmsTemplate.setMessageConverter(messageConverter); + } + + public void setMessageIdEnabled(boolean messageIdEnabled) { + this.jmsTemplate.setMessageIdEnabled(messageIdEnabled); + } + + public void setMessageTimestampEnabled(boolean messageTimestampEnabled) { + this.jmsTemplate.setMessageTimestampEnabled(messageTimestampEnabled); + } + + public void setPriority(int priority) { + this.jmsTemplate.setPriority(priority); + } + + public void setTimeToLive(long timeToLive) { + this.jmsTemplate.setTimeToLive(timeToLive); + } + + /* + * Container properties + */ + + public void setAcceptMessagesWhileStopping(boolean acceptMessagesWhileStopping) { + this.acceptMessagesWhileStopping = acceptMessagesWhileStopping; + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setCacheLevelName(String cacheLevelName) { + this.cacheLevelName = cacheLevelName; + } + + public void setClientId(String clientId) { + this.clientId = clientId; + } + + public void setConcurrentConsumers(int concurrentConsumers) { + this.concurrentConsumers = concurrentConsumers; + } + + public void setConnectionFactory(ConnectionFactory connectionFactory) { + this.connectionFactory = connectionFactory; + this.jmsTemplate.setConnectionFactory(this.connectionFactory); + } + + public void setContainerType(Class containerType) { + this.containerType = containerType; + } + + public void setDestination(Destination destination) { + this.destination = destination; + } + + public void setDestinationName(String destinationName) { + this.destinationName = destinationName; + } + + public void setDestinationResolver(DestinationResolver destinationResolver) { + this.destinationResolver = destinationResolver; + this.jmsTemplate.setDestinationResolver(destinationResolver); + } + + public void setDurableSubscriptionName(String durableSubscriptionName) { + this.durableSubscriptionName = durableSubscriptionName; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + + public void setExceptionListener(ExceptionListener exceptionListener) { + this.exceptionListener = exceptionListener; + } + + public void setExposeListenerSession(boolean exposeListenerSession) { + this.exposeListenerSession = exposeListenerSession; + } + + public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) { + this.idleTaskExecutionLimit = idleTaskExecutionLimit; + } + + public void setMaxConcurrentConsumers(int maxConcurrentConsumers) { + this.maxConcurrentConsumers = maxConcurrentConsumers; + } + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public void setMessageSelector(String messageSelector) { + this.messageSelector = messageSelector; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + public void setPubSubDomain(boolean pubSubDomain) { + this.pubSubDomain = pubSubDomain; + this.jmsTemplate.setPubSubDomain(pubSubDomain); + } + + public void setPubSubNoLocal(boolean pubSubNoLocal) { + this.pubSubNoLocal = pubSubNoLocal; + this.jmsTemplate.setPubSubNoLocal(pubSubNoLocal); + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public void setRecoveryInterval(long recoveryInterval) { + this.recoveryInterval = recoveryInterval; + } + + public void setSessionAcknowledgeMode(int sessionAcknowledgeMode) { + this.sessionAcknowledgeMode = sessionAcknowledgeMode; + this.jmsTemplate.setSessionAcknowledgeMode(sessionAcknowledgeMode); + } + + public void setSessionTransacted(boolean sessionTransacted) { + this.sessionTransacted = sessionTransacted; + this.jmsTemplate.setSessionTransacted(sessionTransacted); + } + + public void setSubscriptionDurable(boolean subscriptionDurable) { + this.subscriptionDurable = subscriptionDurable; + } + + public void setTaskExecutor(Executor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setTransactionName(String transactionName) { + this.transactionName = transactionName; + } + + public void setTransactionTimeout(int transactionTimeout) { + this.transactionTimeout = transactionTimeout; + } + + @Override + public Class getObjectType() { + return (this.channel != null) ? this.channel.getClass() : AbstractJmsChannel.class; + } + + @Override + protected AbstractJmsChannel createInstance() throws Exception { + this.initializeJmsTemplate(); + if (this.messageDriven) { + this.container = this.createContainer(); + this.channel = new SubscribableJmsChannel(this.container, this.jmsTemplate); + } + else { + this.channel = new PollableJmsChannel(this.jmsTemplate); + } + this.channel.afterPropertiesSet(); + return this.channel; + } + + private void initializeJmsTemplate() { + Assert.isTrue(this.destination != null ^ this.destinationName != null, + "Exactly one of destination or destinationName is required."); + if (this.destination != null) { + this.jmsTemplate.setDefaultDestination(this.destination); + } + if (this.destinationName != null) { + this.jmsTemplate.setDefaultDestinationName(this.destinationName); + } + } + + private AbstractMessageListenerContainer createContainer() throws Exception { + if (this.containerType == null) { + this.containerType = DefaultMessageListenerContainer.class; + } + AbstractMessageListenerContainer container = this.containerType.newInstance(); + container.setAcceptMessagesWhileStopping(this.acceptMessagesWhileStopping); + container.setAutoStartup(this.autoStartup); + container.setClientId(this.clientId); + container.setConnectionFactory(this.connectionFactory); + if (this.destination != null) { + container.setDestination(this.destination); + } + if (this.destinationName != null) { + container.setDestinationName(this.destinationName); + } + if (this.destinationResolver != null) { + container.setDestinationResolver(this.destinationResolver); + } + container.setDurableSubscriptionName(this.durableSubscriptionName); + container.setErrorHandler(this.errorHandler); + container.setExceptionListener(this.exceptionListener); + if (this.exposeListenerSession != null) { + container.setExposeListenerSession(this.exposeListenerSession); + } + container.setMessageSelector(this.messageSelector); + if (this.phase != null) { + container.setPhase(this.phase); + } + if (this.pubSubDomain != null) { + container.setPubSubDomain(this.pubSubDomain); + } + container.setSessionAcknowledgeMode(this.sessionAcknowledgeMode); + container.setSessionTransacted(this.sessionTransacted); + container.setSubscriptionDurable(this.subscriptionDurable); + if (container instanceof DefaultMessageListenerContainer) { + DefaultMessageListenerContainer dmlc = (DefaultMessageListenerContainer) container; + if (this.cacheLevelName != null) { + dmlc.setCacheLevelName(this.cacheLevelName); + } + if (this.concurrentConsumers != null) { + dmlc.setConcurrentConsumers(this.concurrentConsumers); + } + if (this.idleTaskExecutionLimit != null) { + dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit); + } + if (this.maxConcurrentConsumers != null) { + dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers); + } + if (this.maxMessagesPerTask != null) { + dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask); + } + dmlc.setPubSubNoLocal(this.pubSubNoLocal); + if (this.receiveTimeout != null) { + dmlc.setReceiveTimeout(this.receiveTimeout); + } + if (this.recoveryInterval != null) { + dmlc.setRecoveryInterval(this.recoveryInterval); + } + dmlc.setTaskExecutor(this.taskExecutor); + dmlc.setTransactionManager(this.transactionManager); + if (this.transactionName != null) { + dmlc.setTransactionName(this.transactionName); + } + if (this.transactionTimeout != null) { + dmlc.setTransactionTimeout(this.transactionTimeout); + } + } + else if (container instanceof SimpleMessageListenerContainer) { + SimpleMessageListenerContainer smlc = (SimpleMessageListenerContainer) container; + if (this.concurrentConsumers != null) { + smlc.setConcurrentConsumers(this.concurrentConsumers); + } + smlc.setPubSubNoLocal(this.pubSubNoLocal); + smlc.setTaskExecutor(this.taskExecutor); + } + return container; + } + + + /* + * SmartLifecycle implementation (delegates to the created channel if message-driven) + */ + + public boolean isAutoStartup() { + return (this.channel instanceof SubscribableJmsChannel) ? + ((SubscribableJmsChannel) this.channel).isAutoStartup() : false; + } + + public int getPhase() { + return (this.channel instanceof SubscribableJmsChannel) ? + ((SubscribableJmsChannel) this.channel).getPhase() : 0; + } + + public boolean isRunning() { + return (this.channel instanceof SubscribableJmsChannel) ? + ((SubscribableJmsChannel) this.channel).isRunning() : false; + } + + public void start() { + if (this.channel instanceof SubscribableJmsChannel) { + ((SubscribableJmsChannel) this.channel).start(); + } + } + + public void stop() { + if (this.channel instanceof SubscribableJmsChannel) { + ((SubscribableJmsChannel) this.channel).stop(); + } + } + + public void stop(Runnable callback) { + if (this.channel instanceof SubscribableJmsChannel) { + ((SubscribableJmsChannel) this.channel).stop(callback); + } + } + + protected void destroyInstance(AbstractJmsChannel instance) throws Exception { + if (instance instanceof SubscribableJmsChannel) { + ((SubscribableJmsChannel) this.channel).destroy(); + } + } + +} diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java index 549d0ae6c2..bcc972b687 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelParser.java @@ -45,16 +45,17 @@ public class JmsChannelParser extends AbstractSingleBeanDefinitionParser { @Override protected String getBeanClassName(Element element) { - return "org.springframework.integration.jms.JmsDestinationBackedMessageChannel"; + return "org.springframework.integration.jms.config.JmsChannelFactoryBean"; } @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + builder.addConstructorArgValue(element.getAttribute("message-driven")); String connectionFactory = element.getAttribute("connection-factory"); if (!StringUtils.hasText(connectionFactory)) { connectionFactory = "connectionFactory"; } - builder.addConstructorArgReference(connectionFactory); + builder.addPropertyReference("connectionFactory", connectionFactory); if ("channel".equals(element.getLocalName())) { this.parseDestination(element, parserContext, builder, "queue"); } @@ -135,11 +136,11 @@ public class JmsChannelParser extends AbstractSingleBeanDefinitionParser { "' or '" + type + "-name' attributes is required.", element); } if (isReference) { - builder.addConstructorArgReference(ref); + builder.addPropertyReference("destination", ref); } else if (isName) { - builder.addConstructorArgValue(name); - builder.addConstructorArgValue(isPubSub); + builder.addPropertyValue("destinationName", name); + builder.addPropertyValue("pubSubDomain", isPubSub); String destinationResolver = element.getAttribute("destination-resolver"); if (StringUtils.hasText(destinationResolver)) { builder.addPropertyReference("destinationResolver", destinationResolver); diff --git a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd index acb6f6c09d..1eeccb9cf1 100644 --- a/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd +++ b/spring-integration-jms/src/main/resources/org/springframework/integration/jms/config/spring-integration-jms-2.0.xsd @@ -117,6 +117,14 @@ + + + + Specifies whether this channel should be Message-Driven. The value is "true" by default. + Set to "false" if this channel should be pollable. + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java similarity index 68% rename from spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannelTests.java rename to spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java index 99cf7847bc..e6bc8410bb 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/JmsDestinationBackedMessageChannelTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/SubscribableJmsChannelTests.java @@ -36,16 +36,20 @@ import org.apache.activemq.command.ActiveMQTopic; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.jms.config.JmsChannelFactoryBean; import org.springframework.integration.message.GenericMessage; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; /** * @author Mark Fisher */ -public class JmsDestinationBackedMessageChannelTests { +public class SubscribableJmsChannelTests { private static final int TIMEOUT = 30000; @@ -82,8 +86,11 @@ public class JmsDestinationBackedMessageChannelTests { latch.countDown(); } }; - JmsDestinationBackedMessageChannel channel = - new JmsDestinationBackedMessageChannel(this.connectionFactory, this.queue); + JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true); + factoryBean.setConnectionFactory(this.connectionFactory); + factoryBean.setDestination(this.queue); + factoryBean.afterPropertiesSet(); + SubscribableJmsChannel channel = (SubscribableJmsChannel) factoryBean.getObject(); channel.afterPropertiesSet(); channel.start(); channel.subscribe(handler1); @@ -117,13 +124,16 @@ public class JmsDestinationBackedMessageChannelTests { latch.countDown(); } }; - JmsDestinationBackedMessageChannel channel = - new JmsDestinationBackedMessageChannel(this.connectionFactory, this.topic); + JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true); + factoryBean.setConnectionFactory(this.connectionFactory); + factoryBean.setDestination(this.topic); + factoryBean.afterPropertiesSet(); + SubscribableJmsChannel channel = (SubscribableJmsChannel) factoryBean.getObject(); channel.afterPropertiesSet(); channel.subscribe(handler1); channel.subscribe(handler2); channel.start(); - if (!channel.waitRegisteredWithDestination(10000)) { + if (!waitUntilRegisteredWithDestination(channel, 10000)) { fail("Listener failed to subscribe to topic"); } channel.send(new GenericMessage("foo")); @@ -155,8 +165,12 @@ public class JmsDestinationBackedMessageChannelTests { latch.countDown(); } }; - JmsDestinationBackedMessageChannel channel = - new JmsDestinationBackedMessageChannel(this.connectionFactory, "dynamicQueue", false); + JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true); + factoryBean.setConnectionFactory(this.connectionFactory); + factoryBean.setDestinationName("dynamicQueue"); + factoryBean.setPubSubDomain(false); + factoryBean.afterPropertiesSet(); + SubscribableJmsChannel channel = (SubscribableJmsChannel) factoryBean.getObject(); channel.afterPropertiesSet(); channel.start(); channel.subscribe(handler1); @@ -190,11 +204,16 @@ public class JmsDestinationBackedMessageChannelTests { latch.countDown(); } }; - JmsDestinationBackedMessageChannel channel = - new JmsDestinationBackedMessageChannel(this.connectionFactory, "dynamicTopic", true); + + JmsChannelFactoryBean factoryBean = new JmsChannelFactoryBean(true); + factoryBean.setConnectionFactory(this.connectionFactory); + factoryBean.setDestinationName("dynamicTopic"); + factoryBean.setPubSubDomain(true); + factoryBean.afterPropertiesSet(); + SubscribableJmsChannel channel = (SubscribableJmsChannel) factoryBean.getObject(); channel.afterPropertiesSet(); channel.start(); - if (!channel.waitRegisteredWithDestination(10000)) { + if (!waitUntilRegisteredWithDestination(channel, 10000)) { fail("Listener failed to subscribe to topic"); } channel.subscribe(handler1); @@ -211,15 +230,16 @@ public class JmsDestinationBackedMessageChannelTests { channel.stop(); } - @Test + @Test //@Ignore public void contextManagesLifecycle() { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsDestinationBackedMessageChannel.class); - builder.addConstructorArgValue(this.connectionFactory); - builder.addConstructorArgValue("dynamicQueue"); - builder.addConstructorArgValue(false); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(JmsChannelFactoryBean.class); + builder.addConstructorArgValue(true); + builder.addPropertyValue("connectionFactory", this.connectionFactory); + builder.addPropertyValue("destinationName", "dynamicQueue"); + builder.addPropertyValue("pubSubDomain", false); StaticApplicationContext context = new StaticApplicationContext(); context.registerBeanDefinition("channel", builder.getBeanDefinition()); - JmsDestinationBackedMessageChannel channel = context.getBean("channel", JmsDestinationBackedMessageChannel.class); + SubscribableJmsChannel channel = context.getBean("channel", SubscribableJmsChannel.class); assertFalse(channel.isRunning()); context.refresh(); assertTrue(channel.isRunning()); @@ -227,4 +247,38 @@ public class JmsDestinationBackedMessageChannelTests { assertFalse(channel.isRunning()); } + + /** + * Blocks until the listener container has subscribed; if the container does not support + * this test, or the caching mode is incompatible, true is returned. Otherwise blocks + * until timeout milliseconds have passed, or the consumer has registered. + * @see DefaultMessageListenerContainer#isRegisteredWithDestination() + * @param timeout Timeout in milliseconds. + * @return True if a subscriber has connected or the container/attributes does not support + * the test. False if a valid container does not have a registered consumer within + * timeout milliseconds. + */ + private static boolean waitUntilRegisteredWithDestination(SubscribableJmsChannel channel, long timeout) { + AbstractMessageListenerContainer container = + (AbstractMessageListenerContainer) new DirectFieldAccessor(channel).getPropertyValue("container"); + if (container instanceof DefaultMessageListenerContainer) { + DefaultMessageListenerContainer listenerContainer = + (DefaultMessageListenerContainer) container; + if (listenerContainer.getCacheLevel() != DefaultMessageListenerContainer.CACHE_CONSUMER) { + return true; + } + while (timeout > 0) { + if (listenerContainer.isRegisteredWithDestination()) { + return true; + } + try { + Thread.sleep(100); + } catch (InterruptedException e) { } + timeout -= 100; + } + return false; + } + return true; + } + } diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java index 5ff6c92684..d6d7c32469 100644 --- a/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2009 the original author or authors. + * Copyright 2002-2010 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. @@ -30,7 +30,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.MessageChannel; -import org.springframework.integration.jms.JmsDestinationBackedMessageChannel; +import org.springframework.integration.jms.SubscribableJmsChannel; import org.springframework.jms.core.JmsTemplate; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; @@ -75,8 +75,8 @@ public class JmsChannelParserTests { @Test public void queueReferenceChannel() { - assertEquals(JmsDestinationBackedMessageChannel.class, queueReferenceChannel.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) queueReferenceChannel; + assertEquals(SubscribableJmsChannel.class, queueReferenceChannel.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) queueReferenceChannel; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); @@ -86,8 +86,8 @@ public class JmsChannelParserTests { @Test public void queueNameChannel() { - assertEquals(JmsDestinationBackedMessageChannel.class, queueNameChannel.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) queueNameChannel; + assertEquals(SubscribableJmsChannel.class, queueNameChannel.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) queueNameChannel; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); @@ -97,8 +97,8 @@ public class JmsChannelParserTests { @Test public void queueNameWithResolverChannel() { - assertEquals(JmsDestinationBackedMessageChannel.class, queueNameWithResolverChannel.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) queueNameWithResolverChannel; + assertEquals(SubscribableJmsChannel.class, queueNameWithResolverChannel.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) queueNameWithResolverChannel; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); @@ -108,8 +108,8 @@ public class JmsChannelParserTests { @Test public void topicReferenceChannel() { - assertEquals(JmsDestinationBackedMessageChannel.class, topicReferenceChannel.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) topicReferenceChannel; + assertEquals(SubscribableJmsChannel.class, topicReferenceChannel.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) topicReferenceChannel; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); @@ -119,8 +119,8 @@ public class JmsChannelParserTests { @Test public void topicNameChannel() { - assertEquals(JmsDestinationBackedMessageChannel.class, topicNameChannel.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) topicNameChannel; + assertEquals(SubscribableJmsChannel.class, topicNameChannel.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) topicNameChannel; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); @@ -130,8 +130,8 @@ public class JmsChannelParserTests { @Test public void topicNameWithResolverChannel() { - assertEquals(JmsDestinationBackedMessageChannel.class, topicNameWithResolverChannel.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) topicNameWithResolverChannel; + assertEquals(SubscribableJmsChannel.class, topicNameWithResolverChannel.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) topicNameWithResolverChannel; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); JmsTemplate jmsTemplate = (JmsTemplate) accessor.getPropertyValue("jmsTemplate"); AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container"); @@ -141,8 +141,8 @@ public class JmsChannelParserTests { @Test public void channelWithConcurrencySettings() { - assertEquals(JmsDestinationBackedMessageChannel.class, channelWithConcurrencySettings.getClass()); - JmsDestinationBackedMessageChannel channel = (JmsDestinationBackedMessageChannel) channelWithConcurrencySettings; + assertEquals(SubscribableJmsChannel.class, channelWithConcurrencySettings.getClass()); + SubscribableJmsChannel channel = (SubscribableJmsChannel) channelWithConcurrencySettings; DirectFieldAccessor accessor = new DirectFieldAccessor(channel); DefaultMessageListenerContainer container = (DefaultMessageListenerContainer) accessor.getPropertyValue("container"); assertEquals(11, container.getConcurrentConsumers());