MicroPipelineFactory { protected DelayedResponseOperatorWaitStrategy getResponseWaitStrategy(final MicroPipelineComponentConfiguration delayedResponseOperatorCfg) throws RequiredInputMissingException, UnknownWaitStrategyException { if(delayedResponseOperatorCfg == null) throw new RequiredInputMissingException("Missing required delayed response operator configuration"); if(delayedResponseOperatorCfg.getSettings() == null) throw new RequiredInputMissingException("Missing required delayed response operator settings"); String strategyName = StringUtils.lowerCase(StringUtils.trim(delayedResponseOperatorCfg.getSettings().getProperty(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME))); if(StringUtils.isBlank(strategyName)) throw new RequiredInputMissingException("Missing required strategy name expected as part of operator settings ('"+DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME+"')"); if(logger.isDebugEnabled()) logger.debug("Settings provided for strategy '"+strategyName+"'"); Properties strategyProperties = new Properties(); for(Enumeration<Object> keyEnumerator = delayedResponseOperatorCfg.getSettings().keys(); keyEnumerator.hasMoreElements();) { String key = (String)keyEnumerator.nextElement(); if(StringUtils.startsWith(key, DelayedResponseOperator.CFG_WAIT_STRATEGY_SETTINGS_PREFIX)) { String waitStrategyCfgKey = StringUtils.substring(key, StringUtils.lastIndexOf(key, ".") + 1); if(StringUtils.isNoneBlank(waitStrategyCfgKey)) { String waitStrategyCfgValue = delayedResponseOperatorCfg.getSettings().getProperty(key); strategyProperties.put(waitStrategyCfgKey, waitStrategyCfgValue); if(logger.isDebugEnabled()) logger.debug("\t" + waitStrategyCfgKey + ": " + waitStrategyCfgValue); } } } if(StringUtils.equalsIgnoreCase(strategyName, MessageCountResponseWaitStrategy.WAIT_STRATEGY_NAME)) { MessageCountResponseWaitStrategy strategy = new MessageCountResponseWaitStrategy(); strategy.initialize(strategyProperties); return strategy; } else if(StringUtils.equalsIgnoreCase(strategyName, TimerBasedResponseWaitStrategy.WAIT_STRATEGY_NAME)) { TimerBasedResponseWaitStrategy strategy = new TimerBasedResponseWaitStrategy(); strategy.initialize(strategyProperties); return strategy; } else if(StringUtils.equalsIgnoreCase(strategyName, OperatorTriggeredWaitStrategy.WAIT_STRATEGY_NAME)) { OperatorTriggeredWaitStrategy strategy = new OperatorTriggeredWaitStrategy(); strategy.initialize(strategyProperties); return strategy; } throw new UnknownWaitStrategyException("Unknown wait strategy '"+strategyName+"'"); } MicroPipelineFactory(final String processingNodeId, final ComponentRepository componentRepository); MicroPipeline instantiatePipeline(final MicroPipelineConfiguration cfg, final ExecutorService executorService);  }
@Test public void testGetResponseWaitStrategy_withEmptyStrategyName() throws UnknownWaitStrategyException{ try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).getResponseWaitStrategy(new MicroPipelineComponentConfiguration()); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testGetResponseWaitStrategy_withInvalidStrategyName() throws RequiredInputMissingException { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME, "test-strategy"); try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).getResponseWaitStrategy(cfg); Assert.fail("Unknown wait strategy"); } catch(UnknownWaitStrategyException e) { } }
@Test public void testGetResponseWaitStrategy_withValidStrategyAndSettings() throws RequiredInputMissingException, UnknownWaitStrategyException { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_NAME, MessageCountResponseWaitStrategy.WAIT_STRATEGY_NAME); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_SETTINGS_PREFIX + "key-2", "value-2"); cfg.getSettings().put(DelayedResponseOperator.CFG_WAIT_STRATEGY_SETTINGS_PREFIX + "key-1", "value-1"); DelayedResponseOperatorWaitStrategy strategy = new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).getResponseWaitStrategy(cfg); Assert.assertNotNull("The strategy must not be null", strategy); Assert.assertEquals("Types must be equal", MessageCountResponseWaitStrategy.class, strategy.getClass()); }
@Test public void testGetResponseWaitStrategy_withNullInput() throws UnknownWaitStrategyException { try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).getResponseWaitStrategy(null); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }
@Test public void testGetResponseWaitStrategy_withEmptySettings() throws UnknownWaitStrategyException { MicroPipelineComponentConfiguration cfg = new MicroPipelineComponentConfiguration(); cfg.setSettings(null); try { new MicroPipelineFactory("id", Mockito.mock(ComponentRepository.class)).getResponseWaitStrategy(cfg); Assert.fail("Invalid input"); } catch(RequiredInputMissingException e) { } }