@Test public void handlesTwoProcessors() { ProcessorHolder processorHolder1 = new ProcessorHolder(); processorHolder1.setProcessor(Mockito.mock(Processor.class)); ProcessorHolder processorHolder2 = new ProcessorHolder(); processorHolder2.setProcessor(Mockito.mock(Processor.class)); RouteStatistics stat1 = new RouteStatistics(); stat1.setProcessorHolder(processorHolder1); stat1.setMeanProcessingTime(12345L); RouteStatistics stat2 = new RouteStatistics(); stat2.setProcessorHolder(processorHolder2); stat2.setMeanProcessingTime(54321L); List<RouteStatistics> stats = new ArrayList<RouteStatistics>(); stats.add(stat1); stats.add(stat2); List<Processor> processors = new LinkedList<Processor>(); processors.add(processorHolder1.getProcessor()); processors.add(processorHolder2.getProcessor()); RouteStatisticsCollector routeStatisticsCollectorMocked = Mockito.mock(RouteStatisticsCollector.class); Mockito.when(routeStatisticsCollectorMocked.query(processors, createExchange())).thenReturn(stats); DynamicLoadBalancerConfiguration config = new DynamicLoadBalancerConfiguration(); config.setRouteStatisticsCollector(routeStatisticsCollectorMocked); config.setDeterministicCollectorStrategy(new EveryXDeterministicCollectorStrategy(0, 0)); config.setRouteStatsSelectorStrategy(new MeanProcessingTimeProcessorSelectorStrategy()); DynamicRoundRobinLoadBalancer loadBalancer = new DynamicRoundRobinLoadBalancer(config); Processor answer = loadBalancer.chooseProcessor(processors, createExchange()); Assert.assertNotNull(answer); Assert.assertEquals(processorHolder1.getProcessor(), answer); }
@Override protected synchronized Processor chooseProcessor(List<Processor> processors, Exchange exchange) { Processor answer = null; Processor defaultSelection = super.chooseProcessor(processors, exchange); DeterministicCollectorStrategy deterministicCollectorStrategy = config.getDeterministicCollectorStrategy(); if (deterministicCollectorStrategy.shouldCollect()) { RouteStatisticsCollector routeStatisticsCollector = config.getRouteStatisticsCollector(); List<RouteStatistics> stats = routeStatisticsCollector.query(processors, exchange); if (stats.size() > 0) { ProcessorSelectorStrategy selectorStrategy = config.getRouteStatsSelectorStrategy(); answer = selectorStrategy.getProcessor(stats); LOG.info("About to update processor to '{}'", answer); } } if (answer == null) { answer = defaultSelection; } return answer; }
DynamicRoundRobinLoadBalancer extends RoundRobinLoadBalancer { @Override protected synchronized Processor chooseProcessor(List<Processor> processors, Exchange exchange) { Processor answer = null; Processor defaultSelection = super.chooseProcessor(processors, exchange); DeterministicCollectorStrategy deterministicCollectorStrategy = config.getDeterministicCollectorStrategy(); if (deterministicCollectorStrategy.shouldCollect()) { RouteStatisticsCollector routeStatisticsCollector = config.getRouteStatisticsCollector(); List<RouteStatistics> stats = routeStatisticsCollector.query(processors, exchange); if (stats.size() > 0) { ProcessorSelectorStrategy selectorStrategy = config.getRouteStatsSelectorStrategy(); answer = selectorStrategy.getProcessor(stats); LOG.info("About to update processor to '{}'", answer); } } if (answer == null) { answer = defaultSelection; } return answer; } }
DynamicRoundRobinLoadBalancer extends RoundRobinLoadBalancer { @Override protected synchronized Processor chooseProcessor(List<Processor> processors, Exchange exchange) { Processor answer = null; Processor defaultSelection = super.chooseProcessor(processors, exchange); DeterministicCollectorStrategy deterministicCollectorStrategy = config.getDeterministicCollectorStrategy(); if (deterministicCollectorStrategy.shouldCollect()) { RouteStatisticsCollector routeStatisticsCollector = config.getRouteStatisticsCollector(); List<RouteStatistics> stats = routeStatisticsCollector.query(processors, exchange); if (stats.size() > 0) { ProcessorSelectorStrategy selectorStrategy = config.getRouteStatsSelectorStrategy(); answer = selectorStrategy.getProcessor(stats); LOG.info("About to update processor to '{}'", answer); } } if (answer == null) { answer = defaultSelection; } return answer; } DynamicRoundRobinLoadBalancer(DynamicLoadBalancerConfiguration config); }
DynamicRoundRobinLoadBalancer extends RoundRobinLoadBalancer { @Override protected synchronized Processor chooseProcessor(List<Processor> processors, Exchange exchange) { Processor answer = null; Processor defaultSelection = super.chooseProcessor(processors, exchange); DeterministicCollectorStrategy deterministicCollectorStrategy = config.getDeterministicCollectorStrategy(); if (deterministicCollectorStrategy.shouldCollect()) { RouteStatisticsCollector routeStatisticsCollector = config.getRouteStatisticsCollector(); List<RouteStatistics> stats = routeStatisticsCollector.query(processors, exchange); if (stats.size() > 0) { ProcessorSelectorStrategy selectorStrategy = config.getRouteStatsSelectorStrategy(); answer = selectorStrategy.getProcessor(stats); LOG.info("About to update processor to '{}'", answer); } } if (answer == null) { answer = defaultSelection; } return answer; } DynamicRoundRobinLoadBalancer(DynamicLoadBalancerConfiguration config); @Override String toString(); }
DynamicRoundRobinLoadBalancer extends RoundRobinLoadBalancer { @Override protected synchronized Processor chooseProcessor(List<Processor> processors, Exchange exchange) { Processor answer = null; Processor defaultSelection = super.chooseProcessor(processors, exchange); DeterministicCollectorStrategy deterministicCollectorStrategy = config.getDeterministicCollectorStrategy(); if (deterministicCollectorStrategy.shouldCollect()) { RouteStatisticsCollector routeStatisticsCollector = config.getRouteStatisticsCollector(); List<RouteStatistics> stats = routeStatisticsCollector.query(processors, exchange); if (stats.size() > 0) { ProcessorSelectorStrategy selectorStrategy = config.getRouteStatsSelectorStrategy(); answer = selectorStrategy.getProcessor(stats); LOG.info("About to update processor to '{}'", answer); } } if (answer == null) { answer = defaultSelection; } return answer; } DynamicRoundRobinLoadBalancer(DynamicLoadBalancerConfiguration config); @Override String toString();  }