default StreamMessage<T> recoverAndResume(
            Function<? super Throwable, ? extends StreamMessage<T>> function) {
        requireNonNull(function, "function");
        return new RecoverableStreamMessage<>(this, function, /* allowResuming */ true);
    }
Recovers a failed {@link StreamMessage} and resumes by subscribing to a returned fallback
{@link StreamMessage} when any error occurs.

<p>Example:<pre>{@code
StreamWriter<Integer> stream = StreamMessage.streaming();
stream.write(1);
stream.write(2);
stream.close(new IllegalStateException("Oops..."));
StreamMessage<Integer> resumed = stream.recoverAndResume(cause -> StreamMessage.of(3, 4));

assert resumed.collect().join().equals(List.of(1, 2, 3, 4));
}</pre>
core/src/main/java/com/linecorp/armeria/common/stream/StreamMessage.java