SourceRuntimeEnvironment implements Runnable, IncomingMessageCallback { public void shutdown() { try { this.source.shutdown(); } catch(Exception e) { logger.error("source shutdown error [node="+this.processingNodeId+", pipeline="+this.pipelineId+", source="+this.sourceId+"]: " + e.getMessage(), e); } if(this.localExecutorService) { try { this.executorService.shutdownNow(); } catch(Exception e) { logger.error("exec service shutdown error [node="+this.processingNodeId+", pipeline="+this.pipelineId+", source="+this.sourceId+"]: " + e.getMessage(), e); } } if(logger.isDebugEnabled()) logger.debug("source shutdown [node="+this.processingNodeId+", pipeline="+this.pipelineId+", source="+this.sourceId+"]"); } SourceRuntimeEnvironment(final String processingNodeId, final String pipelineId, final Source source, final StreamingMessageQueueProducer queueProducer);  SourceRuntimeEnvironment(final String processingNodeId, final String pipelineId, final Source source, final StreamingMessageQueueProducer queueProducer, 
			final ExecutorService executorService); void run(); void onMessage(StreamingDataMessage message); void shutdown(); void setMessageCounter(final Counter counter);  }
@Test public void testRun_withFullyFunctionalSource() throws RequiredInputMissingException, IOException, InterruptedException, ComponentInitializationFailedException { final int numGenerated = 10000; final CountDownLatch latch = new CountDownLatch(numGenerated); ExecutorService svc = Executors.newCachedThreadPool(); Properties rndGenProps = new Properties(); rndGenProps.setProperty(RandomNumberTestSource.CFG_MAX_NUM_GENERATED, String.valueOf(numGenerated)); RandomNumberTestSource source = new RandomNumberTestSource(); source.initialize(rndGenProps); source.setId("testRun_withFullyFunctionalSource"); InMemoryStreamingMessageQueue queue = new InMemoryStreamingMessageQueue(); queue.setId("testRun_withFullyFunctionalSource"); queue.initialize(new Properties()); final StreamingMessageQueueConsumer consumer = queue.getConsumer(); SourceRuntimeEnvironment env = null; try { env = new SourceRuntimeEnvironment("node", "pipe", source, queue.getProducer()); svc.submit(env); svc.submit(new Runnable() { public void run() { int numElements = numGenerated; while(numElements > 0) { if(consumer.next() != null) { numElements--; latch.countDown(); } } } }); Assert.assertTrue(latch.await(5, TimeUnit.SECONDS)); } finally { env.shutdown(); svc.shutdownNow(); } }