Java 8 stream attaching error handling for later consumption

Question

Let's say I have the following method I want to refactor

protected Stream<T> parseFile(File file, Consumer<File> cleanup) {
  try {
    return parser.parse(file); // returns a Stream<T>
  } catch (XmlParseException e) { // child of RuntimeException
    throw new CustomRuntimeException(e);
  } finally {
    if (file != null) {
      cleanup.accept(file);
    }
  }

  throw new IllegalStateException("Should not happen");
}

This method's purpose is to act as a proxy attaching error handling on the stream rethrowing in a wrapping exception CustomRuntimeException. So when we consume it later in the flow, I don't have to handle those exceptions everywhere but only CustomRuntimeException.

Upstream, I used that method as follow

try {
  Stream<T> stream = parseFile(someFile);
  stream.map(t -> ...);
catch (CustomRuntimeException e) {
  // do some stuff
}

And here's what the parser.parse method looks like

public Stream<T> parse() {
  // ValueIterator<T> implements Iterator<T>, AutoCloseable
  XmlRootParser.ValueIterator<T> valueIterator = new XmlRootParser.ValueIterator(this.nodeConverter, this.reader, this.nodeLocalName, this.nodeName);
  Stream<T> stream = StreamSupport.stream(Spliterators.spliteratorUnknownSize(valueIterator, 1040), false);
  stream.onClose(valueIterator::close);
  return stream;
}

The exceptions I want to handle will be thrown by the ValueIterator.hasNext method. Which means they won't be thrown at Stream creation but only at Stream consumption (calling foreach/map/count/collect/... on the stream).

How do I attach error handling on my stream in method parseFile nicely without having to consume the stream? Is it possible?

Obviously this code will work only if the parser.parse method consume its stream before returning it. Which is against using streams.

Answer 1

The Stream’s backend which provides the iterator logic, is the Spliterator.

So you can wrap the element processing using a wrapper Spliterator like this:

class Wrapper<T> implements Spliterator<T> {
    final Spliterator<T> source;
    public Wrapper(Spliterator<T> source) {
        this.source = source;
    }
    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        try {
            return source.tryAdvance(action);
        }
        catch(XmlParseException ex) {
            throw new CustomRuntimeException(ex);
        }
    }
    @Override
    public void forEachRemaining(Consumer<? super T> action) {
        try {
            source.forEachRemaining(action);
        }
        catch(XmlParseException ex) {
            throw new CustomRuntimeException(ex);
        }
    }
    @Override public Spliterator<T> trySplit() {
        Spliterator<T> srcPrefix = source.trySplit();
        return srcPrefix == null? null: new Wrapper<>(srcPrefix);
    }
    @Override public long estimateSize() { return source.estimateSize(); }
    @Override public int characteristics() { return source.characteristics(); }
    @Override public Comparator<? super T> getComparator(){return source.getComparator();}
}

It retains all properties of the original Spliterator and only translates exceptions thrown during the iteration.

Then you can use it like

protected Stream<T> parseFile(File file) {
    Stream<T> s = parser.parse();
    return StreamSupport.stream(new Wrapper<>(s.spliterator()), s.isParallel())
                        .onClose(s::close);
}

And the caller should not forget to close the stream properly:

    ResultType result;
    try(Stream<T> s = parseFile(file)) {
        result = s.
         // other intermediate ops
         // terminal operation
    }

or

    ResultType result;
    try(Stream<T> s = parseFile(file)) {
        result = s.
         // other intermediate ops
         // terminal operation
    }
    finally {
        // other cleanup actions
    }

Answer 2

You could use helper stream initialization class that handles the process of preparing stream and catch any exception there. Consider following example:

import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class SafeInitializationStreamExample {

    public static void main(String[] args) {
        int sum = SafeInitializationStream.from(() -> Stream.of(1,2,3,4))
                .onInitializationError(t -> System.out.println(t.getMessage()))
                .mapToInt(it -> it)
                .sum();

        System.out.println(sum);

        List<Object> list = SafeInitializationStream.from(() -> parse("/tmp/test.log"))
                .onInitializationError(t -> System.out.println(t.getMessage()))
                .map(it -> it.toString())
                .collect(Collectors.toList());

        System.out.println(list);
    }

    private static <T> Stream<T> parse(String filename) {
        throw new RuntimeException("File does not exist!");
    }

    static class SafeInitializationStream<T> {
        private final Supplier<Stream<T>> streamSupplier;

        private SafeInitializationStream(Supplier<Stream<T>> streamSupplier) {
            this.streamSupplier = streamSupplier;
        }

        public static <T> SafeInitializationStream<T> from(Supplier<Stream<T>> streamSupplier) {
            return new SafeInitializationStream<>(streamSupplier);
        }

        public Stream<T> onInitializationError(Consumer<Throwable> onError) {
            try {
                return streamSupplier.get();
            } catch (Exception e) {
                onError.accept(e);
            }
            return Stream.empty();
        }
    }
}

In this example we introduce SafeInitializationStream class which expects a Supplier<Stream<T>>:

SafeInitializationStream.from(() -> Stream.of(1,2,3,4))

Using Supplier<Stream<T>> in this case makes stream initialization lazy - until we call Supplier.get() the body of this supplier is not executed. Now, when we call:

.onInitializationError(t -> System.out.println(t.getMessage()))

we execute suppliers body, catch any exception that may be thrown and handle this exception by passing Throwable to Consumer<Throwable> that was passed as a parameter to onInitializationError method. In case of exception Stream.empty() is returned (so you can safely apply all other transformations in chain. When there is no exception, Stream<T> provided by supplier is returned.

If you run following example you will get in console:

10
File does not exist!
[]

The first stream was consumed without any errors and the sum was returned correctly.

The second stream thrown exception during initialization, we have caught it and printed to the console and finally empty list was returned after consuming the stream.

Of course you can use Function<Throwable, Stream<T>> in onInitializationError method if you want to specify what Stream<T> is returned in case of an exception. In this case we assume that Stream.empty() is always returned in this case. Hope it helps.