为什么我在单元测试中的 RxJava 订阅丢失而没有被处置

Question

我正在尝试为基于RxJava的视图模型编写单元测试。在生产中它运行良好，但在我的单元测试中，我的一个订阅丢失了，没有释放或调用任何内容。

我正在使用 MVI 模式。出于简单原因，所有类(意向、操作、结果、视图状态(都只有一个文本。

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 被调用两次，一次是在初始化期间，一次是在实际测试中处理另一个意图时。但是，当处理意图时，视图模型中的 IntentEmitter 不再有任何订阅者，因此 Next 不执行任何操作。

我已经计算出ViewModel类中的行：

.compose(actionProcessHolder.getActionProcessor(((

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

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

(((

(与操作处理器应执行的操作相同(不会释放订阅。 我在释放订阅的所有位置都设置了断点，但从未调用过它们。 有谁知道为什么会这样？

Answer 1

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

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

Mockito立即使Observable发出其价值并破坏了订阅过程。 我现在通过在不使用 Mockito 的情况下模拟 actionProcessor 来解决它，只需将上面的行替换为以下内容：

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