Как отменить логику блока восходящего потока Flowable при возникновении исключения TimeoutException ⇐ Android

[Anonymous]

У меня есть код rxjava2, как показано ниже:

Код: Выделить всё

public class Demo {

/** @noinspection ResultOfMethodCallIgnored*/
@SuppressLint("CheckResult")
public static void test() {
Flowable.just(false).map(aBoolean -> block())
.retryWhen(throwableFlowable -> {
AtomicInteger retryCounter = new AtomicInteger();
return throwableFlowable.takeWhile(throwable -> {
if (retryCounter.getAndIncrement() < 5) {
return true;
}

Exception exception;
try {
exception = (Exception) throwable;
} catch (Exception e) {
exception = new Exception(throwable);
}
throw exception;
});
}).timeout(1000, TimeUnit.MILLISECONDS).subscribe(aBoolean -> {
Log.e("Demo", "onNext");
}, throwable -> {
Log.e("Demo", "onError", throwable);
}, () -> {
Log.e("Demo", "onComplete");
}, subscription -> {
subscription.request(Long.MAX_VALUE);
});
}

private static boolean block() {
Log.e("Demo", "block");
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
// do nothing
}
Log.e("Demo", "block end");
return true;
}

}

При возникновении исключения TimeoutException код метода block() все еще выполняется, и через некоторое время будет распечатан журнал «конец блока».
Понятия не имею. Я ожидаю, что смогу отменить логику блока восходящего потока Flowable при возникновении исключения TimeoutException. И более того, я мог бы остановить выполнение метода block! Что мне попробовать дальше?

Подробнее здесь: https://stackoverflow.com/questions/791 ... ion-throws