测试 RxJava BehaviorProcessor 是否发出一个值

Question

我无法理解为什么所有这些处理器都通过了测试，但 Behavior 却没有：

package com.example;

import org.junit.Test;

import io.reactivex.Flowable;
import io.reactivex.processors.*;

public class ProcessorTest {
    private static Flowable<String> justFoo() {
        return Flowable.just("foo");
    }

    private static FlowableProcessor<String> subscribeToFoo(
            FlowableProcessor<String> processor) {
        justFoo().subscribe(processor);
        return processor;
    }

    @Test public void flowable() {  // pass
        justFoo().test().assertValue("foo");
    }

    @Test public void async() {  // pass
        subscribeToFoo(AsyncProcessor.create()).test().assertValue("foo");
    }

    @Test public void replay() {  // pass
        subscribeToFoo(ReplayProcessor.create()).test().assertValue("foo");
    }

    @Test public void unicast() {  // pass
        subscribeToFoo(UnicastProcessor.create()).test().assertValue("foo");
    }

    @Test public void behaviorFail() {  // fail
        subscribeToFoo(BehaviorProcessor.create()).test().assertValue("foo");
    }

    @Test public void behaviorPassing() {  // pass
        subscribeToFoo(BehaviorProcessor.create())
                .test()
                .assertNoValues()
                .assertSubscribed()
                .assertComplete()
                .assertNoErrors()
                .assertNoTimeout()
                .assertTerminated();
    }
}

文档说BehaviorProcessor 是：

向每个订阅的订阅者发送它观察到的最新项目和所有后续观察到的项目的处理器。

所以据我了解，它应该通过behaviorFail 测试，而不是behaviorPassing。这是怎么回事？

我如何编写一个有效的测试，以知道 BehaviorProcessor 发出了某个值？

Answer 1

摆脱传递给处理器的终端事件会有所帮助：

@Test public void behavior() {
    final BehaviorProcessor<String> processor = BehaviorProcessor.create();
    justFoo().concatWith(Flowable.never()).subscribe(processor);
    processor.test().assertValue("foo");
}