Note: Please see the edit at the end of this post, which fixes a bug in my original answer. I'm leaving my original answer anyway, because it's still useful for many cases and I think it helps solve OP's question, at least with some restrictions.
Your approach with Iterator goes in the right direction. The solution might be drafted as follows: convert the stream to an iterator, wrap the iterator as you have already done, and then create a stream from the wrapper iterator, except that you should use a Spliterator instead. Here's the code:
private static <T> Stream<T> asNonThrowingStream(
Stream<T> stream,
Supplier<? extends T> valueOnException) {
// Get spliterator from original stream
Spliterator<T> spliterator = stream.spliterator();
// Return new stream from wrapper spliterator
return StreamSupport.stream(
// Extending AbstractSpliterator is enough for our purpose
new Spliterators.AbstractSpliterator<T>(
spliterator.estimateSize(),
spliterator.characteristics()) {
// We only need to implement tryAdvance
@Override
public boolean tryAdvance(Consumer<? super T> action) {
try {
return spliterator.tryAdvance(action);
} catch (RuntimeException e) {
action.accept(valueOnException.get());
return true;
}
}
}, stream.isParallel());
}
We are extending AbstractSpliterator to wrap the spliterator returned by the original stream. We only need to implement the tryAdvance method, which either delegates to the original spliterator's tryAdvance method, or catches RuntimeException and invokes the action with the supplied valueOnException value.
Spliterator's contract specifies that the return value of tryAdvance must be true if the action is consumed, so if a RuntimeException is catched, it means that the original spliterator has thrown it from within its own tryAdvance method. Thus, we return true in this case, meaning that the element was consumed anyway.
The original spliterator's estimate size and characteristics are preserved by passing these values as arguments to the constructor of AbstractSpliterator.
Finally, we create a new stream from the new spliterator via the StreamSupport.stream method. The new stream is parallel if the original one was also parallel.
Here's how to use the above method:
public Stream<String> convertToString(Stream<Integer> input) {
return asNonThrowingStream(input.map(String::valueOf), () -> "NaN");
}
Edit
As per Holger's comment below, user holi-java has kindly provided a solution that avoids the pitfalls pointed out by Holger.
Here's the code:
<T> Stream<T> exceptionally(Stream<T> source, BiConsumer<Exception, Consumer<? super T>> handler) {
class ExceptionallySpliterator extends AbstractSpliterator<T>
implements Consumer<T> {
private Spliterator<T> source;
private T value;
private long fence;
ExceptionallySpliterator(Spliterator<T> source) {
super(source.estimateSize(), source.characteristics());
this.fence = source.getExactSizeIfKnown();
this.source = source;
}
@Override
public Spliterator<T> trySplit() {
Spliterator<T> it = source.trySplit();
return it == null ? null : new ExceptionallySpliterator(it);
}
@Override
public boolean tryAdvance(Consumer<? super T> action) {
return fence != 0 && consuming(action);
}
private boolean consuming(Consumer<? super T> action) {
Boolean state = tryConsuming(action);
if (state == null) {
return true;
}
if (state) {
action.accept(value);
value = null;
return true;
}
return false;
}
private Boolean tryConsuming(Consumer<? super T> action) {
fence--;
try {
return source.tryAdvance(this);
} catch (Exception ex) {
handler.accept(ex, action);
return null;
}
}
@Override
public void accept(T value) {
this.value = value;
}
}
return stream(new ExceptionallySpliterator(source.spliterator()), source.isParallel()).onClose(source::close);
}
Please refer to the tests if you want to further know about this solution.
