java - 如何在调用代码输入和基于计时器的输入之间切换并使用响应式(Reactive)编程对其进行单元测试

Question

要求

我正在尝试实现一个类来计算完成操作的估计剩余时间。我想将解决方案基于 RxJava - 让类将更新的估计公开为 Observable。

为了这个问题，我将我的解决方案精简到最低限度，放弃了计算和所有其他花里胡哨的东西。

剩下的核心思想是我们正在监听进度更新，但是如果有一段时间没有进展(想象一下大量文件被复制，并且应用程序在一个特别大的文件上“窒息”)，我希望更新继续定期发布，同时等待来自实际来源的新信息。

(显然，在所描述的示例场景中，使用代码会观察到在处理文件时估计值越来越差)。

我还想对解决方案进行单元测试。

简化的、可编译的实现(不工作)

这是我的 Watcher 类(如果是那样的话，显然我不需要两个 Subject，但它反射(reflect)了原始的、更复杂的实现的结构):

public class Watcher {
    public List<String> buffer = new ArrayList<>();
    public BehaviorSubject<String> retriever = BehaviorSubject.create();
    public BehaviorSubject<String> publisher = BehaviorSubject.create();

    public Watcher() {
        retriever
                // when no data is coming, start the timer
                .switchMap(
                        new Func1<String, Observable<String>>() {
                            @Override
                            public Observable<String> call(String s) {
                                return Observable
                                        .interval(1, 1, TimeUnit.SECONDS)
                                        .map(new Func1<Long, String>() {
                                            @Override
                                            public String call(Long tick) {
                                                return "tick " + tick;
                                            }
                                        });
                            }
                        })
                .doOnNext(
                        new Action1<String>() {
                            @Override
                            public void call(String str) {
                                buffer.add(str);
                            }
                        }
                )
                .subscribe(publisher);
    }

    public void add(String str) {
        retriever.onNext(str);
    }

    public Observable<String> asObservable() {
        return publisher.asObservable();
    }
}

测试代码(失败)

下面是使用 TestScheduler 模拟时间流的测试:

TestScheduler testScheduler = new TestScheduler();
TestSubscriber subscriber = new TestSubscriber();

Watcher watcher = new Watcher();
// adding this before the subscription occurs
// i'd also like to ensure that the observer "catches up" on whatever it missed
// - that's why I used PublishSubjects 
watcher.add("A");

watcher
        .asObservable()
        // this bit should get the whole thing running on TestScheduler
        // so that it reacts to artificial shifts of time
        .subscribeOn(testScheduler)
        .subscribe(subscriber);
watcher.add("B");
// this should get the timer going...
testScheduler.advanceTimeBy(5, TimeUnit.SECONDS);

// and here I expect it to get disabled
watcher.add("C");

// over to the ticking timer
testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);

预期与实际结果

我的预期结果将是收到:A、B，几笔报价，C，又是几笔报价。

我得到的都是滴答声!看不到 A、B 或 C。既不在 testScheduler 中，也不在 Watcher 自己的缓冲区中。

注意事项

有趣的是，实际的实现遇到了相反的问题:我得到的是实际输入，而缺少的是滴答。

显然我没有在我的简化版本中准确地重新创建它。不过，该错误似乎具有相似的性质:由于某种原因，在两个来源之间切换并没有像我想象的那样起作用。

问题

我做错了什么？我是否误解了 switchMap 应该如何工作？我是否滥用 TestScheduler 及其虚拟时间安排？

除了我可能犯的任何错误之外，是否有更好、更惯用的替代方法可以完成它？

Answer 1

@Konrad，感谢您提出的详细问题和您发布的答案。我不太确定我是否完全理解您的要求，但在我看来 startWith 应该可以完成这项工作。

此外，如果您询问“Rx idomatic”，最好从 Transformer 派生您的 Watcher。这使得在 Rx 链中重用您的方法变得更加容易。

这是一个完整的代码，它可以实现我认为您想要实现的目标。它是 Java 8，编写 Rx 代码更符合人体工学。

package com.reactive;
import rx.Observable;
import rx.Scheduler;
import rx.schedulers.TestScheduler;
import rx.subjects.BehaviorSubject;
import sun.jvm.hotspot.utilities.Assert;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;


// According to https://github.com/ReactiveX/RxJava/wiki/Implementing-Your-Own-Operators#transformational-operators
// you should use transformers to implement your own observables
class Watcher implements Observable.Transformer<String,String> {

    Scheduler _scheduler;

    // As you already realized in your answer you need to specify the scheduler if you want to control the interval observable
    public Watcher(Scheduler scheduler) {
       _scheduler = scheduler;
    }

    @Override
    public Observable<String> call(Observable<String> retriever) {
        return retriever.switchMap(s ->
                // Create the sequence of ticks
            Observable.interval(1,TimeUnit.SECONDS,_scheduler)
            .map(tick -> "tick " + tick)
            // but prepend the actual signal from the retriever
            .startWith(s)
            );
    }
}


public class Main {

    public static void main(String[] args) {

        TestScheduler testScheduler = new TestScheduler();
        BehaviorSubject<String> retriever = BehaviorSubject.create();

        ArrayList<String> results = new ArrayList<>();
        retriever.compose(new Watcher(testScheduler))
                .subscribe(s->results.add(s));

        retriever.onNext("A");
        retriever.onNext("B");
        testScheduler.advanceTimeBy(3, TimeUnit.SECONDS);
        retriever.onNext("C");
        testScheduler.advanceTimeBy(2, TimeUnit.SECONDS);

        String result = String.join(" , ",results);
        Assert.that(result.equals("A , B , tick 0 , tick 1 , tick 2 , C , tick 0 , tick 1"),result);
    }
}