Почему мой RxJava Observable не завершен?

Question

Я пытаюсь проверить это наблюдаемое для запроса на вход, но подписчики 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); 

Answer 1

Как насчет assertTerminalEvent() вместо assertCompleted() в TestSubscriber? Все ваши операторы работают в одном потоке, поэтому, если утверждения TestSubscriber были достигнуты, тогда исключение было выбрано где-то в потоке

Answer 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); 
      } 
     });