@Test public void testBindSignals() { final Collection<Stage<Object>> futures = ImmutableList.of(f1, f2); doAnswer(new Answer<Void>() { @Override public Void answer(InvocationOnMock invocation) throws Throwable { final Runnable cancelled = invocation.getArgumentAt(0, Runnable.class); cancelled.run(); return null; } }).when(future).whenCancelled(any(Runnable.class)); underTest.bindSignals(future, futures); verify(future).whenCancelled(any(Runnable.class)); verify(f1).cancel(); verify(f2).cancel(); }
void bindSignals( final Stage<?> target, final Collection<? extends Stage<?>> stages ) { target.whenCancelled(() -> { for (final Stage<?> f : stages) { f.cancel(); } }); }
CoreAsync implements Async { void bindSignals( final Stage<?> target, final Collection<? extends Stage<?>> stages ) { target.whenCancelled(() -> { for (final Stage<?> f : stages) { f.cancel(); } }); } }
CoreAsync implements Async { void bindSignals( final Stage<?> target, final Collection<? extends Stage<?>> stages ) { target.whenCancelled(() -> { for (final Stage<?> f : stages) { f.cancel(); } }); }  }
CoreAsync implements Async { void bindSignals( final Stage<?> target, final Collection<? extends Stage<?>> stages ) { target.whenCancelled(() -> { for (final Stage<?> f : stages) { f.cancel(); } }); }  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 { void bindSignals( final Stage<?> target, final Collection<? extends Stage<?>> stages ) { target.whenCancelled(() -> { for (final Stage<?> f : stages) { f.cancel(); } }); }  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();  }