@Test public void testAcceptMock() { FastFuture cf = Mockito.mock(FastFuture.class); BDDMockito.given(cf.isDone()).willReturn(true); for(int i=0;i<1000;i++){ collector.accept(cf); } Mockito.verify(cf, Mockito.atLeastOnce()).isDone(); }
@Override public void accept(final FastFuture<T> t) { active.add(t); if (active.size() > maxActive.getMaxActive()) { while (active.size() > maxActive.getReduceTo()) { final List<FastFuture> toRemove = active.stream() .filter(cf -> cf.isDone()) .peek(this::handleExceptions) .collect(Collectors.toList()); active.removeAll(toRemove); if (active.size() > maxActive.getReduceTo()) { final CompletableFuture promise = new CompletableFuture(); FastFuture.xOf(active.size() - maxActive.getReduceTo(), () -> promise.complete(true), active.toArray(new FastFuture[0])); promise.join(); } } } }
EmptyCollector implements LazyResultConsumer<T> { @Override public void accept(final FastFuture<T> t) { active.add(t); if (active.size() > maxActive.getMaxActive()) { while (active.size() > maxActive.getReduceTo()) { final List<FastFuture> toRemove = active.stream() .filter(cf -> cf.isDone()) .peek(this::handleExceptions) .collect(Collectors.toList()); active.removeAll(toRemove); if (active.size() > maxActive.getReduceTo()) { final CompletableFuture promise = new CompletableFuture(); FastFuture.xOf(active.size() - maxActive.getReduceTo(), () -> promise.complete(true), active.toArray(new FastFuture[0])); promise.join(); } } } } }
EmptyCollector implements LazyResultConsumer<T> { @Override public void accept(final FastFuture<T> t) { active.add(t); if (active.size() > maxActive.getMaxActive()) { while (active.size() > maxActive.getReduceTo()) { final List<FastFuture> toRemove = active.stream() .filter(cf -> cf.isDone()) .peek(this::handleExceptions) .collect(Collectors.toList()); active.removeAll(toRemove); if (active.size() > maxActive.getReduceTo()) { final CompletableFuture promise = new CompletableFuture(); FastFuture.xOf(active.size() - maxActive.getReduceTo(), () -> promise.complete(true), active.toArray(new FastFuture[0])); promise.join(); } } } } EmptyCollector(); }
EmptyCollector implements LazyResultConsumer<T> { @Override public void accept(final FastFuture<T> t) { active.add(t); if (active.size() > maxActive.getMaxActive()) { while (active.size() > maxActive.getReduceTo()) { final List<FastFuture> toRemove = active.stream() .filter(cf -> cf.isDone()) .peek(this::handleExceptions) .collect(Collectors.toList()); active.removeAll(toRemove); if (active.size() > maxActive.getReduceTo()) { final CompletableFuture promise = new CompletableFuture(); FastFuture.xOf(active.size() - maxActive.getReduceTo(), () -> promise.complete(true), active.toArray(new FastFuture[0])); promise.join(); } } } } EmptyCollector(); @Override void accept(final FastFuture<T> t); void add(final FastFuture<T> t); @Override EmptyCollector<T> withResults(final Collection<FastFuture<T>> t); @Override void block(final Function<FastFuture<T>, T> safeJoin); void afterResults(Runnable r); @Override Collection<FastFuture<T>> getResults(); @Override Collection<FastFuture<T>> getAllResults(); boolean hasCapacity(final int i); }
EmptyCollector implements LazyResultConsumer<T> { @Override public void accept(final FastFuture<T> t) { active.add(t); if (active.size() > maxActive.getMaxActive()) { while (active.size() > maxActive.getReduceTo()) { final List<FastFuture> toRemove = active.stream() .filter(cf -> cf.isDone()) .peek(this::handleExceptions) .collect(Collectors.toList()); active.removeAll(toRemove); if (active.size() > maxActive.getReduceTo()) { final CompletableFuture promise = new CompletableFuture(); FastFuture.xOf(active.size() - maxActive.getReduceTo(), () -> promise.complete(true), active.toArray(new FastFuture[0])); promise.join(); } } } } EmptyCollector(); @Override void accept(final FastFuture<T> t); void add(final FastFuture<T> t); @Override EmptyCollector<T> withResults(final Collection<FastFuture<T>> t); @Override void block(final Function<FastFuture<T>, T> safeJoin); void afterResults(Runnable r); @Override Collection<FastFuture<T>> getResults(); @Override Collection<FastFuture<T>> getAllResults(); boolean hasCapacity(final int i);  }