@Test public void sizeOrDurationConcurrent() throws Exception { final int maxBoundaries = 1_000; final Collection<Integer> items = unmodifiableCollection(range(1, 1_000).toFuture().get()); final List<Integer> receivedFromBoundaries = new ArrayList<>(items.size()); BufferStrategy<Integer, Accumulator<Integer, Iterable<Integer>>, Iterable<Integer>> strategy = forCountOrTime(1, ofDays(1), executor); CountDownLatch boundariesDone = new CountDownLatch(1); BlockingQueue<Accumulator<Integer, Iterable<Integer>>> boundaries = new LinkedBlockingQueue<>(); BlockingQueue<Accumulator<Integer, Iterable<Integer>>> boundariesToFinish = new LinkedBlockingQueue<>(); strategy.boundaries() .beforeOnNext(boundaries::add) .takeAtMost(maxBoundaries) .afterFinally(boundariesDone::countDown) .ignoreElements().subscribe(); CyclicBarrier accumulateAndTimerStarted = new CyclicBarrier(2); Future<Object> timersFuture = jdkExecutor.submit(() -> { accumulateAndTimerStarted.await(); while (boundariesDone.getCount() > 0) { timers.take().onComplete(); } return null; }); Future<Object> accumulateFuture = jdkExecutor.submit(() -> { accumulateAndTimerStarted.await(); int boundariesReceived = 0; Iterator<Integer> itemsToPopulate = items.iterator(); while (itemsToPopulate.hasNext()) { Accumulator<Integer, Iterable<Integer>> boundary = boundaries.take(); if (++boundariesReceived < maxBoundaries) { boundary.accumulate(itemsToPopulate.next()); } else { do { boundary.accumulate(itemsToPopulate.next()); } while (itemsToPopulate.hasNext()); } boundariesToFinish.add(boundary); } return null; }); boundariesDone.await(); timersFuture.get(); accumulateFuture.get(); Accumulator<Integer, Iterable<Integer>> boundary; while ((boundary = boundariesToFinish.poll()) != null) { for (Integer item : boundary.finish()) { receivedFromBoundaries.add(item); } } assertThat("Unexpected items received.", receivedFromBoundaries, contains(items.toArray())); }
public static <T> BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime( final int count, final Duration duration) { return forCountOrTime(count, duration, IMMEDIATE_EXECUTOR); }
BufferStrategies { public static <T> BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime( final int count, final Duration duration) { return forCountOrTime(count, duration, IMMEDIATE_EXECUTOR); } }
BufferStrategies { public static <T> BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime( final int count, final Duration duration) { return forCountOrTime(count, duration, IMMEDIATE_EXECUTOR); } private  BufferStrategies(); }
BufferStrategies { public static <T> BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime( final int count, final Duration duration) { return forCountOrTime(count, duration, IMMEDIATE_EXECUTOR); } private  BufferStrategies(); static BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime(
            final int count, final Duration duration); static BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime(
            final int count, final Duration duration, final Executor executor); static BufferStrategy<T, Accumulator<T, B>, B> forCountOrTime(
            final int count, final Duration duration, Supplier<BC> accumulatorSupplier); static BufferStrategy<T, Accumulator<T, B>, B> forCountOrTime(
            final int count, final Duration duration, Supplier<BC> accumulatorSupplier,
            final Executor executor); }
BufferStrategies { public static <T> BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime( final int count, final Duration duration) { return forCountOrTime(count, duration, IMMEDIATE_EXECUTOR); } private  BufferStrategies(); static BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime(
            final int count, final Duration duration); static BufferStrategy<T, Accumulator<T, Iterable<T>>, Iterable<T>> forCountOrTime(
            final int count, final Duration duration, final Executor executor); static BufferStrategy<T, Accumulator<T, B>, B> forCountOrTime(
            final int count, final Duration duration, Supplier<BC> accumulatorSupplier); static BufferStrategy<T, Accumulator<T, B>, B> forCountOrTime(
            final int count, final Duration duration, Supplier<BC> accumulatorSupplier,
            final Executor executor);  }