@Test public void testCallFutureDone() throws Exception { whenExecutorSubmitSetup(); when(completable.isDone()).thenReturn(true); underTest.doCall(callable, executor, completable); verifyCall(0, 0, 0, 0); }
public <C> Stage<C> doCall( final Callable<? extends C> callable, final ExecutorService executor, final Completable<C> stage ) { final Runnable runnable = () -> { if (stage.isDone()) { return; } final C result; try { result = callable.call(); } catch (final Exception error) { stage.fail(error); return; } stage.complete(result); }; final Future<?> task; try { task = executor.submit(runnable); } catch (final Exception e) { stage.fail(e); return stage; } stage.whenCancelled(() -> { task.cancel(false); }); return stage; }
CoreAsync implements Async { public <C> Stage<C> doCall( final Callable<? extends C> callable, final ExecutorService executor, final Completable<C> stage ) { final Runnable runnable = () -> { if (stage.isDone()) { return; } final C result; try { result = callable.call(); } catch (final Exception error) { stage.fail(error); return; } stage.complete(result); }; final Future<?> task; try { task = executor.submit(runnable); } catch (final Exception e) { stage.fail(e); return stage; } stage.whenCancelled(() -> { task.cancel(false); }); return stage; } }
CoreAsync implements Async { public <C> Stage<C> doCall( final Callable<? extends C> callable, final ExecutorService executor, final Completable<C> stage ) { final Runnable runnable = () -> { if (stage.isDone()) { return; } final C result; try { result = callable.call(); } catch (final Exception error) { stage.fail(error); return; } stage.complete(result); }; final Future<?> task; try { task = executor.submit(runnable); } catch (final Exception e) { stage.fail(e); return stage; } stage.whenCancelled(() -> { task.cancel(false); }); return stage; }  }
CoreAsync implements Async { public <C> Stage<C> doCall( final Callable<? extends C> callable, final ExecutorService executor, final Completable<C> stage ) { final Runnable runnable = () -> { if (stage.isDone()) { return; } final C result; try { result = callable.call(); } catch (final Exception error) { stage.fail(error); return; } stage.complete(result); }; final Future<?> task; try { task = executor.submit(runnable); } catch (final Exception e) { stage.fail(e); return stage; } stage.whenCancelled(() -> { task.cancel(false); }); return stage; }  ExecutorService executor(); Caller caller(); @Override Stage<C> call(final Callable<? extends C> callable); @Override Stage<C> call(
      final Callable<? extends C> callable, final ExecutorService executor
  ); Stage<C> doCall(
      final Callable<? extends C> callable, final ExecutorService executor,
      final Completable<C> stage
  ); @Override Completable<T> completable(); @Override Stage<Void> completed(); @Override Stage<T> completed(T value); @Override Stage<T> failed(Throwable e); @Override Stage<T> cancelled(); @SuppressWarnings("unchecked") @Override Stage<Collection<T>> collect(
      final Collection<? extends Stage<? extends T>> stages
  ); @Override Stage<T> collect(
      final Collection<? extends Stage<? extends C>> stages,
      final Function<? super Collection<C>, ? extends T> collector
  ); @Override Stage<U> streamCollect(
      final Collection<? extends Stage<? extends T>> stages, final Consumer<? super T> consumer,
      final Supplier<? extends U> supplier
  ); @Override Stage<T> eventuallyCollect(
      final Collection<? extends Callable<? extends Stage<? extends C>>> callables,
      final Consumer<? super C> consumer, Supplier<? extends T> supplier, int parallelism
  ); @Override Stage<Void> collectAndDiscard(
      Collection<? extends Stage<?>> stages
  ); @Override Managed<C> managed(
      Supplier<? extends Stage<C>> setup, Function<? super C, ? extends Stage<Void>> teardown
  ); @Override ReloadableManaged<C> reloadableManaged(
      final Supplier<? extends Stage<C>> setup,
      final Function<? super C, ? extends Stage<Void>> teardown
  ); @Override Stage<RetryResult<T>> retryUntilCompleted(
      final Callable<? extends Stage<T>> callable, final RetryPolicy policy
  ); @Override Stage<RetryResult<T>> retryUntilCompleted(
      final Callable<? extends Stage<T>> callable, final RetryPolicy policy,
      final ClockSource clockSource
  ); static Builder builder(); }
CoreAsync implements Async { public <C> Stage<C> doCall( final Callable<? extends C> callable, final ExecutorService executor, final Completable<C> stage ) { final Runnable runnable = () -> { if (stage.isDone()) { return; } final C result; try { result = callable.call(); } catch (final Exception error) { stage.fail(error); return; } stage.complete(result); }; final Future<?> task; try { task = executor.submit(runnable); } catch (final Exception e) { stage.fail(e); return stage; } stage.whenCancelled(() -> { task.cancel(false); }); return stage; }  ExecutorService executor(); Caller caller(); @Override Stage<C> call(final Callable<? extends C> callable); @Override Stage<C> call(
      final Callable<? extends C> callable, final ExecutorService executor
  ); Stage<C> doCall(
      final Callable<? extends C> callable, final ExecutorService executor,
      final Completable<C> stage
  ); @Override Completable<T> completable(); @Override Stage<Void> completed(); @Override Stage<T> completed(T value); @Override Stage<T> failed(Throwable e); @Override Stage<T> cancelled(); @SuppressWarnings("unchecked") @Override Stage<Collection<T>> collect(
      final Collection<? extends Stage<? extends T>> stages
  ); @Override Stage<T> collect(
      final Collection<? extends Stage<? extends C>> stages,
      final Function<? super Collection<C>, ? extends T> collector
  ); @Override Stage<U> streamCollect(
      final Collection<? extends Stage<? extends T>> stages, final Consumer<? super T> consumer,
      final Supplier<? extends U> supplier
  ); @Override Stage<T> eventuallyCollect(
      final Collection<? extends Callable<? extends Stage<? extends C>>> callables,
      final Consumer<? super C> consumer, Supplier<? extends T> supplier, int parallelism
  ); @Override Stage<Void> collectAndDiscard(
      Collection<? extends Stage<?>> stages
  ); @Override Managed<C> managed(
      Supplier<? extends Stage<C>> setup, Function<? super C, ? extends Stage<Void>> teardown
  ); @Override ReloadableManaged<C> reloadableManaged(
      final Supplier<? extends Stage<C>> setup,
      final Function<? super C, ? extends Stage<Void>> teardown
  ); @Override Stage<RetryResult<T>> retryUntilCompleted(
      final Callable<? extends Stage<T>> callable, final RetryPolicy policy
  ); @Override Stage<RetryResult<T>> retryUntilCompleted(
      final Callable<? extends Stage<T>> callable, final RetryPolicy policy,
      final ClockSource clockSource
  ); static Builder builder();  }