Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Implementing custom intermediate operations on Java 8 Streams

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

like image 899
Graham Avatar asked Jun 14 '19 05:06

Graham


People also ask

What is intermediate operation Java 8?

Intermediate Operation- These operations are used to pipeline other methods and to transform into the other streams. They don't produce results because these operation does not invoke until the terminal operation gets executed. Below are the examples − sorted(Comparator<T>) peek(Consumer<T>)

What is intermediate operation in stream Java?

Stream. Intermediate operators do not execute until a terminal operation is invoked, i.e. they are not executed until a result of processing is actually needed. We will be discussing a few of the important and most frequently used: filter(predicate) Method.

Does Java 8 support streams?

Java 8 offers the possibility to create streams out of three primitive types: int, long and double. As Stream<T> is a generic interface, and there is no way to use primitives as a type parameter with generics, three new special interfaces were created: IntStream, LongStream, DoubleStream.

What are the differences between intermediate and terminal operations in Java 8 streams?

An intermediate operation is short circuiting in Java, if when presented with infinite input, it may produce a finite stream as a result. Terminal operations are called as short-circuiting if when presented with infinite input, it may terminate in finite time.


1 Answers

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.

like image 187
Holger Avatar answered Oct 21 '22 23:10

Holger