INT-813, INT-965 refactoring JmsDestinationBackedMessageChannel into SubscribableJmsChannel and PollableJmsChannel with a common FactoryBean for creation
This commit is contained in:
Mark Fisher
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<? extends AbstractMessageListenerContainer> 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<? extends AbstractMessageListenerContainer> 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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<AbstractJmsChannel> implements SmartLifecycle, DisposableBean {
|
||||
|
||||
private volatile AbstractJmsChannel channel;
|
||||
|
||||
private final boolean messageDriven;
|
||||
|
||||
private final JmsTemplate jmsTemplate = new JmsTemplate();
|
||||
|
||||
private volatile AbstractMessageListenerContainer container;
|
||||
|
||||
private volatile Class<? extends AbstractMessageListenerContainer> 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<? extends AbstractMessageListenerContainer> 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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
@@ -117,6 +117,14 @@
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="message-driven" type="xsd:boolean" default="true">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Specifies whether this channel should be Message-Driven. The value is "true" by default.
|
||||
Set to "false" if this channel should be pollable.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="connection-factory" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
|
||||
@@ -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<String>("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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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());
|
||||
|
||||
Reference in New Issue
Block a user