JmsQueueMonitorTask implements Runnable { @Override public void run() { getSafeguardLimits(); String jmxServiceUrl = JMX_SERVICE_PRE_URL + safeguardLimits.getJmxPort() + JMX_SERVICE_POST_URL; if (mBeanServerConnection == null || responseQueue == null || loggerQueue == null || dlqQueue == null || amqObject == null) { mBeanServerConnection = jmxConnectionBuilder.buildMBeanServerConnection(jmxServiceUrl); responseQueue = jmxObjectNameProvider.getObjectName(RESPONSE_QUEUE_OBJECT_NAME); loggerQueue = jmxObjectNameProvider.getObjectName(LOGGER_QUEUE_OBJECT_NAME); dlqQueue = jmxObjectNameProvider.getObjectName(DLQ_QUEUE_OBJECT_NAME); amqObject = jmxObjectNameProvider.getObjectName(AMQ_OBJECT_NAME); } Long totalQueueSize = retrieveQuerySize(); setSafeguardFlags(totalQueueSize); resetDlqAsNecessary(); resetAMQueueAsNecessary(totalQueueSize); }  @Override void run();  }
@Test public final void testRunShouldSwitchOnlyFIOffWhenFIOffLimitExceededButMWOffLimitNotExceeded() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(91)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(logger).info("Due to High load, FI decompression is turned OFF."); verify(logger, never()).info("Due to High load, Message Logging is turned OFF."); verify(logger, never()).info("Due to Normal load, Message Logging is restored."); verify(logger, never()).info("Due to Normal load, FI decompression is restored."); }
@Test public final void testRunShouldSwitchFIAndMWOffWhenFIAndMwOffLimitExceeded() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(191)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(logger).info("Due to High load, FI decompression is turned OFF."); verify(logger).info("Due to High load, Message Logging is turned OFF."); verify(logger, never()).info("Due to Normal load, Message Logging is restored."); verify(logger, never()).info("Due to Normal load, FI decompression is restored."); }
@Test public final void testRunShouldNotSwitchOffAnythingWhenLimitsAreNotExceeded() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(11)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(logger, never()).info("Due to High load, FI decompression is turned OFF."); verify(logger, never()).info("Due to High load, Message Logging is turned OFF."); verify(logger, never()).info("Due to Normal load, Message Logging is restored."); verify(logger, never()).info("Due to Normal load, FI decompression is restored."); }
@Test public final void testRunShouldntSwitchOnAnythingWhenQueueSizeDidntDropBelowAnyLimit() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", false); Whitebox.setInternalState(underTest, "messageWritingEnabled", false); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(111)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(logger, never()).info("Due to High load, FI decompression is turned OFF."); verify(logger, never()).info("Due to High load, Message Logging is turned OFF."); verify(logger, never()).info("Due to Normal load, Message Logging is restored."); verify(logger, never()).info("Due to Normal load, FI decompression is restored."); }
@Test public final void testRunShouldSwitchMWOnWhenQueueSizeDroppedBelowMWLimit() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", false); Whitebox.setInternalState(underTest, "messageWritingEnabled", false); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(58)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(logger, never()).info("Due to High load, FI decompression is turned OFF."); verify(logger, never()).info("Due to High load, Message Logging is turned OFF."); verify(logger).info("Due to Normal load, Message Logging is restored."); verify(logger, never()).info("Due to Normal load, FI decompression is restored."); }
@Test public final void testRunShouldSwitchMWAndFIOnWhenQueueSizeDroppedBelowFILimit() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", false); Whitebox.setInternalState(underTest, "messageWritingEnabled", false); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(2)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(3)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(logger, never()).info("Due to High load, FI decompression is turned OFF."); verify(logger, never()).info("Due to High load, Message Logging is turned OFF."); verify(logger).info("Due to Normal load, Message Logging is restored."); verify(logger).info("Due to Normal load, FI decompression is restored."); }
@Test(expectedExceptions = SystemException.class) public final void testRunShouldThrowExceptionWhenQueueSizeCannotBeRetrieved() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willThrow(new AttributeNotFoundException()); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(91)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); }
@Test(expectedExceptions = SystemException.class) public final void testRunShouldThrowExceptionWhenMemoryUsageCannotBeRetrieved() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(2)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(91)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willThrow(new AttributeNotFoundException()); underTest.run(); }
@Test public final void testRunShouldInitConnectionWhenLoggerQueueIsNull() throws Exception { Whitebox.setInternalState(underTest, "loggerQueue", null); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); givenJmxConnectionBuilder(); underTest.run(); verifyConnectionBuilder(); }
@Test public final void testRunShouldInitConnectionWhenResponseQueueIsNull() throws Exception { Whitebox.setInternalState(underTest, "responseQueue", null); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); givenJmxConnectionBuilder(); underTest.run(); verifyConnectionBuilder(); }
@Test public final void testRunShouldInitConnectionWhenMBeanServerConnectionIsNull() throws Exception { Whitebox.setInternalState(underTest, "mBeanServerConnection", null); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); givenJmxConnectionBuilder(); underTest.run(); verifyConnectionBuilder(); }
@Test public final void testRunShouldInitConnectionWhenAmqObjectIsNull() throws Exception { Whitebox.setInternalState(underTest, "amqObject", null); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); givenJmxConnectionBuilder(); underTest.run(); verifyConnectionBuilder(); }
@Test public final void testRunShouldSetQueueSizesIntoQueueSizeProvider() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(91)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(queueSizeProvider).setResponseQueueSize(new Long(12)); verify(queueSizeProvider).setLoggerQueueSize(new Long(91)); }
@Test public final void testRunShouldSaveQueueSizesIntoQueueSizeProvider() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(12)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(91)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); long loggerQueueSize = ((QueueSizeProvider) Whitebox.getInternalState(underTest, "queueSizeProvider")).getLoggerQueueSize(); Assert.assertEquals(loggerQueueSize, 91); long responseQueueSize = ((QueueSizeProvider) Whitebox.getInternalState(underTest, "queueSizeProvider")).getResponseQueueSize(); Assert.assertEquals(responseQueueSize, 12); }
@Test public final void testRunShouldPurgeDlqQueue() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(1)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(0)); underTest.run(); verify(mBeanServerConnection).invoke(dlqQueue, "purge", null, null); }
@Test public final void testRunShouldRestartAmqWhenAmqMemoryIsFull() throws Exception { Whitebox.setInternalState(underTest, "fIDecompressionEnabled", true); Whitebox.setInternalState(underTest, "messageWritingEnabled", true); QueueSizeProvider queueSizeProvider = new QueueSizeProvider(); Whitebox.setInternalState(underTest, "queueSizeProvider", queueSizeProvider); given(mBeanServerConnection.getAttribute(responseQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(loggerQueue, "QueueSize")).willReturn(new Long(0)); given(mBeanServerConnection.getAttribute(dlqQueue, "QueueSize")).willReturn(new Long(1)); given(mBeanServerConnection.getAttribute(amqObject, "MemoryPercentUsage")).willReturn(new Integer(JmsQueueMonitorTask.MAX_AMQ_MEMORY_USAGE + 1)); underTest.run(); verify(mBeanServerConnection).invoke(amqObject, "restart", null, null); }