为什么我的 RxJava Observable 没有完成？答案

【问题标题】：Why does my RxJava Observable not complete?为什么我的 RxJava Observable 没有完成？
【发布时间】：2016-05-19 19:11:47
【问题描述】：

我正在尝试测试这个 observable 的登录请求，但订阅者的 onNext 和 onCompleted 方法从未被调用。

return mDeviceRepository.getDeviceId()

            .flatMap(new Func1<String, Observable<LoginResponse>>() {
                @Override
                public Observable<LoginResponse> call(String deviceId) {
                    return mAuthRepository.login(deviceId, mUsername, mHashedPassword);
                }
            })

            .flatMap(new Func1<LoginResponse, Observable<LoginResponse>>() {
                @Override
                public Observable<LoginResponse> call(LoginResponse loginResponse) {
                    mSessionRepository.storeSession(UserSession.fromLoginReponse(loginResponse));
                    mUserRepository.storeUser(loginResponse.getUser());
                    return Observable.just(loginResponse);
                }
            });

我的测试是这样的：

loginUseCase.setCredentials("testuser", "testpassword");
when(mockDeviceRepository.getDeviceId()).thenReturn(Observable.just("testdeviceid"));
when(mockAuthRepository.login("testdeviceid", "testuser", "testpassword")).thenReturn(Observable.just(new LoginResponse()));

testSubscriber = new TestSubscriber();
loginUseCase.buildUseCaseObservable().toBlocking().subscribe(testSubscriber);

testSubscriber.assertCompleted();
testSubscriber.assertValueCount(1);

【问题讨论】：

标签： android rx-java

【解决方案1】：

TestSubscriber 中的 assertTerminalEvent() 而不是 assertCompleted() 呢？您的所有运算符都在同一个线程上运行，因此如果达到 TestSubscriber 断言，则异常会在流中的某处引发

【讨论】：

我没有收到任何 NullPointers。登录方法使用 mockito 存根：when(mockAuthRepository.login("testdeviceid", "testuser", "testpassword")).thenReturn(Observable.just(new LoginResponse()));
您调用 mAuthRepository.login(deviceId, mUsername, mHashedPassword) - 最后两个参数来自哪里？他们究竟是“testuser”、“testpassword”吗？
这些是创建 observable 的类的成员变量。我在存根登录方法之前初始化它们。我相应地编辑了我的问题。
用 assertTerminalEvent() 替换 assertCompleted() 也失败了？
所有有问题的运算符都在同一个线程上运行，因此如果达到 testSubscriber 的断言，则在流中抛出异常

【解决方案2】：

试试

  loginUseCase.buildUseCaseObservable().subscribe(testSubscriber);

PS：如果您想要副作用，请不要在 flatMap 中进行。使用副作用运算符，例如

return mDeviceRepository.getDeviceId()
        .flatMap(new Func1<String, Observable<LoginResponse>>() {
            @Override
            public Observable<LoginResponse> call(String deviceId) {
                return mAuthRepository.login(deviceId, mUsername, mHashedPassword);
            }
        })
        .doOnNext(new Action1<LoginResponse>() {
            @Override
            public void call(String loginResponse) {
                mSessionRepository.storeSession(UserSession.fromLoginReponse(loginResponse));
                mUserRepository.storeUser(loginResponse.getUser());                   
            }           
        })
        .flatMap(new Func1<LoginResponse, Observable<LoginResponse>>() {
            @Override
            public Observable<LoginResponse> call(LoginResponse loginResponse) {

                return Observable.just(loginResponse);
            }
        });

【讨论】：