MoreAsyncUtil { @API(API.Status.MAINTAINED) @Nonnull public static CompletableFuture<Void> delayedFuture(long delay, @Nonnull TimeUnit unit) { if (delay <= 0) { return AsyncUtil.DONE; } CompletableFuture<Void> future = new CompletableFuture<>(); scheduledThreadPoolExecutor.schedule(() -> future.complete(null), delay, unit); return future; } private  MoreAsyncUtil(); @Nonnull static AsyncIterable<T> limitIterable(@Nonnull final AsyncIterable<T> iterable,
                                                     final int limit); @Nonnull static AsyncIterable<T> filterIterable(@Nonnull final AsyncIterable<T> iterable,
                                                      @Nonnull final Function<T,Boolean> filter); @Nonnull static AsyncIterable<T> filterIterable(@Nonnull Executor executor,
                                                      @Nonnull final AsyncIterable<T> iterable,
                                                      @Nonnull final Function<T,Boolean> filter); @Nonnull static AsyncIterable<T> dedupIterable(@Nonnull final AsyncIterable<T> iterable); @Nonnull static AsyncIterable<T> dedupIterable(@Nonnull Executor executor,
                                                     @Nonnull final AsyncIterable<T> iterable); @Nonnull @SuppressWarnings("unchecked") // parameterized vararg static AsyncIterable<T> concatIterables(@Nonnull final AsyncIterable<T>... iterables); @Nonnull @SuppressWarnings("unchecked") // parameterized vararg static AsyncIterable<T> concatIterables(@Nonnull Executor executor, @Nonnull final AsyncIterable<T>... iterables); @Nonnull static AsyncIterable<T2> mapConcatIterable(@Nonnull final AsyncIterable<T1> iterable,
                                                              @Nonnull final Function<T1,AsyncIterable<T2>> func,
                                                              final int pipelineSize); @Nonnull static AsyncIterable<T2> mapConcatIterable(@Nonnull Executor executor,
                                                              @Nonnull final AsyncIterable<T1> iterable,
                                                              @Nonnull final Function<T1,AsyncIterable<T2>> func,
                                                              final int pipelineSize); @Nonnull static AsyncIterable<T> filterIterablePipelined(@Nonnull AsyncIterable<T> iterable,
                                                               @Nonnull final Function<T,CompletableFuture<Boolean>> filter,
                                                               int pipelineSize); @Nonnull static AsyncIterable<T> filterIterablePipelined(@Nonnull Executor executor,
                                                               @Nonnull AsyncIterable<T> iterable,
                                                               @Nonnull final Function<T,CompletableFuture<Boolean>> filter,
                                                               int pipelineSize); @Nonnull static AsyncIterable<T2> mapIterablePipelined(@Nonnull AsyncIterable<T1> iterable,
                                                                 @Nonnull final Function<T1,CompletableFuture<T2>> func,
                                                                 int pipelineSize); @Nullable static CompletableFuture<U> reduce(@Nonnull AsyncIterator<T> iterator, U identity,
                                                    BiFunction<U, ? super T, U> accumulator); @Nullable static CompletableFuture<U> reduce(@Nonnull Executor executor,
                                                    @Nonnull AsyncIterator<T> iterator, U identity,
                                                    BiFunction<U, ? super T, U> accumulator); @API(API.Status.MAINTAINED) static boolean isCompletedNormally(@Nonnull CompletableFuture<?> future); @API(API.Status.MAINTAINED) @Nonnull static CompletableFuture<Void> delayedFuture(long delay, @Nonnull TimeUnit unit); @API(API.Status.EXPERIMENTAL) static CompletableFuture<T> getWithDeadline(long deadlineTimeMillis,
                                                           @Nonnull Supplier<CompletableFuture<T>> supplier); @API(API.Status.MAINTAINED) static void closeIterator(@Nonnull Iterator<?> iterator); static CompletableFuture<V> composeWhenComplete(
            @Nonnull CompletableFuture<V> future,
            @Nonnull BiFunction<V,Throwable,CompletableFuture<Void>> handler,
            @Nullable Function<Throwable,RuntimeException> exceptionMapper); static CompletableFuture<T> composeWhenCompleteAndHandle(
            @Nonnull CompletableFuture<V> future,
            @Nonnull BiFunction<V,Throwable,? extends CompletableFuture<T>> handler,
            @Nullable Function<Throwable,RuntimeException> exceptionMapper);  }
@Test public void delaySimple() { long start = System.currentTimeMillis(); CompletableFuture<Void> delayed = MoreAsyncUtil.delayedFuture(30, TimeUnit.MILLISECONDS); delayed.join(); long end = System.currentTimeMillis(); assertTrue(end - start >= 30, "Delay was not long enough"); }
@Test public void manyParallelDelay() { long start = System.currentTimeMillis(); CompletableFuture<?>[] futures = new CompletableFuture<?>[1000]; for (int i = 0; i < futures.length; i++) { futures[i] = MoreAsyncUtil.delayedFuture(i * 100, TimeUnit.MICROSECONDS); } CompletableFuture.allOf(futures).join(); long end = System.currentTimeMillis(); assertTrue(end - start >= 100, "Delay was not long enough"); }