Futurity { public static <V> CompletableFuture<V> shift(Future<V> future) { return CommonFuturityWheel.getCommon().shift(future); } private  Futurity(); static CompletableFuture<V> shift(Future<V> future); static CompletableFuture<V> shiftWithPoll(Future<V> future, long pollPeriod, TimeUnit unit); static FuturityBuilder builder();  }
@Test public void returnCompletableFutureBack() { CompletableFuture origin = new CompletableFuture(); assertSame(origin, Futurity.shift(origin)); }
@Test public void trivial() throws InterruptedException { AtomicReference<Boolean> futureResult = new AtomicReference<>(); Future<Boolean> origin = new FutureWithSource<>(futureResult); CompletableFuture<Boolean> shift = Futurity.shift(origin); assertFalse(shift.isDone()); futureResult.set(Boolean.TRUE); await().pollInterval(1, TimeUnit.MILLISECONDS).pollDelay(0, TimeUnit.MILLISECONDS) .atMost(1, TimeUnit.SECONDS).until(shift::isDone); }
@Test(expected = IllegalStateException.class) public void passExceptionCorrectly() throws Throwable { AtomicReference<Throwable> exceptionResult = new AtomicReference<>(); Future<Boolean> origin = new FutureWithSource<>(null, exceptionResult); CompletableFuture<Boolean> shift = Futurity.shift(origin); assertFalse(shift.isDone()); Throwable ex = new IllegalStateException(); exceptionResult.set(ex); await().pollInterval(1, TimeUnit.MILLISECONDS).pollDelay(0, TimeUnit.MILLISECONDS) .atMost(1, TimeUnit.SECONDS).until(shift::isDone); try { shift.get(); } catch (ExecutionException e) { Throwable ex1 = e.getCause(); assertSame(ex1, ex); throw ex1; } fail(); }
@Test public void finalizingALotOfFutures() { ExecutorService executorService = Executors.newFixedThreadPool(50); try { List<CompletableFuture<?>> futures = new ArrayList<>(1000); for (int i = 0; i < 1000; i++) { futures.add(Futurity.shift(executorService.submit(new Callable<Object>() { private final Random rnd = new Random(); @Override public Object call() throws Exception { Thread.sleep(rnd.nextInt(5)); return new Object(); } }))); } CountDownLatch countDownLatch = new CountDownLatch(500); futures.forEach(future -> future.thenAccept(i -> countDownLatch.countDown())); await("Some futures are not finalized") .atMost(5, TimeUnit.SECONDS).until(() -> countDownLatch.getCount() == 0); } finally { executorService.shutdown(); } }
@Test public void sourceFutureHasBeenCancelled_correctlyFillCompletableFuture() throws InterruptedException { Future<Boolean> origin = new FutureWithSource<>(new AtomicReference<>()); CompletableFuture<Boolean> shifted = Futurity.shift(origin); assertFalse(shifted.isDone()); origin.cancel(false); await().pollInterval(1, TimeUnit.MILLISECONDS).pollDelay(0, TimeUnit.MILLISECONDS) .atMost(1, TimeUnit.SECONDS).until(shifted::isDone); assertTrue(shifted.isCancelled()); }