@Test public void testDoEventuallyCollectImmediate() throws Exception { final List<Callable<Stage<Object>>> callables = ImmutableList.of(c, c2); final List<Stage<Object>> futures = ImmutableList.of(f1, f2); doReturn(f1).when(c).call(); doThrow(e).when(c2).call(); doReturn(f2).when(underTest).failed(e); doReturn(future).when(underTest).streamCollect(futures, consumer, supplier); assertEquals(future, underTest.doEventuallyCollectImmediate(callables, consumer, supplier)); verify(c).call(); verify(c2).call(); verify(underTest).failed(e); verify(underTest).streamCollect(futures, consumer, supplier); }
<C, T> Stage<T> doEventuallyCollectImmediate( final Collection<? extends Callable<? extends Stage<? extends C>>> callables, final Consumer<? super C> consumer, Supplier<? extends T> supplier ) { final List<Stage<? extends C>> stages = new ArrayList<>(callables.size()); for (final Callable<? extends Stage<? extends C>> c : callables) { final Stage<? extends C> stage; try { stage = c.call(); } catch (final Exception e) { stages.add(this.failed(e)); continue; } stages.add(stage); } return streamCollect(stages, consumer, supplier); }
CoreAsync implements Async { <C, T> Stage<T> doEventuallyCollectImmediate( final Collection<? extends Callable<? extends Stage<? extends C>>> callables, final Consumer<? super C> consumer, Supplier<? extends T> supplier ) { final List<Stage<? extends C>> stages = new ArrayList<>(callables.size()); for (final Callable<? extends Stage<? extends C>> c : callables) { final Stage<? extends C> stage; try { stage = c.call(); } catch (final Exception e) { stages.add(this.failed(e)); continue; } stages.add(stage); } return streamCollect(stages, consumer, supplier); } }
CoreAsync implements Async { <C, T> Stage<T> doEventuallyCollectImmediate( final Collection<? extends Callable<? extends Stage<? extends C>>> callables, final Consumer<? super C> consumer, Supplier<? extends T> supplier ) { final List<Stage<? extends C>> stages = new ArrayList<>(callables.size()); for (final Callable<? extends Stage<? extends C>> c : callables) { final Stage<? extends C> stage; try { stage = c.call(); } catch (final Exception e) { stages.add(this.failed(e)); continue; } stages.add(stage); } return streamCollect(stages, consumer, supplier); }  }
CoreAsync implements Async { <C, T> Stage<T> doEventuallyCollectImmediate( final Collection<? extends Callable<? extends Stage<? extends C>>> callables, final Consumer<? super C> consumer, Supplier<? extends T> supplier ) { final List<Stage<? extends C>> stages = new ArrayList<>(callables.size()); for (final Callable<? extends Stage<? extends C>> c : callables) { final Stage<? extends C> stage; try { stage = c.call(); } catch (final Exception e) { stages.add(this.failed(e)); continue; } stages.add(stage); } return streamCollect(stages, consumer, supplier); }  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 { <C, T> Stage<T> doEventuallyCollectImmediate( final Collection<? extends Callable<? extends Stage<? extends C>>> callables, final Consumer<? super C> consumer, Supplier<? extends T> supplier ) { final List<Stage<? extends C>> stages = new ArrayList<>(callables.size()); for (final Callable<? extends Stage<? extends C>> c : callables) { final Stage<? extends C> stage; try { stage = c.call(); } catch (final Exception e) { stages.add(this.failed(e)); continue; } stages.add(stage); } return streamCollect(stages, consumer, supplier); }  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();  }