MicroPipelineFactory { protected StreamingMessageQueue initializeQueue(final StreamingMessageQueueConfiguration queueConfiguration) throws RequiredInputMissingException, QueueInitializationFailedException { if(queueConfiguration == null) throw new RequiredInputMissingException("Missing required queue configuration"); if(StringUtils.isBlank(queueConfiguration.getId())) throw new RequiredInputMissingException("Missing required queue identifier"); boolean inMemoryQueue = false; if(queueConfiguration.getProperties() != null && !queueConfiguration.getProperties().isEmpty()) { String queueType = StringUtils.lowerCase(StringUtils.trim(queueConfiguration.getProperties().getProperty(StreamingMessageQueue.CFG_QUEUE_TYPE))); inMemoryQueue = StringUtils.equalsIgnoreCase(queueType, InMemoryStreamingMessageQueue.CFG_QUEUE_TYPE); } if(inMemoryQueue) { try { StreamingMessageQueue queue = new InMemoryStreamingMessageQueue(); queue.setId(StringUtils.lowerCase(StringUtils.trim(queueConfiguration.getId()))); queue.initialize((queueConfiguration.getProperties() != null ? queueConfiguration.getProperties() : new Properties())); return queue; } catch(Exception e) { throw new QueueInitializationFailedException("Failed to initialize streaming message queue '"+queueConfiguration.getId()+"'. Error: " + e.getMessage()); } } try { StreamingMessageQueue queue = new DefaultStreamingMessageQueue(); queue.setId(StringUtils.lowerCase(StringUtils.trim(queueConfiguration.getId()))); queue.initialize((queueConfiguration.getProperties() != null ? queueConfiguration.getProperties() : new Properties())); return queue; } catch(Exception e) { throw new QueueInitializationFailedException("Failed to initialize streaming message queue '"+queueConfiguration.getId()+"'. Error: " + e.getMessage()); } } MicroPipelineFactory(final String processingNodeId, final ComponentRepository componentRepository); MicroPipeline instantiatePipeline(final MicroPipelineConfiguration cfg, final ExecutorService executorService);  }
@Test public void testInitializeQueue_withNullInput() throws QueueInitializationFailedException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeQueue(null); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeQueue_withEmptyInput() throws QueueInitializationFailedException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeQueue(new StreamingMessageQueueConfiguration()); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeQueue_withEmptyQueueId() throws QueueInitializationFailedException { try { StreamingMessageQueueConfiguration cfg = new StreamingMessageQueueConfiguration(); cfg.setProperties(new Properties()); cfg.setId(""); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeQueue(cfg); Assert.fail("Missing required input"); } catch(RequiredInputMissingException e) { } }
@Test public void testInitializeQueue_withNullQueueSettings() throws QueueInitializationFailedException, RequiredInputMissingException { StreamingMessageQueueConfiguration cfg = new StreamingMessageQueueConfiguration(); cfg.setProperties(null); cfg.setId("test-id"); new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeQueue(cfg); }
@Test public void testInitializeQueue_withValidSettingsForDefaultQueue() throws RequiredInputMissingException, QueueInitializationFailedException { Properties props = new Properties(); props.put(DefaultStreamingMessageQueue.CFG_CHRONICLE_QUEUE_DELETE_ON_EXIT, "true"); props.put(DefaultStreamingMessageQueue.CFG_CHRONICLE_QUEUE_PATH, System.getProperty("java.io.tmpdir")); StreamingMessageQueueConfiguration cfg = new StreamingMessageQueueConfiguration(); cfg.setProperties(props); cfg.setId("testInitializeQueue_withValidSettingsForDefaultQueue"); StreamingMessageQueue queue = new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).initializeQueue(cfg); Assert.assertNotNull("The queue instance must not be null");; Assert.assertEquals("The classes must be equal", DefaultStreamingMessageQueue.class, queue.getClass()); }