在Java8流上实现自定义中间操作

Question

我试图找出如何在Java8 Stream上实现自定义中间操作。似乎我被锁在外面了：（

具体地说，我想获取一个流，并返回每个条目，包括第一个具有特定值的条目。我想在那之后停止发电，让它短路。

它正在对输入数据进行一系列验证检查。如果有第一个错误，我想停止，但我想在途中整理警告。而且因为这些验证检查可能很昂贵——例如，涉及数据库查找——我只想运行所需的最小集。

所以代码应该是这样的：

Optional<ValidationResult> result = validators.stream()
    .map(validator -> validator.validate(data))
    .takeUntil(result -> result.isError()) // This is the bit I can't do
    .reduce(new ValidationResult(), ::mergeResults);

看来我应该可以用ReferencePipeline做点什么。StatefulOp，除了它是所有包的作用域，所以我不能扩展它。所以我想知道实现这一目标的正确方法是什么？或者如果可能的话？

还要注意——这需要在Java8中实现，而不是在9中，因为我们由于各种不相关的原因还没有实现。

干杯

Answer 1

你可以用一个技巧来做到：

List<ValidationResult> res = new ArrayList<>(); // Can modify it with your `mergeResults` instead of list

Optional<ValidationResult> result = validators.stream()
    .map(validator -> validator.validate(data))
    .map(v -> {
       res.add(v);
       return v;
    })
    .filter(result -> result.isError())
    .findFirst();

列表

Answer 2

我承认在代码方面，霍尔格的答案要性感得多，但也许这更容易阅读：

public static <T> Stream<T> takeUntilIncluding(Stream<T> s, Predicate<? super T> condition) {

    class Box implements Consumer<T> {

        boolean stop = false;

        T t;

        @Override
        public void accept(T t) {
            this.t = t;
        }
    }

    Box box = new Box();

    Spliterator<T> original = s.spliterator();

    return StreamSupport.stream(new AbstractSpliterator<>(
        original.estimateSize(),
        original.characteristics() & ~(Spliterator.SIZED | Spliterator.SUBSIZED)) {

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {

            if (!box.stop && original.tryAdvance(box) && condition.test(box.t)) {
                action.accept(box.t);
                return true;
            }

            box.stop = true;

            return false;
        }
    }, s.isParallel());

}

Answer 3

通常，自定义操作需要处理Spliterator接口。它扩展了Iterator的概念，添加了特性和大小信息，并能够将一部分元素拆分为另一个拆分器（因此得名）。它还简化了迭代逻辑，只需要一种方法。

public static <T> Stream<T> takeWhile(Stream<T> s, Predicate<? super T> condition) {
    boolean parallel = s.isParallel();
    Spliterator<T> spliterator = s.spliterator();
    return StreamSupport.stream(new Spliterators.AbstractSpliterator<T>(
        spliterator.estimateSize(),
        spliterator.characteristics()&~(Spliterator.SIZED|Spliterator.SUBSIZED)) {
            boolean active = true;
            Consumer<? super T> current;
            Consumer<T> adapter = t -> {
                if((active = condition.test(t))) current.accept(t);
            };

            @Override
            public boolean tryAdvance(Consumer<? super T> action) {
                if(!active) return false;
                current = action;
                try {
                    return spliterator.tryAdvance(adapter) && active;
                }
                finally {
                    current = null;
                }
            }
        }, parallel).onClose(s::close);
}

为了保留流的属性，我们首先查询并行状态，以便为新流重新建立它。此外，我们注册一个关闭操作，将关闭原始流。

主要工作是实现一个拆分器装饰先前流状态的拆分器。

除了大小的和补贴的之外，这些特征都被保留下来，因为我们的操作会导致不可预测的大小。原始尺寸仍在传递中，现在将用作估计值。

此解决方案在操作期间存储传递给tryAdvance的Consumer，以便能够使用相同的适配器Consumer，避免为每次迭代创建新的适配器Consumer。这是可行的，因为可以保证tryAdvance永远不会被并发调用。

并行性是通过拆分实现的，拆分是从AbstractSpliterator继承的。这种继承的实现将缓冲一些元素，这是合理的，因为为像takeWhile这样的操作实现更好的策略非常复杂。

所以你可以像这样使用它

    takeWhile(Stream.of("foo", "bar", "baz", "hello", "world"), s -> s.length() == 3)
        .forEach(System.out::println);

哪个会打印

foo
bar
baz

或

takeWhile(Stream.of("foo", "bar", "baz", "hello", "world")
    .peek(s -> System.out.println("before takeWhile: "+s)), s -> s.length() == 3)
    .peek(s -> System.out.println("after takeWhile: "+s))
    .forEach(System.out::println);

哪个会打印

before takeWhile: foo
after takeWhile: foo
foo
before takeWhile: bar
after takeWhile: bar
bar
before takeWhile: baz
after takeWhile: baz
baz
before takeWhile: hello

这表明它不会处理超出需要的内容。在takeWhile阶段之前，我们必须遇到第一个不匹配的元素，在那之后，我们只会遇到之前的元素。