RxJava2 Completable 没有按预期组合

Question

我的意图是能够通过管道传输一定数量的 Completable 但是，从以下测试用例可以看出，它没有按预期工作。

@Test
public void test_and_then() {

    Completable c1 = Completable.fromAction(() -> {
        System.out.println("<1>");
    });

    Completable c2 = Completable.fromAction(() -> {
        System.out.println("<2.1>");
        boolean test = true;
        if (test) {
            throw new Exception();
        }
        System.out.println("<2.2>");
    });

    Completable c3 = Completable.fromAction(() -> {
        System.out.println("<3>");
    });


    Completable.complete()
            .andThen(c1)
            .andThen(c2)
            .andThen(c3)
            .test()
            .assertError(t -> true)
            .dispose();

    c3.test().assertNotSubscribed();
}

这会导致以下输出。

<1>
<2.1>
<3>
java.lang.AssertionError: Subscribed! (latch = 0, values = 0, errors = 0, completions = 1)

如在 documentation of andThen 中所见，如果在先前的 Completable 中检测到错误，它似乎不应订阅任何稍后添加的 Completables。我的问题是，为什么打印“<3>”，基本上，Completable 是运行。我怎样才能防止这种情况发生并简单地结束可竞争链呢？当文档表明这不应该发生时，为什么它订阅了 Completable c3？

Answer 1

实际上，它如您所愿。但是，您正在订阅 c3:

c3.test().assertNotSubscribed();

这将订阅 c3 并打印 <3>。

如果删除该行，它将按预期工作。

test() 订阅并为您提供一个测试观察器，您可以对其进行断言。不幸的是输出顺序 - 它让你认为流没有停止。

编辑

关于评论 "what's the purpose of assertNotSubscribed if it always fails"。

老实说，这是一个很大的误解。 test() 不是创建 TestObserver 的唯一方法。你可以像这样创建一个测试观察者

TestObserver<Void> testObserver = new TestObserver<>();

您可以使用assertNotSubscribed 来确定您还没有订阅测试观察者。实际上，该方法应该检查 onSubscribe 是否未被调用。这并不是要测试调用了 dispose()。

对于这种情况，您必须执行类似 assertTrue(testObserver.isDisposed()) 的操作。可以关注讨论here.

RxJava2 Completable 没有按预期组合

RxJava2 Completable does not combine as expected

java

reactive-programming

rx-java

reactive

rx-java2