@Test public void usesDefaultRoundRobin() { List<Processor> processors = new LinkedList<Processor>(); processors.add(Mockito.mock(Processor.class)); processors.add(Mockito.mock(Processor.class)); DynamicLoadBalancerConfiguration config = new DynamicLoadBalancerConfiguration(); config.setRouteStatisticsCollector(Mockito.mock(RouteStatisticsCollector.class)); config.setDeterministicCollectorStrategy(new EveryXDeterministicCollectorStrategy(10, 10)); config.setRouteStatsSelectorStrategy(new MeanProcessingTimeProcessorSelectorStrategy()); DynamicRoundRobinLoadBalancer loadBalancer = new DynamicRoundRobinLoadBalancer(config); Processor answer = loadBalancer.chooseProcessor(processors, createExchange()); Assert.assertNotNull(answer); Assert.assertEquals(processors.get(0), 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();  }