AggregationStrategy { public void execute(Aggregate aggregate) { Executor executor = Executors.newCachedThreadPool(); CompletionService<OutboundOrchestratable> completionService = new ExecutorCompletionService<>(executor); Collection<OutboundOrchestratable> aggregationRequests = aggregate.getAggregateRequests(); int size = aggregationRequests.size(); for (OutboundOrchestratable orchestrationContext : aggregationRequests) { completionService.submit(new CallableAggregation(orchestrationContext)); } int i = 0; for (; i < size; i++) { try { aggregate.aggregate(completionService.take().get()); LOG.trace("got response " + (i + 1) + " of " + size); } catch (InterruptedException e) { LOG.error("Failure during aggregation, aborting: " + e.getLocalizedMessage(), e); break; } catch (ExecutionException e) { LOG.error("Could not aggregrate response #" + (i + 1) + ": " + e.getLocalizedMessage(), e); continue; } } if (i < size) { LOG.info((size - i) + " responses failed."); } }  void execute(Aggregate aggregate);  }
@Test public void testAggregation() { AggregationStrategy strategy = new AggregationStrategy(); Aggregate aggregate = mock(Aggregate.class); OutboundOrchestratable message = mock(OutboundOrchestratable.class); OutboundDelegate delegate = mock(OutboundDelegate.class); ArrayList<OutboundOrchestratable> values = new ArrayList<>(); values.add(message); when(aggregate.getAggregateRequests()).thenReturn(values); when(message.getDelegate()).thenReturn(delegate); strategy.execute(aggregate); verify(delegate).process(eq(message)); verify(aggregate).aggregate(not(eq(message))); }