Observable Vs Completable 为什么没有调用 Completable 任务？

Question

对于 RX Java 新手，我有一个问题。

在我学习 RXJava 这个野兽的过程中，这是我正在测试的课程。

public class PollingLoop {

public static <T> Observable<T> buildObservable(
    final int interval,
    final TimeUnit timeUnit,
    final int maxJitter,
    final Scheduler scheduler,
    final Supplier<Observable<T>> scheduledTask) {

    if (maxJitter <= 0) throw new IllegalArgumentException("Jitter must be greater than 0");

    final Random randomJitter = new Random();
    return Observable.timer(interval, timeUnit, scheduler)
        .map(x -> {
            System.out.println("Flat map jitter");
            return randomJitter.nextInt(maxJitter);
        })
        .flatMap(jitter -> {
            System.out.println("Flat map timer");
            return Observable.timer(jitter, timeUnit, scheduler);
        })
        .flatMap(ignored -> {
            System.out.println("Flat map task");
            return scheduledTask.get();
        })
        .retry()
        .repeat();
}

public static <T> Completable buildCompletable(
    final int interval,
    final TimeUnit timeUnit,
    final int maxJitter,
    final Scheduler scheduler,
    final Supplier<Completable> scheduledTask) {

    if (maxJitter <= 0) throw new IllegalArgumentException("Jitter must be greater than 0");

    final Random randomJitter = new Random();
    return Observable.timer(interval, timeUnit, scheduler)
        .map(x -> {
            System.out.println("Flat map jitter");
            return randomJitter.nextInt(maxJitter);
        })
        .flatMapCompletable(jitter -> {
            System.out.println("Flat map timer");
            return Completable.timer(jitter, timeUnit, scheduler);
        })
        .flatMapCompletable(ignored -> {
            System.out.println("Flat map task that is not called");
            return scheduledTask.get();
        })
        .retry()
        .repeat()
        .toCompletable();
}
}

当我测试 Observable 的执行延迟时，我得到了输出

Flat map jitter
Flat map timer
Flat map task //(observable is being called)

但是当我测试 Completable 的执行延迟时，我得到了输出

Flat map jitter
Flat map timer

//（可完成的任务没有被调用）

我做错了什么？为什么没有从 buildCompletable 中调用 Completable 任务？

这里是测试（用 spock 写的）

def 'should delay execution of observable'() {
    given:
    def subscriber = new TestSubscriber<>()
    def scheduler = new TestScheduler()
    def supplier = Mock Supplier

    supplier.get() >> Observable.just(true)

    when:
    PollingLoop.buildObservable(100, TimeUnit.MILLISECONDS, 1, scheduler, supplier).subscribe(subscriber)
    scheduler.advanceTimeBy(101, TimeUnit.MILLISECONDS)

    then:
    subscriber.assertValueCount(1)
    subscriber.assertValue(true)
}

    def 'should delay execution of completable'(){
    given:
    def subscriber = new TestSubscriber<>()
    def scheduler = new TestScheduler()
    def supplier = Mock Supplier
    supplier.get() >> Completable.complete()

    when:
    PollingLoop.buildCompletable(100, TimeUnit.MILLISECONDS, 1, scheduler, supplier).subscribe(subscriber)

    scheduler.advanceTimeBy(1001, TimeUnit.MILLISECONDS)
enter code here
    then:
    1 * supplier.get()
}

Answer 1

您的第一个flatMapCompletable() 的结果是可完成的，因为这就是您要返回的内容。但是，该可完成组件永远不会发出值（根据定义），因此后续的 flatMapCompletable() 没有值映射。

由于您的第一个 Completable 没有发出值，因此您需要使用 andThen() 运算符或类似的东西来绑定下一步。

您的代码可以编译，因为flatMapCompletable() 运算符的签名为Observable<Long>。您需要将andThen() 运算符放在flatMapCompletable() 函数中。