@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<T>> toRemove = active.stream() .filter(cf -> cf.isDone()) .collect(Collectors.toList()); active.removeAll(toRemove); results.addAll(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(); } } } }
BatchingCollector 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<T>> toRemove = active.stream() .filter(cf -> cf.isDone()) .collect(Collectors.toList()); active.removeAll(toRemove); results.addAll(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(); } } } } }
BatchingCollector 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<T>> toRemove = active.stream() .filter(cf -> cf.isDone()) .collect(Collectors.toList()); active.removeAll(toRemove); results.addAll(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(); } } } } BatchingCollector(final MaxActive maxActive, final BlockingStream<T> blocking); }
BatchingCollector 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<T>> toRemove = active.stream() .filter(cf -> cf.isDone()) .collect(Collectors.toList()); active.removeAll(toRemove); results.addAll(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(); } } } } BatchingCollector(final MaxActive maxActive, final BlockingStream<T> blocking); @Override void accept(final FastFuture<T> t); @Override void block(final Function<FastFuture<T>, T> safeJoin); @Override Collection<FastFuture<T>> getResults(); @Override Collection<FastFuture<T>> getAllResults(); }
BatchingCollector 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<T>> toRemove = active.stream() .filter(cf -> cf.isDone()) .collect(Collectors.toList()); active.removeAll(toRemove); results.addAll(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(); } } } } BatchingCollector(final MaxActive maxActive, final BlockingStream<T> blocking); @Override void accept(final FastFuture<T> t); @Override void block(final Function<FastFuture<T>, T> safeJoin); @Override Collection<FastFuture<T>> getResults(); @Override Collection<FastFuture<T>> getAllResults();  }