StoragePolicyProcessorJmsMessageListener { @Scheduled(fixedDelay = 60000) public void controlStoragePolicyProcessorJmsMessageListener() { try { Boolean jmsMessageListenerEnabled = Boolean.valueOf(configurationHelper.getProperty(ConfigurationValue.STORAGE_POLICY_PROCESSOR_JMS_LISTENER_ENABLED)); JmsListenerEndpointRegistry registry = ApplicationContextHolder.getApplicationContext() .getBean("org.springframework.jms.config.internalJmsListenerEndpointRegistry", JmsListenerEndpointRegistry.class); MessageListenerContainer jmsMessageListenerContainer = registry.getListenerContainer(HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE); LOGGER.debug("controlStoragePolicyProcessorJmsMessageListener(): {}={} jmsMessageListenerContainer.isRunning()={}", ConfigurationValue.STORAGE_POLICY_PROCESSOR_JMS_LISTENER_ENABLED.getKey(), jmsMessageListenerEnabled, jmsMessageListenerContainer.isRunning()); if (!jmsMessageListenerEnabled && jmsMessageListenerContainer.isRunning()) { LOGGER.info("controlStoragePolicyProcessorJmsMessageListener(): Stopping the storage policy processor JMS message listener ..."); jmsMessageListenerContainer.stop(); LOGGER.info("controlStoragePolicyProcessorJmsMessageListener(): Done"); } else if (jmsMessageListenerEnabled && !jmsMessageListenerContainer.isRunning()) { LOGGER.info("controlStoragePolicyProcessorJmsMessageListener(): Starting the storage policy processor JMS message listener ..."); jmsMessageListenerContainer.start(); LOGGER.info("controlStoragePolicyProcessorJmsMessageListener(): Done"); } } catch (Exception e) { LOGGER.error("controlStoragePolicyProcessorJmsMessageListener(): Failed to control the storage policy processor Jms message listener service.", e); } }  @Scheduled(fixedDelay = 60000) void controlStoragePolicyProcessorJmsMessageListener(); @JmsListener(id = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE, containerFactory = "storagePolicyProcessorJmsListenerContainerFactory", destination = HerdJmsDestinationResolver.SQS_DESTINATION_STORAGE_POLICY_SELECTOR_JOB_SQS_QUEUE) void processMessage(String payload, @Headers Map<Object, Object> allHeaders);  }
@Test public void testControlListener() { configurationHelper = Mockito.mock(ConfigurationHelper.class); ReflectionTestUtils.setField(storagePolicyProcessorJmsMessageListener, "configurationHelper", configurationHelper); MessageListenerContainer mockMessageListenerContainer = Mockito.mock(MessageListenerContainer.class); 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); when(mockMessageListenerContainer.isRunning()).thenReturn(false); storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener(); verify(mockMessageListenerContainer, Mockito.times(0)).stop(); verify(mockMessageListenerContainer, Mockito.times(0)).start(); when(mockMessageListenerContainer.isRunning()).thenReturn(true); storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener(); verify(mockMessageListenerContainer).stop(); when(configurationHelper.getProperty(ConfigurationValue.STORAGE_POLICY_PROCESSOR_JMS_LISTENER_ENABLED)).thenReturn("true"); when(mockMessageListenerContainer.isRunning()).thenReturn(true); storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener(); verify(mockMessageListenerContainer, Mockito.times(0)).start(); when(mockMessageListenerContainer.isRunning()).thenReturn(false); storagePolicyProcessorJmsMessageListener.controlStoragePolicyProcessorJmsMessageListener(); verify(mockMessageListenerContainer).start(); }