Implementing custom intermediate operations on Java 8 Streams

Question

I'm trying to work out how to implement a custom intermediate operation on a Java 8 Stream. And it seems that I'm locked out :(

Specifically I want to take a stream and return every entry up to and including the first one that has a particular value. And I want to stop generating any after that - making it short-circuiting.

It's running a series of validation checks on input data. I want to stop on the first Error, if there is one, but I want to collate Warnings on the way. And because these validation checks might be expensive - involving database lookups, for example - I want to only run the minimum set needed.

So the code would be something like:

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);

It seems that I should be able to do something with ReferencePipeline.StatefulOp, except that it's all package scope and so I can't extend it. And so I'm wondering what the correct way to achieve this is? Or if it's even possible?

Note as well - this needs to be in Java 8, and not 9+ since we're not there yet for various unrelated reasons.

Cheers

Answer 1

Generally, custom operations will need to deal with the Spliterator interface. It extends the concept of the Iterator by adding characteristics and size information and the ability to split off a part of the elements as another spliterator (hence its name). It also simplifies the iteration logic by only needing one method.

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);
}

To keep the stream’s properties, we query the parallel status first, to reestablish it for the new stream. Also, we register a close action that will close the original stream.

The main work is to implement a Spliterator decorating the previous stream state’s spliterator.

The characteristics are kept, except for the SIZED and SUBSIZED, as our operation results in an unpredictable size. The original size is still passed through, it will now be used as an estimate.

This solution stores the Consumer passed to tryAdvance for the duration of the operation, to be able to use the same adapter consumer, avoiding to create a new one for each iteration. This works, as it is guaranteed that tryAdvance is never invoked concurrently.

Parallelism is done via splitting, which is inherited from AbstractSpliterator. This inherited implementation will buffer some elements, which is reasonable, as implementing a better strategy for an operation like takeWhile is really complicated.

So you can use it like

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

which will print

foo
bar
baz

or

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);

which will print

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

which shows that it does not process more than necessary. Before the takeWhile stage, we have to encounter the first non-matching element, after that, we only encounter the elements up to that.

Answer 2

I admit that code wise, Holger's answer is a lot more sexy, but may be this is somehow easier to read:

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

You can do it with a trick:

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();

The List<ValidationResult> res will contains your interested values.

Answer 4

You can use the following structure;

AtomicBoolean gateKeeper = new AtomicBoolean(true);    
Optional<Foo> result = validators.stream()
    .filter(validator -> gateKeeper.get() 
                && gateKeeper.compareAndSet(true, !validator.validate(data).isError()) 
                && gateKeeper.get())
    .reduce(...) //have the first n non-error validators here

The filter with gateKeeper acts as an short circuiting logic and keeps going until it encounters the first isError() == true case, rejects it, & then shuts the doors for other validate() calls from then on. It looks a bit crazy, but it is much simpler than other custom implementations & might work perfectly if it suits your requirement.

Not 100% sure if this is helpful since I ignore the result of validator.validate(data) apart from isError() result, and the fact that it belongs to whichever validator in the list.