import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * Create a message listener container for the given {@link JmsListenerEndpoint}.
 * <p>This create the necessary infrastructure to honor that endpoint
 * with regards to its configuration.
 * <p>The {@code startImmediately} flag determines if the container should be
 * started immediately.
 * @param endpoint the endpoint to add
 * @param factory the listener factory to use
 * @param startImmediately start the container immediately if necessary
 * @see #getListenerContainers()
 * @see #getListenerContainer(String)
 */
public void registerListenerContainer(JmsListenerEndpoint endpoint, JmsListenerContainerFactory<?> factory,
		boolean startImmediately) {

	Assert.notNull(endpoint, "Endpoint must not be null");
	Assert.notNull(factory, "Factory must not be null");

	String id = endpoint.getId();
	Assert.notNull(id, "Endpoint id must not be null");
	synchronized (this.listenerContainers) {
		Assert.state(!this.listenerContainers.containsKey(id),
				"Another endpoint is already registered with id '" + id + "'");
		MessageListenerContainer container = createListenerContainer(endpoint, factory);
		this.listenerContainers.put(id, container);
		if (startImmediately) {
			startIfNecessary(container);
		}
	}
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * Create and start a new container using the specified factory.
 */
protected MessageListenerContainer createListenerContainer(JmsListenerEndpoint endpoint,
		JmsListenerContainerFactory<?> factory) {

	MessageListenerContainer listenerContainer = factory.createListenerContainer(endpoint);

	if (listenerContainer instanceof InitializingBean) {
		try {
			((InitializingBean) listenerContainer).afterPropertiesSet();
		}
		catch (Exception ex) {
			throw new BeanInitializationException("Failed to initialize message listener container", ex);
		}
	}

	int containerPhase = listenerContainer.getPhase();
	if (containerPhase < Integer.MAX_VALUE) {  // a custom phase value
		if (this.phase < Integer.MAX_VALUE && this.phase != containerPhase) {
			throw new IllegalStateException("Encountered phase mismatch between container factory definitions: " +
					this.phase + " vs " + containerPhase);
		}
		this.phase = listenerContainer.getPhase();
	}

	return listenerContainer;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Test
public void setupMessageContainerUnsupportedContainer() {
	MessageListenerContainer container = mock(MessageListenerContainer.class);
	SimpleJmsListenerEndpoint endpoint = new SimpleJmsListenerEndpoint();
	endpoint.setMessageListener(new MessageListenerAdapter());

	thrown.expect(IllegalArgumentException.class);
	endpoint.setupListenerContainer(container);
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Bean
public MessageListenerContainer hornetQMessageListenerContainer() {
	Map<String, Object> map = new HashMap<String, Object>();
       map.put(TransportConstants.HOST_PROP_NAME, hornetq_host);  
       map.put(TransportConstants.PORT_PROP_NAME, hornetq_port);
	TransportConfiguration transportConfiguration =new TransportConfiguration(
			NettyConnectorFactory.class.getName(), map);
	
	HornetQConnectionFactory hornetQConnectionFactory = HornetQJMSClient.createConnectionFactoryWithoutHA(
			JMSFactoryType.CF, transportConfiguration);
	
	DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
	container.setConnectionFactory((ConnectionFactory) hornetQConnectionFactory);
	container.setDestination(HornetQJMSClient.createTopic(IMConstant.GATEWAY_USERREGISTER_TOPIC));
	container.setPubSubDomain(true);
	container.setPubSubNoLocal(true);
	container.setMessageListener(userRegisterListener);
	container.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
	return container;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Override
public void setupListenerContainer(MessageListenerContainer listenerContainer) {
	if (listenerContainer instanceof AbstractMessageListenerContainer) {
		setupJmsListenerContainer((AbstractMessageListenerContainer) listenerContainer);
	}
	else {
		new JcaEndpointConfigurer().configureEndpoint(listenerContainer);
	}
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Override
public void destroy() {
	for (MessageListenerContainer listenerContainer : getListenerContainers()) {
		if (listenerContainer instanceof DisposableBean) {
			try {
				((DisposableBean) listenerContainer).destroy();
			}
			catch (Throwable ex) {
				logger.warn("Failed to destroy message listener container", ex);
			}
		}
	}
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Override
public void stop(Runnable callback) {
	Collection<MessageListenerContainer> listenerContainers = getListenerContainers();
	AggregatingCallback aggregatingCallback = new AggregatingCallback(listenerContainers.size(), callback);
	for (MessageListenerContainer listenerContainer : listenerContainers) {
		listenerContainer.stop(aggregatingCallback);
	}
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Override
public boolean isRunning() {
	for (MessageListenerContainer listenerContainer : getListenerContainers()) {
		if (listenerContainer.isRunning()) {
			return true;
		}
	}
	return false;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Override
protected MessagingMessageListenerAdapter createMessageListener(MessageListenerContainer container) {
	Assert.state(this.messageHandlerMethodFactory != null,
			"Could not create message listener - MessageHandlerMethodFactory not set");
	MessagingMessageListenerAdapter messageListener = createMessageListenerInstance();
	InvocableHandlerMethod invocableHandlerMethod =
			this.messageHandlerMethodFactory.createInvocableHandlerMethod(getBean(), getMethod());
	messageListener.setHandlerMethod(invocableHandlerMethod);
	String responseDestination = getDefaultResponseDestination();
	if (StringUtils.hasText(responseDestination)) {
		if (container.isReplyPubSubDomain()) {
			messageListener.setDefaultResponseTopicName(responseDestination);
		}
		else {
			messageListener.setDefaultResponseQueueName(responseDestination);
		}
	}
	MessageConverter messageConverter = container.getMessageConverter();
	if (messageConverter != null) {
		messageListener.setMessageConverter(messageConverter);
	}
	DestinationResolver destinationResolver = container.getDestinationResolver();
	if (destinationResolver != null) {
		messageListener.setDestinationResolver(destinationResolver);
	}
	return messageListener;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
private MessagingMessageListenerAdapter createInstance(
		DefaultMessageHandlerMethodFactory factory, Method method, MessageListenerContainer container) {

	MethodJmsListenerEndpoint endpoint = new MethodJmsListenerEndpoint();
	endpoint.setBean(sample);
	endpoint.setMethod(method);
	endpoint.setMessageHandlerMethodFactory(factory);
	return endpoint.createMessageListener(container);
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Bean
public MessageListenerContainer mockServerListener(ActiveMQConnectionFactory connectionFactory) {
  DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
  container.setConnectionFactory(connectionFactory);
  container.setDestination(new ActiveMQQueue("c2mon.client.request"));
  container.setMessageListener((SessionAwareMessageListener) (message, session) -> {
    session.createProducer(message.getJMSReplyTo()).send(session.createTextMessage("[]"));
  });
  return container;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * 配置基于发布订阅模式的消息监听容器,用于接收MQ转发过来的消息
 * 
 * @return
 */
@Bean
public MessageListenerContainer mqMessageReceiver(ImBrokerMessageListener imBrokerMessageListener) {
    MessageListenerContainer container = mqMessageReceiver(IMConstant.MQ_FORWARD_TOPIC_NAME, imBrokerMessageListener);
    logger.info("Message listener based on pub-sub mode was created!");
    return container;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * 构建基于队列模式的消息监听容器
 *
 * @param queueName 队列名字
 * @param rpcServiceExport 服务暴露接口
 * @return
 */
protected MessageListenerContainer buildListenerContainer(JmsInvokerServiceExporter rpcServiceExport,
        String queueName) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(jmsConnectionFactory());
    container.setDestinationName(queueName);
    container.setMessageConverter(messageConverter());
    container.setMessageListener(rpcServiceExport);
    container.setConcurrentConsumers(Runtime.getRuntime().availableProcessors());
    return container;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * 配置基于发布订阅模式的消息监听容器,用于接收MQ转发过来的消息
 *
 * @param topicName
 * @param messageListener
 * @return
 */
public MessageListenerContainer mqMessageReceiver(String topicName, MessageListener messageListener) {
    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
    container.setConnectionFactory(jmsConnectionFactory());
    container.setDestinationName(topicName);
    container.setPubSubDomain(true);
    container.setPubSubNoLocal(true);
    container.setMessageListener(messageListener);
    container.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
    return container;
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * Periodically check the configuration and apply the action to the storage policy processor JMS message listener service, if needed.
 */
@Scheduled(fixedDelay = 60000)
public void controlSearchIndexUpdateJmsMessageListener()
{
    try
    {
        // Get the configuration setting.
        Boolean jmsMessageListenerEnabled = Boolean.valueOf(configurationHelper.getProperty(ConfigurationValue.SEARCH_INDEX_UPDATE_JMS_LISTENER_ENABLED));

        // Get the registry bean.
        JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext()
            .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class);

        // Get the search index update JMS message listener container.
        MessageListenerContainer jmsMessageListenerContainer =
            registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_SEARCH_INDEX_UPDATE_QUEUE);

        // Get the current JMS message listener status and the configuration value.
        LOGGER.debug("controlSearchIndexUpdateJmsMessageListener(): {}={} jmsMessageListenerContainer.isRunning()={}",
            ConfigurationValue.SEARCH_INDEX_UPDATE_JMS_LISTENER_ENABLED.getKey(), jmsMessageListenerEnabled, jmsMessageListenerContainer.isRunning());

        // Apply the relative action if needed.
        if (!jmsMessageListenerEnabled && jmsMessageListenerContainer.isRunning())
        {
            LOGGER.info("controlSearchIndexUpdateJmsMessageListener(): Stopping the search index update JMS message listener ...");
            jmsMessageListenerContainer.stop();
            LOGGER.info("controlSearchIndexUpdateJmsMessageListener(): Done");
        }
        else if (jmsMessageListenerEnabled && !jmsMessageListenerContainer.isRunning())
        {
            LOGGER.info("controlSearchIndexUpdateJmsMessageListener(): Starting the search index update JMS message listener ...");
            jmsMessageListenerContainer.start();
            LOGGER.info("controlSearchIndexUpdateJmsMessageListener(): Done");
        }
    }
    catch (Exception e)
    {
        LOGGER.error("controlSearchIndexUpdateJmsMessageListener(): Failed to control the search index update Jms message listener service.", e);
    }
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
/**
 * Periodically check the configuration and apply the action to the herd JMS message listener service, if needed.
 */
@Scheduled(fixedDelay = 60000)
public void controlHerdJmsMessageListener()
{
    try
    {
        // Get the configuration setting.
        Boolean jmsMessageListenerEnabled = Boolean.valueOf(configurationHelper.getProperty(ConfigurationValue.JMS_LISTENER_ENABLED));

        // Get the registry bean.
        JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext()
            .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class);

        // Get the herd JMS message listener container.
        MessageListenerContainer jmsMessageListenerContainer = registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_HERD_INCOMING);

        // Get the current JMS message listener status and the configuration value.
        LOGGER.debug("controlHerdJmsMessageListener(): {}={} jmsMessageListenerContainer.isRunning()={}", ConfigurationValue.JMS_LISTENER_ENABLED.getKey(),
            jmsMessageListenerEnabled, jmsMessageListenerContainer.isRunning());

        // Apply the relative action if needed.
        if (!jmsMessageListenerEnabled && jmsMessageListenerContainer.isRunning())
        {
            LOGGER.info("controlHerdJmsMessageListener(): Stopping the herd JMS message listener ...");
            jmsMessageListenerContainer.stop();
            LOGGER.info("controlHerdJmsMessageListener(): Done");
        }
        else if (jmsMessageListenerEnabled && !jmsMessageListenerContainer.isRunning())
        {
            LOGGER.info("controlHerdJmsMessageListener(): Starting the herd JMS message listener ...");
            jmsMessageListenerContainer.start();
            LOGGER.info("controlHerdJmsMessageListener(): Done");
        }
    }
    catch (Exception e)
    {
        LOGGER.error("controlHerdJmsMessageListener(): Failed to control the herd Jms message listener service.", e);
    }
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Test
public void testControlListener()
{
    configurationHelper = Mockito.mock(ConfigurationHelper.class);

    ReflectionTestUtils.setField(sampleDataJmsMessageListener, "configurationHelper", configurationHelper);
    MessageListenerContainer mockMessageListenerContainer = Mockito.mock(MessageListenerContainer.class);

    //The listener is not enabled
    when(configurationHelper.getProperty(ConfigurationValue.SAMPLE_DATA_JMS_LISTENER_ENABLED)).thenReturn("false");
    JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext()
        .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class);
    when(registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_SAMPLE_DATA_QUEUE)).thenReturn(mockMessageListenerContainer);
    //the listener is not running, nothing happened
    when(mockMessageListenerContainer.isRunning()).thenReturn(false);
    sampleDataJmsMessageListener.controlSampleDataJmsMessageListener();
    verify(mockMessageListenerContainer, Mockito.times(0)).stop();
    verify(mockMessageListenerContainer, Mockito.times(0)).start();
    // the listener is running, but it is not enable, should stop
    when(mockMessageListenerContainer.isRunning()).thenReturn(true);
    sampleDataJmsMessageListener.controlSampleDataJmsMessageListener();
    verify(mockMessageListenerContainer).stop();

    //The listener is enabled
    when(configurationHelper.getProperty(ConfigurationValue.SAMPLE_DATA_JMS_LISTENER_ENABLED)).thenReturn("true");
    //the listener is running, should not call the start method
    when(mockMessageListenerContainer.isRunning()).thenReturn(true);
    sampleDataJmsMessageListener.controlSampleDataJmsMessageListener();
    verify(mockMessageListenerContainer, Mockito.times(0)).start();
    // the listener is not running, but it is enabled, should start        
    when(mockMessageListenerContainer.isRunning()).thenReturn(false);
    sampleDataJmsMessageListener.controlSampleDataJmsMessageListener();
    verify(mockMessageListenerContainer).start();
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Test
public void testControlListener()
{
    ReflectionTestUtils.setField(searchIndexUpdateJmsMessageListener, "configurationHelper", configurationHelper);
    MessageListenerContainer mockMessageListenerContainer = Mockito.mock(MessageListenerContainer.class);

    // The listener is not enabled
    when(configurationHelper.getProperty(ConfigurationValue.SEARCH_INDEX_UPDATE_JMS_LISTENER_ENABLED)).thenReturn("false");
    JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext()
        .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class);
    when(registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_SEARCH_INDEX_UPDATE_QUEUE)).thenReturn(mockMessageListenerContainer);

    // The listener is not running, nothing happened
    when(mockMessageListenerContainer.isRunning()).thenReturn(false);
    searchIndexUpdateJmsMessageListener.controlSearchIndexUpdateJmsMessageListener();
    verify(mockMessageListenerContainer, times(0)).stop();
    verify(mockMessageListenerContainer, times(0)).start();

    // The listener is running, but it is not enable, should stop
    when(mockMessageListenerContainer.isRunning()).thenReturn(true);
    searchIndexUpdateJmsMessageListener.controlSearchIndexUpdateJmsMessageListener();
    verify(mockMessageListenerContainer, times(1)).stop();

    // The listener is enabled
    when(configurationHelper.getProperty(ConfigurationValue.SEARCH_INDEX_UPDATE_JMS_LISTENER_ENABLED)).thenReturn("true");

    // The listener is running, should not call the start method
    when(mockMessageListenerContainer.isRunning()).thenReturn(true);
    searchIndexUpdateJmsMessageListener.controlSearchIndexUpdateJmsMessageListener();
    verify(mockMessageListenerContainer, times(0)).start();

    // The listener is not running, but it is enabled, should start
    when(mockMessageListenerContainer.isRunning()).thenReturn(false);
    searchIndexUpdateJmsMessageListener.controlSearchIndexUpdateJmsMessageListener();
    verify(mockMessageListenerContainer, times(1)).start();
}
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Test
 public void testControlListener()
{
     configurationHelper = Mockito.mock(ConfigurationHelper.class);

     ReflectionTestUtils.setField(herdJmsMessageListener, "configurationHelper", configurationHelper);
     MessageListenerContainer mockMessageListenerContainer = Mockito.mock(MessageListenerContainer.class);

     //The listener is not enabled
     when(configurationHelper.getProperty(ConfigurationValue.JMS_LISTENER_ENABLED)).thenReturn("false");
     JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext()
             .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class);
     when(registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_HERD_INCOMING)).thenReturn(mockMessageListenerContainer);
     //the listener is not running, nothing happened
     when(mockMessageListenerContainer.isRunning()).thenReturn(false);
     herdJmsMessageListener.controlHerdJmsMessageListener();
     verify(mockMessageListenerContainer, Mockito.times(0)).stop();
     verify(mockMessageListenerContainer, Mockito.times(0)).start();
     // the listener is running, but it is not enable, should stop
     when(mockMessageListenerContainer.isRunning()).thenReturn(true);
     herdJmsMessageListener.controlHerdJmsMessageListener();
     verify(mockMessageListenerContainer).stop();
     
     //The listener is enabled
     when(configurationHelper.getProperty(ConfigurationValue.JMS_LISTENER_ENABLED)).thenReturn("true");
     //the listener is running, should not call the start method
     when(mockMessageListenerContainer.isRunning()).thenReturn(true);
     herdJmsMessageListener.controlHerdJmsMessageListener();
     verify(mockMessageListenerContainer, Mockito.times(0)).start();     
     // the listener is not running, but it is enabled, should start        
     when(mockMessageListenerContainer.isRunning()).thenReturn(false);
     herdJmsMessageListener.controlHerdJmsMessageListener();
     verify(mockMessageListenerContainer).start();
 }
 

import org.springframework.jms.listener.MessageListenerContainer; //导入依赖的package包/类
@Test
public void testControlListener()
{
    configurationHelper = Mockito.mock(ConfigurationHelper.class);

    ReflectionTestUtils.setField(storagePolicyProcessorJmsMessageListener, "configurationHelper", configurationHelper);
    MessageListenerContainer mockMessageListenerContainer = Mockito.mock(MessageListenerContainer.class);

    //The listener is not enabled
    when(configurationHelper.getProperty(ConfigurationValue.STORAGE_POLICY_PROCESSOR_JMS_LISTENER_ENABLED)).thenReturn("false");
    JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext()
        .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class);
    when(registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE))
        .thenReturn(mockMessageListenerContainer);
    //the listener is not running, nothing happened
    when(mockMessageListenerContainer.isRunning()).thenReturn(false);
    storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener();
    verify(mockMessageListenerContainer, Mockito.times(0)).stop();
    verify(mockMessageListenerContainer, Mockito.times(0)).start();
    // the listener is running, but it is not enable, should stop
    when(mockMessageListenerContainer.isRunning()).thenReturn(true);
    storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener();
    verify(mockMessageListenerContainer).stop();

    //The listener is enabled
    when(configurationHelper.getProperty(ConfigurationValue.STORAGE_POLICY_PROCESSOR_JMS_LISTENER_ENABLED)).thenReturn("true");
    //the listener is running, should not call the start method
    when(mockMessageListenerContainer.isRunning()).thenReturn(true);
    storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener();
    verify(mockMessageListenerContainer, Mockito.times(0)).start();
    // the listener is not running, but it is enabled, should start
    when(mockMessageListenerContainer.isRunning()).thenReturn(false);
    storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener();
    verify(mockMessageListenerContainer).start();
}