1
\$\begingroup\$

The company I work for is stuck with Java 8, and the Streams implementation in Java 8 is nice but somewhat rough around the edges.

Here is a helper class I cobbled together, using various snippets from the interwebs. But this worries me a bit, because I don't understand some of the Spliterator magic, and don't have the time to dive deeper into it. Of course I wrote tests, but it would be still reassuring to get some feedback.

package com.acme;

import java.util.Optional;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Predicate;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public final class Streams {

    private Streams() {
        // do not instantiate
    }

    public static <A, B, C> Stream<C> zipWith(Stream<A> leftStream, Stream<B> rightStream,
            BiFunction<A, B, C> combiner) {
        Spliterator<A> lefts = leftStream.spliterator();
        Spliterator<B> rights = rightStream.spliterator();
        return StreamSupport
                .stream(new Spliterators.AbstractSpliterator<C>(Long.min(lefts.estimateSize(), rights.estimateSize()),
                        lefts.characteristics() & rights.characteristics()) {
                    @Override
                    public boolean tryAdvance(Consumer<? super C> action) {
                        return lefts.tryAdvance(
                                left -> rights.tryAdvance(right -> action.accept(combiner.apply(left, right))));
                    }
                }, leftStream.isParallel() || rightStream.isParallel());
    }

    public static <A, C> Stream<C> zipWithIndex(Stream<A> leftStream, BiFunction<A, Integer, C> combiner) {
        return zipWith(leftStream, Stream.iterate(0, index -> index + 1), combiner);
    }

    public static <A> Stream<A> takeWhile(Stream<A> stream, Predicate<? super A> predicate) {
        Spliterator<A> spliterator = stream.spliterator();
        return StreamSupport.stream(new Spliterators.AbstractSpliterator<A>(spliterator.estimateSize(), 0) {
            boolean stillGoing = true;
            @Override
            public boolean tryAdvance(Consumer<? super A> consumer) {
                if (stillGoing) {
                    boolean hadNext = spliterator.tryAdvance(elem -> {
                        if (predicate.test(elem)) {
                            consumer.accept(elem);
                        } else {
                            stillGoing = false;
                        }
                    });
                    return hadNext && stillGoing;
                }
                return false;
            }
        }, false);
    }

    public static <A> Stream<A> dropWhile(Stream<A> stream, Predicate<? super A> predicate) {
        Spliterator<A> spliterator = stream.spliterator();
        return StreamSupport.stream(new Spliterators.AbstractSpliterator<A>(spliterator.estimateSize(), 0) {
            boolean dropped;
            public boolean tryAdvance(Consumer<? super A> action) {
                if (dropped) {
                    return spliterator.tryAdvance(action);
                }
                while (true) {
                    if (!(!dropped && spliterator.tryAdvance(t -> {
                        if (!predicate.test(t)) {
                            dropped = true;
                            action.accept(t);
                        }
                    }))) break;
                }
                return dropped;
            }
            public void forEachRemaining(Consumer<? super A> action) {
                while (!dropped) {
                    if (!tryAdvance(action)) {
                        return;
                    }
                }
                spliterator.forEachRemaining(action);
            }
        }, false);
    }

    public static <A> Stream<A> from(Iterable<A> iterable) {
        return StreamSupport.stream(iterable.spliterator(), false);
    }

    public static <A> Stream<A> from(Optional<A> optional) {
        return optional.map(Stream::of).orElseGet(Stream::empty);
    }
}
\$\endgroup\$
  • \$\begingroup\$ Maybe a bit off-topic in CodeReview, but I would advise to re-use a lib like github.com/amaembo/streamex for this. \$\endgroup\$ – RobAu Feb 14 '18 at 11:00
1
\$\begingroup\$
  • I've made similar Stream extensions in the past. However, I had used a decorator pattern. That way, the ability to chain stream operations (including the ones you add) remains intact. (this does not apply to the from() methods)

  • In zipWith(), I'd consider it more prudent to specify :

    leftStream.isParallel() && rightStream.isParallel()
    

    If either stream is not parallel, someone may have used operations on it, that are not suitable for parallel processing.

  • in dropWhile(), it is unnecessary to override forEachRemaining. the default implementation delegates to tryAdvance, and that works just fine.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.