Java 8 extending stream

Question

I'm attempting to extend Java 8's Stream implementation.

I have this interface:

public interface StreamStuff<T> extends Stream<T> {

    Stream<T> delegate();
    default Stream<T> biggerThanFour() {
        return delegate().filter(i -> ((Double)i > 4));
    }
}

And in my main method:

int arr [] = {1,2,3,4,5,6};

Object array [] = ((StreamStuff)Arrays
            .stream(arr))
            .biggerThanFour()
            .toArray();

I'm trying to cast the Stream, to my interface StreamStuff, and use my method.

Im getting the following error:

Exception in thread "main" java.lang.ClassCastException: java.util.stream.IntPipeline$Head cannot be cast to StreamStuff

I get the same error when I do:

StreamStuff ss = (StreamStuff)Arrays.stream(arr);

I'm wondering if this sort of thing is even possible and if so, how do I achieve this? For reference I'm kind of using this article as a guide.

Answer 1

You are calling stream() on the Arrays class, which creates its own Stream implementation without any connection to yours. You'd have to produce the Stream yourself, or wrap a stream you obtained elsewhere, in order for something like this to work. Something like this:

int[] filtered = new StreamStuff(Arrays.stream(arr)).biggerThanFour().toArray();

However, in your case, why don't you just filter?

int[] filtered = Arrays.stream(arr).filter(i -> i > 4).toArray();

Answer 2

As already mentioned, you can create your own wrapper implementation:

public class MyStream<T> implements Stream<T> {

    private final Stream<T> delegate;

    public MyStream(Stream<T> delegate) {
        this.delegate = delegate;
    }

    @Override
    public Stream<T> filter(Predicate<? super T> predicate) {
        return delegate.filter(predicate);
    }

    @Override
    public void forEach(Consumer<? super T> action) {
        delegate.forEach(action);
    }

    MyStream<T> biggerThanFour() {
        return new MyStream<>(delegate.filter(i -> ((Double) i > 4)));
    }

    // all other methods from the interface
}

You will have to delegate all methods from the interface, so the class will be pretty big. You might consider adding a class StreamWrapper which will delegate all methods from the interface and then have your actual class StreamStuff extend StreamWrapper. This would allow you to have only your custom methods in StreamStuff and no other stream methods. You might also make all overridden methods in StreamWrapper final to avoid accidentally overriding them.

Then you can use it like this:

public static void main(String[] args) {
    Stream<Double> orgStream = Stream.of(1.0, 3.0, 7.0, 2.0, 9.0);

    MyStream<Double> myStream = new MyStream<>(orgStream);

    myStream.biggerThanFour().forEach(System.out::println);
}

The custom method returns a new wrapper so you can chain calls to your custom methods.

Just note that your cast to Double might throw ClassCastException so you might consider replacing generic T with Double, hence limiting the delegate stream to be of that specific type.

Answer 3

Another possibility is, if you don't want to handle all the Stream<T> delegates and new methods in the future, to use an Interface with a more wrapped lambda stream method:

public interface MyStream<T> {

    Stream<T> stream();

    static <T> MyStream<T> of(Stream<T> stream) {
        return () -> stream;
    }

    default <U> MyStream<U> stream(Function<Stream<T>, Stream<U>> stream) {
        return of(stream.apply(stream()));
    }

    //Watch out with Double cast. Check the type in method or restrict it via generic
    default MyStream<T> biggerThanFour() {
        return of(stream().filter(i -> ((Double) i > 4)));
    }

    //Watch out with Double cast. Check the type in method or restrict it via generic
    //Another method
    default MyStream<T> biggerThanFourteen() {
        return of(stream().filter(i -> ((Double) i > 14)));
    }
}

So you have your Interface with your delegate here stream() method which also handles the base stream terminal methods,
the static creation method of(...),
again a stream(...) method but with a Function<T,U> as parameter for handling the base Stream intermediate methods
and of course your custom methods like biggerThanFour(). So the drawback is here that you cant extend directly from Stream<T> (unfortunately the Stream<T> has not only default methods for one standard implementation) to bypass the delegates.
Also the handling is a small drawback but I think in most cases it's fine e.g.:

List<Integer> doubles = MyStream.of(Stream.of(1.0, 3.0, 7.0, 2.0, 9.0)) // create instance
                .biggerThanFour() //call MyStream methods
                .stream(doubleStream -> doubleStream.map(aDouble -> aDouble * 2)) //Do youre base stream intermediate methods and return again MyStream so you can call more specific custom methods
                .biggerThanFourteen()
                .stream() // call the base stream more or less your delegate for last intermediate methods and terminal method
                .mapToInt(Double::intValue)
                .boxed() //Ah if you have IntStreams and similar you can call the boxed() method to get an equivalent stream method.
                .collect(Collectors.toList()); // terminal method call

So list content is [18] ;)