EmitterRuntimeEnvironment implements Runnable { public boolean isRunning() { return running; } EmitterRuntimeEnvironment(final String processingNodeId, final String pipelineId, final Emitter emitter, 
			final StreamingMessageQueueConsumer queueConsumer); void run(); void shutdown(); boolean isRunning(); void setMessageCounter(Counter messageCounter); void setMessageEmitDurationTimer(Timer messageEmitDurationTimer);  }
@Test public void testConstructor_withValidInput() throws RequiredInputMissingException { EmitterRuntimeEnvironment env = new EmitterRuntimeEnvironment("node-1", "pipe-1", Mockito.mock(Emitter.class), Mockito.mock(StreamingMessageQueueConsumer.class)); Assert.assertTrue("Must return true", env.isRunning()); }
@Test public void testEmitterEnvironment_withValidInput() throws RequiredInputMissingException, InterruptedException { Emitter emitter = Mockito.mock(Emitter.class); StreamingDataMessage message = new StreamingDataMessage("test".getBytes(), System.currentTimeMillis()); StreamingMessageQueueConsumer queueConsumer = Mockito.mock(StreamingMessageQueueConsumer.class); StreamingMessageQueueWaitStrategy queueConsumerWaitStrategy = Mockito.mock(StreamingMessageQueueWaitStrategy.class); Mockito.when(queueConsumer.getWaitStrategy()).thenReturn(queueConsumerWaitStrategy); Mockito.when(queueConsumerWaitStrategy.waitFor(queueConsumer)).thenReturn(message); EmitterRuntimeEnvironment env = new EmitterRuntimeEnvironment("node-1", "pipe-1", emitter, queueConsumer); executorService.submit(env); Assert.assertTrue("Must return true", env.isRunning()); Mockito.verify(queueConsumerWaitStrategy, Mockito.timeout(500).atLeastOnce()).waitFor(queueConsumer); Mockito.verify(queueConsumer, Mockito.atLeastOnce()).getWaitStrategy(); Mockito.verify(emitter, Mockito.atLeastOnce()).onMessage(message); }
@Test public void testEmitterEnvironment_withValidInputButFailingEmitter() throws RequiredInputMissingException, InterruptedException { Emitter emitter = Mockito.mock(Emitter.class); StreamingMessageQueueConsumer queueConsumer = Mockito.mock(StreamingMessageQueueConsumer.class); StreamingDataMessage message = new StreamingDataMessage("test".getBytes(), System.currentTimeMillis()); Mockito.when(queueConsumer.next()).thenReturn(message); Mockito.when(emitter.onMessage(message)).thenThrow(new NullPointerException("error")); StreamingMessageQueueWaitStrategy queueConsumerWaitStrategy = Mockito.mock(StreamingMessageQueueWaitStrategy.class); Mockito.when(queueConsumer.getWaitStrategy()).thenReturn(queueConsumerWaitStrategy); Mockito.when(queueConsumerWaitStrategy.waitFor(queueConsumer)).thenReturn(message); EmitterRuntimeEnvironment env = new EmitterRuntimeEnvironment("node-1", "pipe-1", emitter, queueConsumer); executorService.submit(env); Assert.assertTrue("Must return true", env.isRunning()); Mockito.verify(queueConsumerWaitStrategy, Mockito.timeout(500).atLeastOnce()).waitFor(queueConsumer); Mockito.verify(queueConsumer, Mockito.atLeastOnce()).getWaitStrategy(); Mockito.verify(emitter, Mockito.atLeastOnce()).onMessage(message); }