RxJava中遇TimeoutException时恢复Flowable事件接收的问题

Hey there! Let's break down why your current approach isn't working and how to fix it to meet your requirement of continuing to receive events from the original Flowable even after a timeout.

Why Your Current Code Fails #

The timeout operator in RxJava has a key behavior you're hitting here: when a timeout occurs, it cancels the subscription to the original Flowable and emits a TimeoutException.

In your code, onErrorResumeNext replaces the errored Flowable with Flowable.just(new WorkExperience())—so once the timeout happens, you're no longer listening to the original Flowable at all. Any subsequent events from the original stream will never reach your subscriber because the subscription was canceled.

Solution 1: For Continuous Streams (Keep Listening After Timeout) #

If your getFlowable() is a continuous stream (e.g., real-time updates that might pause and resume), you need to detect timeouts without canceling the original subscription. Here's how to do that by separating the timeout logic from the original stream:

Flowable<WorkExperience> originalStream = getFlowable();

// Create a stream that emits a signal if 5 seconds pass without any event from the original stream
Flowable<Long> timeoutSignal = originalStream
    .map(ignored -> 0L)
    .startWithItem(0L) // Reset timer immediately on subscription
    .switchMap(ignored -> Flowable.timer(5, TimeUnit.SECONDS))
    .observeOn(AndroidSchedulers.mainThread());

// Merge the original stream and timeout signal, handling each type of event
Flowable.merge(
    originalStream.observeOn(AndroidSchedulers.mainThread()),
    timeoutSignal
)
.subscribe(
    item -> {
        if (item instanceof WorkExperience) {
            // Handle your normal WorkExperience event
            WorkExperience workExperience = (WorkExperience) item;
            // do something
        } else {
            // Timeout occurred—hide loading, but keep listening to the original stream
            view.hideLoading();
        }
    },
    throwable -> {
        Timber.e(throwable);
        view.hideLoading();
    }
);

How this works:

• Every time the original stream emits an event, switchMap cancels the current timer and starts a new one.
• If no event comes within 5 seconds, the timer emits a signal, triggering your hideLoading() logic.
• The original stream remains subscribed, so any future events will still be received and processed.

Solution 2: For Single Requests (Show Default on Timeout, Update if Original Succeeds Later) #

If getFlowable() is a single request (e.g., a network call that might take time, and you want to show a default value on timeout but update if the request eventually succeeds), use merge to combine the original stream with a timeout fallback:

Flowable<WorkExperience> originalRequest = getFlowable()
    .observeOn(AndroidSchedulers.mainThread());

Flowable<WorkExperience> timeoutFallback = Flowable.timer(5, TimeUnit.SECONDS)
    .observeOn(AndroidSchedulers.mainThread())
    .doOnNext(ignored -> view.hideLoading())
    .map(ignored -> new WorkExperience()); // Default value on timeout

// Merge both streams—you'll get the fallback first if timeout hits, then the real value if the request succeeds
Flowable.merge(originalRequest, timeoutFallback)
    .subscribe(
        workExperience -> {
            // Update your UI with either the default or real value
            // do something
        },
        throwable -> {
            Timber.e(throwable);
            view.hideLoading();
        }
    );

This approach keeps the original request subscribed even after the timeout, so if it eventually completes, you'll get the real WorkExperience event to update your UI.

The core issue with your initial code was that timeout cancels the original subscription. By separating the timeout logic or using merge instead of onErrorResumeNext, you can keep listening to the original Flowable while handling timeout scenarios.

内容的提问来源于stack exchange，提问作者Abhishek Bansal