Streams { @NotNull public static <S, T> Stream<ResultWrapper<T>> parallelize( @NotNull Collection<S> source, @NotNull Function<S, Supplier<T>> mapper ) { List<CompletableFuture<ResultWrapper<T>>> results = source.stream() .map(s -> CompletableFuture.supplyAsync(new ResultSupplier<>(mapper.apply(s)), parallelExecutor)) .collect(ImmutableList.toImmutableList()); return collect(results); } private  Streams(); @NotNull static Stream<ResultWrapper<T>> parallelize(
            @NotNull Collection<S> source,
            @NotNull Function<S, Supplier<T>> mapper
    ); @NotNull static Stream<ResultWrapper<T>> parallelize(int n, @NotNull Supplier<T> supplier);  }
@Test public void parallelize() { List<Long> delays = Lists.newArrayList(1001L, 1002L); assertEquals((long) Streams.parallelize(delays, delay -> () -> { try { Thread.sleep(delay); } catch (InterruptedException ignored) { } return delay + delay; }) .filter(r -> !r.hasError()) .map(ResultWrapper::getResult) .reduce(0L, (a, b) -> a + b), 4006L); }
@Test public void parallelizeWithException() { List<Long> delays = Lists.newArrayList(1001L, 1002L); assertEquals((long) Streams.parallelize(delays, delay -> () -> { if (delay == 1002L) { throw new RuntimeException(); } try { Thread.sleep(delay); } catch (InterruptedException ignored) { } return delay + delay; }) .filter(r -> !r.hasError()) .map(ResultWrapper::getResult) .reduce(0L, (a, b) -> a + b), 2002L); }
@Test public void parallelizeSupplier() { assertEquals((long) Streams.parallelize(3, () -> 42L) .filter(r -> !r.hasError()) .map(ResultWrapper::getResult) .reduce(0L, (a, b) -> a + b), 126L); }