How to catch exceptions within Java 8 Stream.flatMap(..)

Question

Given a Stream and a method that returns a Stream for different arguments as data source, I'm looking for a way to merge the streams via flatMap(..) and catching certain Exceptions during the execution.

Let's take the following code snippet:

public class FlatMap {

    public static void main(final String[] args) {
        long count;

        // this might throw an exception
        count = Stream.of(0.2, 0.5, 0.99).flatMap(chance -> getGenerator(chance, 20)).count();

        // trying to catch the exception in flatMap() will not work
        count = Stream.of(0.2, 0.5, 0.99).flatMap(chance -> {
            try {
                return getGenerator(chance, 20);
            } catch (final NullPointerException e) {
                return Stream.empty();
            }
        }).count();

        System.out.println(count);
    }

    // !! we cannot change this method, we simply get a Stream
    static Stream<Object> getGenerator(final double chance, final long limit) {
        return Stream.generate(() -> {
            if (Math.random() < chance) return new Object();
            throw new NullPointerException();
        }).limit(limit);
    }
}

Is there any way to catch the exception of each individual Stream that was created by getGenerator(..) and simply suppress the Exception, replacing the "corrupted" Stream with an empty one or skip those elements from the specific generator Stream?

Answer 1

It is possible to wrap the Stream into another using the Spliterator. This method will protect a given Stream by catching the Exception and saving this state:

    static <T> Stream<T> protect(final Stream<T> stream) {
        final Spliterator<T> spliterator = stream.spliterator();
        return StreamSupport.stream(
                new Spliterators.AbstractSpliterator<T>(Long.MAX_VALUE,
                           spliterator.characteristics() & ~Spliterator.SIZED) {

                    private boolean corrupted = false;

                    @Override
                    public boolean tryAdvance(final Consumer<? super T> action) {
                        if (!corrupted) try {
                            return spliterator.tryAdvance(action);
                        } catch (final Exception e) {
                            // we suppress this one, stream ends here
                            corrupted = true;
                        }
                        return false;
                    }
                }, false);
    }

Then we can wrap our Stream method and safely pass it in flatMap(..):

// we protect the stream by a wrapper Stream
count = Stream.of(0.2, 0.5, 0.99)
              .flatMap(chance -> protect(getGenerator(chance, 20)))
              .count();

Answer 2

One work around is to force the Stream created by getGenerator to be evaluated within the flatMap method implementation. This forces the NullPointerException to be thrown within the try-catch block, and therefore, able to be handled.

To do this, you can collect the Stream (to a List for example):

getGenerator(chance, 20).collect(Collectors.toList()).stream()

Incorporating this into your original snippet:

public class FlatMap {

    public static void main(final String[] args) {
        long count;

        // trying to catch the exception in flatMap() will not work
        count = Stream.of(0.2, 0.5, 0.99)
            .flatMap(chance -> {
                try {
                    return getGenerator(chance, 20).collect(Collectors.toList()).stream();
                } 
                catch (final NullPointerException e) {
                    return Stream.empty();
                }
            })
            .count();

        System.out.println(count);
    }

    // !! we cannot change this method, we simply get a Stream
    static Stream<Object> getGenerator(final double chance, final long limit) {
        return Stream.generate(() -> {
            if (Math.random() < chance) return new Object();
            throw new NullPointerException();
        }).limit(limit);
    }
}

Warning: this approach may reduce performance if the getGenerator Stream would be better to evaluate lazily.