@Test(expected = IllegalStateException.class) public void testIncorrectImplementation() { LogDataConsumerSplitter<Integer, Integer> splitter = new LogDataConsumerSplitter<Integer, Integer>() { @Override protected StreamDataAcceptor<Integer> createSplitter(@NotNull Context ctx) { return item -> {}; } }; StreamSupplier.ofIterable(VALUES_1).streamTo(splitter.consume().getConsumer()); }
@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();  }