@Test public void test() throws Exception { TestIterator iterator = new TestIterator(5); final IteratorExecutor<Void> executor = new IteratorExecutor<>(iterator, 2, new ThreadFactoryBuilder().build()); log.info("Starting executor"); ExecutorService singleThread = Executors.newSingleThreadExecutor(); final Future<?> future = singleThread.submit(new Runnable() { @Override public void run() { try { executor.execute(); } catch (InterruptedException ie) { throw new RuntimeException(ie); } } }); log.info("Only two threads, so exactly two tasks retrieved"); verify(2, 0); Assert.assertFalse(future.isDone()); Assert.assertTrue(iterator.hasNext()); log.info("end one of the tasks"); iterator.endOneCallable(); log.info("three tasks retrieved, one completed"); verify(3, 1); Assert.assertFalse(future.isDone()); Assert.assertTrue(iterator.hasNext()); log.info("end two tasks"); iterator.endOneCallable(); iterator.endOneCallable(); log.info("5 (all) tasks retrieved, 3 completed"); verify(5, 3); Assert.assertFalse(future.isDone()); Assert.assertFalse(iterator.hasNext()); log.info("end two tasks"); iterator.endOneCallable(); iterator.endOneCallable(); log.info("all tasks completed, check future is done"); verify(5, 5); AssertWithBackoff.assertTrue(new Predicate<Void>() { @Override public boolean apply(Void input) { return future.isDone(); } }, 10000, "future done", log, 2, 1000); Assert.assertTrue(future.isDone()); log.info("done."); }
public List<Future<T>> execute() throws InterruptedException { List<Future<T>> futures = Lists.newArrayList(); try { if (this.executed) { throw new RuntimeException(String.format("This %s has already been executed.", IteratorExecutor.class.getSimpleName())); } int activeTasks = 0; while (this.iterator.hasNext()) { try { futures.add(this.completionService.submit(this.iterator.next())); activeTasks++; } catch (Exception exception) { futures.add(Futures.<T>immediateFailedFuture(exception)); } if (activeTasks == this.numThreads) { this.completionService.take(); activeTasks--; } } while (activeTasks > 0) { this.completionService.take(); activeTasks--; } } finally { ExecutorsUtils.shutdownExecutorService(this.executor, Optional.of(log), 10, TimeUnit.SECONDS); this.executed = true; } return futures; }
IteratorExecutor { public List<Future<T>> execute() throws InterruptedException { List<Future<T>> futures = Lists.newArrayList(); try { if (this.executed) { throw new RuntimeException(String.format("This %s has already been executed.", IteratorExecutor.class.getSimpleName())); } int activeTasks = 0; while (this.iterator.hasNext()) { try { futures.add(this.completionService.submit(this.iterator.next())); activeTasks++; } catch (Exception exception) { futures.add(Futures.<T>immediateFailedFuture(exception)); } if (activeTasks == this.numThreads) { this.completionService.take(); activeTasks--; } } while (activeTasks > 0) { this.completionService.take(); activeTasks--; } } finally { ExecutorsUtils.shutdownExecutorService(this.executor, Optional.of(log), 10, TimeUnit.SECONDS); this.executed = true; } return futures; } }
IteratorExecutor { public List<Future<T>> execute() throws InterruptedException { List<Future<T>> futures = Lists.newArrayList(); try { if (this.executed) { throw new RuntimeException(String.format("This %s has already been executed.", IteratorExecutor.class.getSimpleName())); } int activeTasks = 0; while (this.iterator.hasNext()) { try { futures.add(this.completionService.submit(this.iterator.next())); activeTasks++; } catch (Exception exception) { futures.add(Futures.<T>immediateFailedFuture(exception)); } if (activeTasks == this.numThreads) { this.completionService.take(); activeTasks--; } } while (activeTasks > 0) { this.completionService.take(); activeTasks--; } } finally { ExecutorsUtils.shutdownExecutorService(this.executor, Optional.of(log), 10, TimeUnit.SECONDS); this.executed = true; } return futures; } IteratorExecutor(Iterator<Callable<T>> runnableIterator, int numThreads, ThreadFactory threadFactory); }
IteratorExecutor { public List<Future<T>> execute() throws InterruptedException { List<Future<T>> futures = Lists.newArrayList(); try { if (this.executed) { throw new RuntimeException(String.format("This %s has already been executed.", IteratorExecutor.class.getSimpleName())); } int activeTasks = 0; while (this.iterator.hasNext()) { try { futures.add(this.completionService.submit(this.iterator.next())); activeTasks++; } catch (Exception exception) { futures.add(Futures.<T>immediateFailedFuture(exception)); } if (activeTasks == this.numThreads) { this.completionService.take(); activeTasks--; } } while (activeTasks > 0) { this.completionService.take(); activeTasks--; } } finally { ExecutorsUtils.shutdownExecutorService(this.executor, Optional.of(log), 10, TimeUnit.SECONDS); this.executed = true; } return futures; } IteratorExecutor(Iterator<Callable<T>> runnableIterator, int numThreads, ThreadFactory threadFactory); List<Future<T>> execute(); List<Either<T, ExecutionException>> executeAndGetResults(); static boolean verifyAllSuccessful(List<Either<T, ExecutionException>> results); static void logFailures(List<Either<T, ExecutionException>> results, Logger useLogger, int atMost); static void logAndThrowFailures(List<Either<T, ExecutionException>> results, Logger useLogger, int atMost); }
IteratorExecutor { public List<Future<T>> execute() throws InterruptedException { List<Future<T>> futures = Lists.newArrayList(); try { if (this.executed) { throw new RuntimeException(String.format("This %s has already been executed.", IteratorExecutor.class.getSimpleName())); } int activeTasks = 0; while (this.iterator.hasNext()) { try { futures.add(this.completionService.submit(this.iterator.next())); activeTasks++; } catch (Exception exception) { futures.add(Futures.<T>immediateFailedFuture(exception)); } if (activeTasks == this.numThreads) { this.completionService.take(); activeTasks--; } } while (activeTasks > 0) { this.completionService.take(); activeTasks--; } } finally { ExecutorsUtils.shutdownExecutorService(this.executor, Optional.of(log), 10, TimeUnit.SECONDS); this.executed = true; } return futures; } IteratorExecutor(Iterator<Callable<T>> runnableIterator, int numThreads, ThreadFactory threadFactory); List<Future<T>> execute(); List<Either<T, ExecutionException>> executeAndGetResults(); static boolean verifyAllSuccessful(List<Either<T, ExecutionException>> results); static void logFailures(List<Either<T, ExecutionException>> results, Logger useLogger, int atMost); static void logAndThrowFailures(List<Either<T, ExecutionException>> results, Logger useLogger, int atMost);  }