MicroPipelineFactory { protected MicroPipelineComponent initializeComponent(final MicroPipelineComponentConfiguration componentConfiguration, final Map<String, StreamingMessageQueue> queues) throws RequiredInputMissingException, ComponentInitializationFailedException { if(componentConfiguration == null) throw new RequiredInputMissingException("Missing required component configuration"); if(StringUtils.isBlank(componentConfiguration.getId())) throw new RequiredInputMissingException("Missing required component identifier"); if(componentConfiguration.getType() == null) throw new RequiredInputMissingException("Missing required component type"); if(StringUtils.isBlank(componentConfiguration.getName())) throw new RequiredInputMissingException("Missing required component name"); if(StringUtils.isBlank(componentConfiguration.getVersion())) throw new RequiredInputMissingException("Missing required component version"); if(componentConfiguration.getSettings() == null) throw new RequiredInputMissingException("Missing required component settings"); if(componentConfiguration.getType() == MicroPipelineComponentType.SOURCE) { if(StringUtils.isBlank(componentConfiguration.getToQueue())) throw new RequiredInputMissingException("Missing required queues to write content to"); if(!queues.containsKey(StringUtils.lowerCase(StringUtils.trim(componentConfiguration.getToQueue())))) throw new RequiredInputMissingException("Unknown destination queue '"+componentConfiguration.getToQueue()+"'"); } else if(componentConfiguration.getType() == MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR) { if(StringUtils.isBlank(componentConfiguration.getToQueue())) throw new RequiredInputMissingException("Missing required queues to write content to"); if(!queues.containsKey(StringUtils.lowerCase(StringUtils.trim(componentConfiguration.getToQueue())))) throw new RequiredInputMissingException("Unknown destination queue '"+componentConfiguration.getToQueue()+"'"); if(StringUtils.isBlank(componentConfiguration.getFromQueue())) throw new RequiredInputMissingException("Missing required queues to retrieve content from"); if(!queues.containsKey(StringUtils.lowerCase(StringUtils.trim(componentConfiguration.getFromQueue())))) throw new RequiredInputMissingException("Unknown source queue '"+componentConfiguration.getFromQueue()+"'"); } else if(componentConfiguration.getType() == MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR) { if(StringUtils.isBlank(componentConfiguration.getToQueue())) throw new RequiredInputMissingException("Missing required queues to write content to"); if(!queues.containsKey(StringUtils.lowerCase(StringUtils.trim(componentConfiguration.getToQueue())))) throw new RequiredInputMissingException("Unknown destination queue '"+componentConfiguration.getToQueue()+"'"); if(StringUtils.isBlank(componentConfiguration.getFromQueue())) throw new RequiredInputMissingException("Missing required queues to retrieve content from"); if(!queues.containsKey(StringUtils.lowerCase(StringUtils.trim(componentConfiguration.getFromQueue())))) throw new RequiredInputMissingException("Unknown source queue '"+componentConfiguration.getFromQueue()+"'"); if(StringUtils.isBlank(componentConfiguration.getSettings().getProperty(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME))) throw new RequiredInputMissingException("Missing required settings for wait strategy applied to delayed response operator"); } else if(componentConfiguration.getType() == MicroPipelineComponentType.EMITTER) { if(StringUtils.isBlank(componentConfiguration.getFromQueue())) throw new RequiredInputMissingException("Missing required queues to retrieve content from"); if(!queues.containsKey(StringUtils.lowerCase(StringUtils.trim(componentConfiguration.getFromQueue())))) throw new RequiredInputMissingException("Unknown source queue '"+componentConfiguration.getFromQueue()+"'"); } try { return this.componentRepository.newInstance(componentConfiguration.getId(), componentConfiguration.getName(), componentConfiguration.getVersion(), componentConfiguration.getSettings()); } catch(Exception e) { throw new ComponentInitializationFailedException("Failed to initialize component '"+componentConfiguration.getId()+"'. Error: " + e.getMessage(), e); } } MicroPipelineFactory(final String processingNodeId, final ComponentRepository componentRepository); MicroPipeline instantiatePipeline(final MicroPipelineConfiguration cfg, final ExecutorService executorService);  }
@Test public void testInitializeComponent_withNullInput() throws ComponentInitializationFailedException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(null, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_withEmptyInput() throws ComponentInitializationFailedException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(new MicroPipelineComponentConfiguration(), Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_withEmptyComponentType() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id");; cfg.setSettings(new Properties()); cfg.setName(""); cfg.setVersion("0.0.1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_withEmptyComponentClass() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id");; cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setName(""); cfg.setVersion("0.0.1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_withEmptyComponentVersion() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id");; cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setName("test-class"); cfg.setVersion(""); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_withEmptyComponentSettings() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id");; cfg.setSettings(null); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setName("test-name"); cfg.setVersion("0.0.1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_source_withEmptyToQueuesSettings() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setName("test name"); cfg.setVersion("0.0.1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_source_withReferenceToUnknownQueue() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("queue-1")).thenReturn(false); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); }
@Test public void testInitializeComponent_source_withValidConfigurationButExceptionThrownByRepository() throws Exception { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.SOURCE); cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); ComponentRepository repo = Mockito.mock(ComponentRepository.class); Mockito.when(repo.newInstance(cfg.getId(), cfg.getName(), cfg.getVersion(), cfg.getSettings())).thenThrow(new NullPointerException()); try { new MicroPipelineFactory("id", repo).initializeComponent(cfg, queues); Assert.fail("Repo access fails"); } catch(ComponentInitializationFailedException e) { } Mockito.verify(queues).containsKey("queue-1"); }
@Test public void testInitializeComponent_direct_response_operator_withEmptyToQueuesSettings() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue(""); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_direct_response_operator_withReferenceToUnknownToQueue() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("unknown-queue")).thenReturn(false); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); cfg.setVersion("0.0.1"); cfg.setToQueue("unknown-queue"); cfg.setFromQueue("queue-1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("unknown-queue"); Mockito.verify(queues, Mockito.never()).containsKey("queue-1"); }
@Test public void testInitializeComponent_direct_response_operator_withEmptyFromQueuesSettings() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("unknown-queue")).thenReturn(false); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); cfg.setFromQueue(""); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); Mockito.verify(queues, Mockito.never()).containsKey("unknown-queue"); }
@Test public void testInitializeComponent_direct_response_operator_withUnknownFromQueuesSettings() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("unknown-queue")).thenReturn(false); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); cfg.setFromQueue("unknown-queue"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); Mockito.verify(queues).containsKey("unknown-queue"); }
@Test public void testInitializeComponent_direct_response_operator_withValidSettings() throws ComponentInitializationFailedException, RequiredInputMissingException, UnknownComponentException, ComponentInstantiationFailedException { DirectResponseOperator directResponseOperator = Mockito.mock(DirectResponseOperator.class); Mockito.when(directResponseOperator.getId()).thenReturn("test-id"); ComponentRepository repo = Mockito.mock(ComponentRepository.class); Mockito.when(repo.newInstance("test-id", "test name", "0.0.1", new Properties())).thenReturn(directResponseOperator); @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); Mockito.when(queues.containsKey("queue-2")).thenReturn(true); MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); cfg.setFromQueue("queue-2"); MicroPipelineComponent operator = new MicroPipelineFactory("id", repo).initializeComponent(cfg, queues); Assert.assertNotNull("The operator must not be null", operator);; Assert.assertEquals("Values must be equal", "test-id", operator.getId()); Mockito.verify(queues).containsKey("queue-1"); Mockito.verify(queues, Mockito.never()).containsKey("unknown-queue"); }
@Test public void testInitializeComponent_delayed_response_operator_withEmptyToQueuesSettings() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); Mockito.when(queues.containsKey("queue-2")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue(""); cfg.setFromQueue("queue-2"); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME, "test-strategy"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues, Mockito.never()).containsKey("queue-1"); Mockito.verify(queues, Mockito.never()).containsKey("queue-2"); }
@Test public void testInitializeComponent_delayed_response_operator_withReferenceToUnknownToQueue() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("unknown-queue")).thenReturn(false); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); cfg.setVersion("0.0.1"); cfg.setToQueue("unknown-queue"); cfg.setFromQueue("queue-1"); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME, "test-strategy"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("unknown-queue"); Mockito.verify(queues, Mockito.never()).containsKey("queue-1"); }
@Test public void testInitializeComponent_delayed_response_operator_withEmptyFromQueuesSettings() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("unknown-queue")).thenReturn(false); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); cfg.setFromQueue(""); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME, "test-strategy"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); Mockito.verify(queues, Mockito.never()).containsKey("unknown-queue"); }
@Test public void testInitializeComponent_delayed_response_operator_withUnknownFromQueuesSettings() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("unknown-queue")).thenReturn(false); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DIRECT_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); cfg.setFromQueue("unknown-queue"); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME, "test-strategy"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); Mockito.verify(queues).containsKey("unknown-queue"); }
@Test public void testInitializeComponent_direct_response_operator_withEmptyResponseWaitStrategy() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("queue-2")).thenReturn(true); Mockito.when(queues.containsKey("queue-1")).thenReturn(true); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.DELAYED_RESPONSE_OPERATOR); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setToQueue("queue-1"); cfg.setFromQueue("queue-2"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); Mockito.verify(queues).containsKey("queue-2"); }
@Test public void testInitializeComponent_emitter_withEmptyFromQueuesSettings() throws ComponentInitializationFailedException { try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.EMITTER); cfg.setName("test name"); cfg.setVersion("0.0.1"); cfg.setFromQueue(""); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, Collections.<String, StreamingMessageQueue>emptyMap()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeComponent_emitter_withReferenceToUnknownFromQueue() throws ComponentInitializationFailedException { @SuppressWarnings("unchecked") Map<String, StreamingMessageQueue> queues = Mockito.mock(HashMap.class); Mockito.when(queues.containsKey("queue-1")).thenReturn(false); try { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setId("test-id"); cfg.setSettings(new Properties()); cfg.setType(MicroPipelineComponentType.EMITTER); cfg.setName("testInitializeComponent_source_withReferenceToUnknownQueue"); cfg.setVersion("0.0.1"); cfg.setFromQueue("queue-1"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeComponent(cfg, queues); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } Mockito.verify(queues).containsKey("queue-1"); }