RXJava - 自收到最后一个项目以来发出“时钟滴答”项目

Question

我有一个 Observable 发射项目，我想将特殊项目合并到其中，因为收到最后一个项目后充当“时间滴答”。

我尝试使用 timeout+onErrorXXX 或 intervals，但无法按预期工作。

import io.reactivex.Observable;
import io.reactivex.functions.Function;
import org.apache.log4j.Logger;
import org.junit.Test;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class RXTest {
    private static final Logger log = Logger.getLogger(RXTest.class);

    @Test
    public void rxTest() throws InterruptedException {
        log.info("Starting");

        Observable.range(0, 26)
                .concatMap(
                        item -> Observable.just(item)
                                .delay(item, TimeUnit.SECONDS)
                )
                .timeout(100, TimeUnit.MILLISECONDS)
//                .retry()
                .onErrorResumeNext((Function) throwable -> {
                    if (throwable instanceof TimeoutException) {
                        return Observable.just(-1);
                    }
                    throw new RuntimeException((Throwable)throwable);
                })
                .subscribe(
                        item -> log.info("Received " + item),
                        throwable -> log.error("Thrown" + throwable),
                        () -> log.info("Completed")
                );

        Thread.sleep(30000);
    }
}

我希望它输出如下内容：

00:00.000 收到 0
00:00.100 收到 -1
00:00.200 收到 -1
...（更多每 100 毫秒收到 -1 个）
00:01.000 收到 1
00:01.100 收到 -1
00:01.200 收到 -1
...
00:03.000 收到 2
00:03.100 收到 -1
00:03.200 收到 -1
...

但相反，它只收到一次 -1 然后完成。

编辑
希望这个弹珠图可以更容易理解：

Answer 1

我不清楚您想要什么，但是如果您将onResumeNext() 更改为返回，则预期的输出将是

Observable.interval(100, TimeUnit.MILLISECONDS)
  .take(2)
  .map(__ -> -1)