@Test public void collectConcurrentResults() throws CheckedFutureException { ExecutorService ex = Executors.newFixedThreadPool(10); try { List<Promise<Integer>> promises = Stream.generate(() -> Promise.<Integer>apply()).limit(20000).collect(toList()); AtomicBoolean start = new AtomicBoolean(); Future<List<Integer>> future = Future.collect(promises); for (Promise<Integer> p : promises) { ex.submit(() -> { while (true) { if (start.get()) break; } p.setValue(p.hashCode()); }); } start.set(true); List<Integer> expected = promises.stream().map(p -> p.hashCode()).collect(toList()); List<Integer> result = future.get(Duration.ofSeconds(1)); assertArrayEquals(expected.toArray(), result.toArray()); } finally { ex.shutdown(); } }
@SuppressWarnings("unchecked") public final void collect(final T value, final int i) { results[i] = value; if (count.decrementAndGet() == 0) setValue((List<T>) Arrays.asList(results)); }
CollectPromise extends Promise<List<T>> { @SuppressWarnings("unchecked") public final void collect(final T value, final int i) { results[i] = value; if (count.decrementAndGet() == 0) setValue((List<T>) Arrays.asList(results)); } }
CollectPromise extends Promise<List<T>> { @SuppressWarnings("unchecked") public final void collect(final T value, final int i) { results[i] = value; if (count.decrementAndGet() == 0) setValue((List<T>) Arrays.asList(results)); } CollectPromise(final List<? extends Future<T>> list); }
CollectPromise extends Promise<List<T>> { @SuppressWarnings("unchecked") public final void collect(final T value, final int i) { results[i] = value; if (count.decrementAndGet() == 0) setValue((List<T>) Arrays.asList(results)); } CollectPromise(final List<? extends Future<T>> list); @SuppressWarnings("unchecked") final void collect(final T value, final int i); }
CollectPromise extends Promise<List<T>> { @SuppressWarnings("unchecked") public final void collect(final T value, final int i) { results[i] = value; if (count.decrementAndGet() == 0) setValue((List<T>) Arrays.asList(results)); } CollectPromise(final List<? extends Future<T>> list); @SuppressWarnings("unchecked") final void collect(final T value, final int i);  }