DirectResponseOperatorRuntimeEnvironment implements Runnable { public boolean isRunning() { return running; } DirectResponseOperatorRuntimeEnvironment(final String processingNodeId, final String pipelineId, final DirectResponseOperator directResponseOperator, final StreamingMessageQueueConsumer queueConsumer, 
			final StreamingMessageQueueProducer queueProducer); void run(); void shutdown(); boolean isRunning(); void setMessageCounter(Counter messageCounter); void setMessageProcessingTimer(Timer messageProcessingTimer);  }
@Test public void testConstructor_withValidInput() throws RequiredInputMissingException { DirectResponseOperatorRuntimeEnvironment env = new DirectResponseOperatorRuntimeEnvironment("proc-id", "pipe-id",Mockito.mock(DirectResponseOperator.class), Mockito.mock(StreamingMessageQueueConsumer.class), Mockito.mock(StreamingMessageQueueProducer.class)); Assert.assertTrue("The environment must be running", env.isRunning()); }
@Test public void testConstructor_withMessagesToProcessAndFailingOperator() throws RequiredInputMissingException, InterruptedException { StreamingDataMessage inputMessage = new StreamingDataMessage("test".getBytes(), System.currentTimeMillis()); DirectResponseOperator operator = Mockito.mock(DirectResponseOperator.class); Mockito.when(operator.onMessage(inputMessage)).thenThrow(new RuntimeException("Failed to process message")); StreamingMessageQueueConsumer queueConsumer = Mockito.mock(StreamingMessageQueueConsumer.class); StreamingMessageQueueWaitStrategy queueConsumerStrategy = Mockito.mock(StreamingMessageQueueWaitStrategy.class); Mockito.when(queueConsumer.getWaitStrategy()).thenReturn(queueConsumerStrategy); Mockito.when(queueConsumerStrategy.waitFor(queueConsumer)).thenReturn(inputMessage); StreamingMessageQueueProducer queueProducer = Mockito.mock(StreamingMessageQueueProducer.class); StreamingMessageQueueWaitStrategy queueProducerStrategy = Mockito.mock(StreamingMessageQueueWaitStrategy.class); Mockito.when(queueProducer.getWaitStrategy()).thenReturn(queueProducerStrategy); DirectResponseOperatorRuntimeEnvironment env = new DirectResponseOperatorRuntimeEnvironment("proc-id", "pipe-id",operator, queueConsumer, queueProducer); executorService.submit(env); Mockito.verify(queueProducer).getWaitStrategy(); Mockito.verify(queueConsumer).getWaitStrategy(); Mockito.verify(operator, Mockito.timeout(500).atLeastOnce()).onMessage(inputMessage); Mockito.verify(queueConsumerStrategy, Mockito.timeout(500).atLeastOnce()).waitFor(queueConsumer); Mockito.verify(queueProducerStrategy, Mockito.never()).forceLockRelease(); Mockito.verify(queueProducer, Mockito.never()).insert(inputMessage); Assert.assertTrue("The environment must be running", env.isRunning()); }
@Test public void testConstructor_withMessagesToProcess() throws RequiredInputMissingException, InterruptedException { StreamingDataMessage inputMessage = new StreamingDataMessage("test".getBytes(), System.currentTimeMillis()); DirectResponseOperator operator = Mockito.mock(DirectResponseOperator.class); Mockito.when(operator.onMessage(inputMessage)).thenReturn(new StreamingDataMessage[]{inputMessage}); StreamingMessageQueueConsumer queueConsumer = Mockito.mock(StreamingMessageQueueConsumer.class); StreamingMessageQueueWaitStrategy queueConsumerStrategy = Mockito.mock(StreamingMessageQueueWaitStrategy.class); Mockito.when(queueConsumer.getWaitStrategy()).thenReturn(queueConsumerStrategy); Mockito.when(queueConsumerStrategy.waitFor(queueConsumer)).thenReturn(inputMessage); StreamingMessageQueueProducer queueProducer = Mockito.mock(StreamingMessageQueueProducer.class); StreamingMessageQueueWaitStrategy queueProducerStrategy = Mockito.mock(StreamingMessageQueueWaitStrategy.class); Mockito.when(queueProducer.getWaitStrategy()).thenReturn(queueProducerStrategy); DirectResponseOperatorRuntimeEnvironment env = new DirectResponseOperatorRuntimeEnvironment("proc-id", "pipe-id",operator, queueConsumer, queueProducer); executorService.submit(env); Mockito.verify(queueProducer).getWaitStrategy(); Mockito.verify(queueConsumer).getWaitStrategy(); Mockito.verify(operator, Mockito.timeout(500).atLeastOnce()).onMessage(inputMessage); Mockito.verify(queueConsumerStrategy, Mockito.timeout(500).atLeastOnce()).waitFor(queueConsumer); Mockito.verify(queueProducerStrategy, Mockito.timeout(500).atLeastOnce()).forceLockRelease(); Mockito.verify(queueProducer, Mockito.timeout(500).atLeastOnce()).insert(inputMessage); Assert.assertTrue("The environment must be running", env.isRunning()); }