FutureUtil { public static <T> List<T> gatherIO(List<? extends CompletableFuture<? extends T>> futures, CompletionStage<Void> timeout) throws IOException { timeout.thenRun(() -> futures.forEach(f -> f.completeExceptionally(new TimeoutException()))); var values = ImmutableList.<T>builder(); for (var future : futures) { try { values.add(future.get()); } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } catch (ExecutionException e) { final Throwable cause = e.getCause(); Throwables.propagateIfPossible(cause, IOException.class); if (cause instanceof TimeoutException) { throw new IOException(cause); } else { throw new RuntimeException(cause); } } } return values.build(); } private  FutureUtil(); static List<T> gatherIO(List<? extends CompletableFuture<? extends T>> futures,
                                     CompletionStage<Void> timeout);  }
@Test public void gatherShouldPropagateInterruptedAsIOException() throws Exception { var futures = List.of(completedFuture("foo"), future); Thread.currentThread().interrupt(); try { FutureUtil.gatherIO(futures, NO_TIMEOUT); fail(); } catch (IOException e) { assertThat(e.getCause(), instanceOf(InterruptedException.class)); } assertThat(Thread.currentThread().isInterrupted(), is(true)); exception.expect(InterruptedException.class); Thread.sleep(1000); }
@Test public void gatherIOShouldApplyTimeouts() throws IOException { var executor = Executors.newSingleThreadExecutor(); var futures = IntStream.range(0, 1000).boxed() .map(i -> CompletableFuture.supplyAsync(() -> { sleepMillis(10); return i; }, executor)) .collect(toList()); var start = System.nanoTime(); var timeout = CompletableFuture.runAsync(() -> {}, delayedExecutor(200, MILLISECONDS)); try { gatherIO(futures, timeout); fail(); } catch (IOException e) { assertThat(e.getCause(), instanceOf(TimeoutException.class)); } var end = System.nanoTime(); var elapsed = Duration.ofNanos(end - start); assertThat(elapsed.getSeconds(), is(lessThan(1L))); }
@Test public void gatherIOShouldReturnValues() throws IOException { var futures = List.of(completedFuture("foo"), completedFuture("bar")); assertThat(FutureUtil.gatherIO(futures, NO_TIMEOUT), contains("foo", "bar")); }
@Test public void gatherIOShouldPropagateIOException() throws IOException { var cause = new IOException("foo"); var futures = List.of(completedFuture("foo"), failedFuture(cause)); exception.expect(is(cause)); FutureUtil.gatherIO(futures, NO_TIMEOUT); }
@Test public void gatherIOShouldPropagateException() throws IOException { var cause = new Exception("foo"); var futures = List.of(completedFuture("foo"), failedFuture(cause)); exception.expect(RuntimeException.class); exception.expectCause(is(cause)); FutureUtil.gatherIO(futures, NO_TIMEOUT); }
@Test public void gatherIOShouldPropagateTimeoutAsIOException() throws IOException { var futures = List.of(completedFuture("foo"), future); exception.expect(IOException.class); exception.expectCause(instanceOf(TimeoutException.class)); FutureUtil.gatherIO(futures, TIMED_OUT); }