@Test public void testConsumes() { List<StreamConsumerToList<Integer>> consumers = asList( StreamConsumerToList.create(), StreamConsumerToList.create()); Iterator<StreamConsumerToList<Integer>> iterator = consumers.iterator(); LogDataConsumerSplitter<Integer, Integer> splitter = new LogDataConsumerSplitterStub<>(() -> { StreamConsumerToList<Integer> next = iterator.next(); return StreamConsumerWithResult.of(next, next.getResult()); }); assertStreamResult(VALUES_1, splitter.consume(), consumers.get(0).getResult()); assertStreamResult(VALUES_2, splitter.consume(), consumers.get(1).getResult()); }
@Override public StreamConsumerWithResult<T, List<D>> consume() { AsyncCollector<List<D>> diffsCollector = AsyncCollector.create(new ArrayList<>()); Context ctx = new Context(); createSplitter(ctx); StreamSplitter<T, Object> splitter = StreamSplitter.create(acceptors -> { ctx.acceptors = Arrays.asList(acceptors).iterator(); return createSplitter(ctx); }); checkState(!ctx.logDataConsumers.isEmpty()); for (LogDataConsumer<?, D> logDataConsumer : ctx.logDataConsumers) { diffsCollector.addPromise( splitter.newOutput().streamTo(((LogDataConsumer<Object, D>) logDataConsumer).consume()), List::addAll); } return StreamConsumerWithResult.of(splitter.getInput(), diffsCollector.run().get()); }
LogDataConsumerSplitter implements LogDataConsumer<T, D> { @Override public StreamConsumerWithResult<T, List<D>> consume() { AsyncCollector<List<D>> diffsCollector = AsyncCollector.create(new ArrayList<>()); Context ctx = new Context(); createSplitter(ctx); StreamSplitter<T, Object> splitter = StreamSplitter.create(acceptors -> { ctx.acceptors = Arrays.asList(acceptors).iterator(); return createSplitter(ctx); }); checkState(!ctx.logDataConsumers.isEmpty()); for (LogDataConsumer<?, D> logDataConsumer : ctx.logDataConsumers) { diffsCollector.addPromise( splitter.newOutput().streamTo(((LogDataConsumer<Object, D>) logDataConsumer).consume()), List::addAll); } return StreamConsumerWithResult.of(splitter.getInput(), diffsCollector.run().get()); } }
LogDataConsumerSplitter implements LogDataConsumer<T, D> { @Override public StreamConsumerWithResult<T, List<D>> consume() { AsyncCollector<List<D>> diffsCollector = AsyncCollector.create(new ArrayList<>()); Context ctx = new Context(); createSplitter(ctx); StreamSplitter<T, Object> splitter = StreamSplitter.create(acceptors -> { ctx.acceptors = Arrays.asList(acceptors).iterator(); return createSplitter(ctx); }); checkState(!ctx.logDataConsumers.isEmpty()); for (LogDataConsumer<?, D> logDataConsumer : ctx.logDataConsumers) { diffsCollector.addPromise( splitter.newOutput().streamTo(((LogDataConsumer<Object, D>) logDataConsumer).consume()), List::addAll); } return StreamConsumerWithResult.of(splitter.getInput(), diffsCollector.run().get()); }  }
LogDataConsumerSplitter implements LogDataConsumer<T, D> { @Override public StreamConsumerWithResult<T, List<D>> consume() { AsyncCollector<List<D>> diffsCollector = AsyncCollector.create(new ArrayList<>()); Context ctx = new Context(); createSplitter(ctx); StreamSplitter<T, Object> splitter = StreamSplitter.create(acceptors -> { ctx.acceptors = Arrays.asList(acceptors).iterator(); return createSplitter(ctx); }); checkState(!ctx.logDataConsumers.isEmpty()); for (LogDataConsumer<?, D> logDataConsumer : ctx.logDataConsumers) { diffsCollector.addPromise( splitter.newOutput().streamTo(((LogDataConsumer<Object, D>) logDataConsumer).consume()), List::addAll); } return StreamConsumerWithResult.of(splitter.getInput(), diffsCollector.run().get()); }  @Override StreamConsumerWithResult<T, List<D>> consume(); }
LogDataConsumerSplitter implements LogDataConsumer<T, D> { @Override public StreamConsumerWithResult<T, List<D>> consume() { AsyncCollector<List<D>> diffsCollector = AsyncCollector.create(new ArrayList<>()); Context ctx = new Context(); createSplitter(ctx); StreamSplitter<T, Object> splitter = StreamSplitter.create(acceptors -> { ctx.acceptors = Arrays.asList(acceptors).iterator(); return createSplitter(ctx); }); checkState(!ctx.logDataConsumers.isEmpty()); for (LogDataConsumer<?, D> logDataConsumer : ctx.logDataConsumers) { diffsCollector.addPromise( splitter.newOutput().streamTo(((LogDataConsumer<Object, D>) logDataConsumer).consume()), List::addAll); } return StreamConsumerWithResult.of(splitter.getInput(), diffsCollector.run().get()); }  @Override StreamConsumerWithResult<T, List<D>> consume();  }