@UnstableApi
    default <E extends Throwable> StreamMessage<T> recoverAndResume(
            Class<E> causeClass, Function<? super E, ? extends StreamMessage<T>> function) {
        requireNonNull(causeClass, "causeClass");
        requireNonNull(function, "function");
        return recoverAndResume(cause -> {
            if (!causeClass.isInstance(cause)) {
                return Exceptions.throwUnsafely(cause);
            }
            try {
                final StreamMessage<T> recoveredStreamMessage = function.apply((E) cause);
                requireNonNull(recoveredStreamMessage, "recoveredStreamMessage");
                return recoveredStreamMessage;
            } catch (Throwable t) {
                return Exceptions.throwUnsafely(cause);
            }
        });
    }
Recovers a failed {@link StreamMessage} and resumes by subscribing to a returned fallback
{@link StreamMessage} when the thrown {@link Throwable} is the same type or a subtype of the
specified {@code causeClass}.

<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(IllegalStateException.class, cause -> StreamMessage.of(3, 4));

assert resumed.collect().join().equals(List.of(1, 2, 3, 4));

StreamWriter<Integer> stream = StreamMessage.streaming();
stream.write(1);
stream.write(2);
stream.write(3);
stream.close(new IllegalStateException("test exception"));
// Use the shortcut recover method as a chain.
StreamMessage<Integer> recoverChain =
    stream.recoverAndResume(RuntimeException.class, cause -> {
        final IllegalArgumentException ex = new IllegalArgumentException("oops..");
        // If a aborted StreamMessage returned from the first chain
        return StreamMessage.aborted(ex);
    })
    // If the shortcut exception type is correct, catch and recover in the second chain.
    .recoverAndResume(IllegalArgumentException.class, cause -> StreamMessage.of(4, 5));

recoverChain.collect().join();

StreamWriter<Integer> stream = StreamMessage.streaming();
stream.write(1);
stream.write(2);
stream.close(ClosedStreamException.get());
// If the exception type does not match
StreamMessage<Integer> mismatchRecovered =
    stream.recoverAndResume(IllegalStateException.class, cause -> StreamMessage.of(3, 4));

// In this case, CompletionException is thrown. (can't recover exception)
mismatchRecovered.collect().join();
}</pre>
core/src/main/java/com/linecorp/armeria/common/stream/StreamMessage.java