MicroPipelineManager { public String executePipeline(final MicroPipelineConfiguration configuration) throws RequiredInputMissingException, QueueInitializationFailedException, ComponentInitializationFailedException, PipelineInstantiationFailedException, NonUniqueIdentifierException { if(configuration == null) throw new RequiredInputMissingException("Missing required pipeline configuration"); String id = StringUtils.lowerCase(StringUtils.trim(configuration.getId())); if(this.pipelines.containsKey(id)) throw new NonUniqueIdentifierException("A pipeline already exists for id '"+id+"'"); MicroPipeline pipeline = this.microPipelineFactory.instantiatePipeline(configuration, this.executorService); if(pipeline != null) this.pipelines.put(id, pipeline); else throw new PipelineInstantiationFailedException("Failed to instantiate pipeline '"+configuration.getId()+"'. Reason: null returned by pipeline factory"); if(logger.isDebugEnabled()) logger.debug("pipeline registered[id="+configuration.getId()+"]"); return id; } MicroPipelineManager(final String processingNodeId, final ComponentRepository componentRepository, final int maxNumberOfThreads); protected  MicroPipelineManager(final String processingNodeId, final MicroPipelineFactory factory, final ExecutorService executorService); String executePipeline(final MicroPipelineConfiguration configuration); String shutdownPipeline(final String pipelineId); boolean hasPipeline(final String pipelineId); int getNumOfRegisteredPipelines(); Set<String> getPipelineIds(); Map<String, MicroPipelineConfiguration> getPipelineConfigurations(); void shutdown(); String getProcessingNodeId();  }
@Test public void testExecutePipeline_withNullInput() throws RequiredInputMissingException, QueueInitializationFailedException, ComponentInitializationFailedException, PipelineInstantiationFailedException, NonUniqueIdentifierException { MicroPipelineFactory factory = Mockito.mock(MicroPipelineFactory.class); MicroPipelineManager manager = new MicroPipelineManager("id", factory, executorService); try { manager.executePipeline(null); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testExecutePipeline_withConfigurationLeadingToFactoryException() throws Exception { MicroPipelineConfiguration cfg = new MicroPipelineConfiguration(); cfg.setId("testExecutePipeline_withConfigurationLeadingToFactoryException"); MicroPipelineFactory factory = Mockito.mock(MicroPipelineFactory.class); Mockito.when(factory.instantiatePipeline(cfg, executorService)).thenThrow(new ComponentInitializationFailedException("Failed to initialize component")); MicroPipelineManager manager = new MicroPipelineManager("id", factory, executorService); try { manager.executePipeline(cfg); Assert.fail("Invalid input"); } catch(ComponentInitializationFailedException e) { } Mockito.verify(factory).instantiatePipeline(cfg, executorService); }
@Test public void testExecutePipeline_withConfigurationLeadingFactoryReturnNull() throws Exception { MicroPipelineConfiguration cfg = new MicroPipelineConfiguration(); cfg.setId("testExecutePipeline_withConfigurationLeadingFactoryReturnNull"); MicroPipelineFactory factory = Mockito.mock(MicroPipelineFactory.class); Mockito.when(factory.instantiatePipeline(cfg, executorService)).thenReturn(null); MicroPipelineManager manager = new MicroPipelineManager("id", factory, executorService); try { manager.executePipeline(cfg); Assert.fail("Invalid input"); } catch(PipelineInstantiationFailedException e) { } Mockito.verify(factory).instantiatePipeline(cfg, executorService); }