java - 为什么我的 RxJava 订阅在单元测试中丢失而没有被释放

Question

我正在尝试为基于 RxJava 的 View 模型编写单元测试。在生产中它运行良好，但在我的单元测试中，我的一个订阅在没有处理或任何调用的情况下丢失了。

我正在使用 MVI 模式。为了简单起见，所有类(Intent、Action、Result、ViewState)都只有一个文本。

public class ViewModel {
    private static final String TAG = ViewModel.class.getSimpleName();
    private final ISchedulerProvider schedulerProvider;
    private Observable<ViewState> viewState;
    private PublishSubject<Intent> intentEmitter = PublishSubject.create();
    private IActionProcessHolder<Action, Result> actionProcessHolder;

    public ViewModel(ISchedulerProvider schedulerProvider, IActionProcessHolder actionProcessHolder){
        this.schedulerProvider = schedulerProvider;
        this.actionProcessHolder = actionProcessHolder;
        viewState = compose();
    }

    private Observable<ViewState> compose() {
        return intentEmitter
                .compose(getIntentToActionTransformer())
                .compose(actionProcessHolder.getActionProcessor())
                .map(result -> new ViewState(result.getText()));
    }

    public void processIntents(Observable<Intent> intent){
        intent
                .subscribeOn(schedulerProvider.io())
                .observeOn(schedulerProvider.io())
                .subscribe(
                        anIntent -> intentEmitter.onNext(anIntent),
                        error -> Log.e(TAG, error.getMessage())
                );
    }

    public Observable<ViewState> getViewState(){
        return viewState;
    }

    private ObservableTransformer<Intent, Action> getIntentToActionTransformer(){
        return intent ->
                intent.map(theIntent -> new Action(theIntent.getText()));
    }
}

public class ATestScheduler implements ISchedulerProvider {

    @Override
    public Scheduler computation() { return Schedulers.trampoline(); }

    @Override
    public Scheduler io() { return Schedulers.trampoline(); }

    @Override
    public Scheduler ui() { return Schedulers.trampoline(); }
}

public class Intent {
    String text;
    public Intent(String text){ this.text = text; }
    public String getText(){ return text; }
}

public class ActionProcessHolder implements IActionProcessHolder<Action,Result>{
    @Override
    public ObservableTransformer<Action,Result> getActionProcessor(){
        return action -> action.map(theAction -> new Result(theAction.getText()));
    }
}

public class ViewModelTest {

    @Mock
    private ObservableTransformer<Action, Result> actionProcessor;
    @Mock
    private IActionProcessHolder<Action, Result> actionProcessHolder;
    private TestObserver<ViewState> testObserver;
    private ViewModel viewModel;
    private CompositeDisposable subscriptions = new CompositeDisposable();

    @Before
    public void setup(){
        MockitoAnnotations.initMocks(this);
        ISchedulerProvider schedulerProvider = new ATestScheduler();
        testObserver = new TestObserver<>();

        when(actionProcessHolder.getActionProcessor()).thenReturn(actionProcessor);
        when(actionProcessor.apply(any())).thenReturn(Observable.just(new Result("mocked text")));

        viewModel = new ViewModel(schedulerProvider, actionProcessHolder);
        subscriptions.add(viewModel.getViewState()
            .observeOn(schedulerProvider.ui())
            .subscribe(
                    viewState -> testObserver.onNext(viewState)
                    , error -> Assert.fail(error.getMessage())));
    }

    @Test
    public void testProcessIntent(){
        viewModel.processIntents(Observable.just(new Intent("newText")));
        testObserver.assertValueAt(0, viewState -> viewState.getText().equals("mocked text"));
        verify(actionProcessor, times(2)).apply(any());
    }

我期望 actionProcessor 被调用两次，一次是在初始化期间，一次是在实际测试中处理另一个意图时。然而，当意图被处理时， View 模型中的intentEmitter不再有任何订阅者，因此onNext什么也不做。

我已经弄清楚 ViewModel 类中的行:

.compose(actionProcessHolder.getActionProcessor())

似乎打破了它。当我简单地将我的操作映射到这样的结果时:

.map(action -> new Result(action.getText()))

(与actionProcessor 应该做的相同)订阅未释放。我在处理订阅的所有地方都设置了断点，但没有一个被调用。有谁知道为什么会发生这种情况吗？

Answer 1

问题实际上不是订阅被释放，而是订阅链从未建立。在设置过程中，当调用 subscribe 时，使用此行

when(actionProcessor.apply(any())).thenReturn(Observable.just(new Result("mocked text")));

Mockito 立即导致 Observable 发出其值并中断订阅过程。我现在通过模拟 actionProcessor 而不使用 Mockito 解决了这个问题，只需用以下内容替换上面的行:

actionProcessor = action -> action.map(ignore -> new Result("mocked text"));