RxJava: AssertError на TestObserver не утверждает true, когда выбрасывается исключение

Question

Я тестирую реальные вызовы API с помощью Retrofit следующим образом:

@Test
public void getList(){
    TestObserver<MyResponse> testObserver = api
            .getResults(params)
            .lift(apiErrorOperator)
            .lift(emptyResponseOperator)
            .test();
    testObserver.awaitTerminalEvent();
    testObserver.assertError(ApiException.class);
}

Проверка не проходит с этими двумя ошибками:

Caused by: java.lang.IllegalStateException: onSubscribe not called in proper order

и

Caused by: com.example.myapplication.repository.ApiException: Search found 0 results

Второе имеет смысл, так как это поведение, которое я ожидаю.Однако я не понимаю, почему testObserver.assertError(ApiException.class) не возвращает true и почему я также получаю первую ошибку.

Для первой ошибки эта строка java.lang.IllegalStateException: onSubscribe not called in proper order выбрасывается в эту строку observer.onError(new ApiException("Search found 0 results")) из emptyResponseOperator.Ниже приведен код для полного класса:

public class EmptyResponseOperator implements ObservableOperator<MyResponse, MyResponse> {
    @Override
    public Observer<? super MyResponse> apply(Observer<? super MyResponse> observer) throws Exception {
        return new DisposableObserver<MyResponse>() {
            @Override
            public void onNext(MyResponse myResponse) {
                if(myResponse.getTotalResultsCount() == 0)
                    observer.onError(new ApiException("Search found 0 results"));
                else{
                    observer.onNext(myResponse);
                    observer.onComplete();
                }
            }

            @Override
            public void onError(Throwable e) {
                observer.onError(e);
            }

            @Override
            public void onComplete() {
                observer.onComplete();
            }
        };
    }
}

А также вот код для класса ApiErrorOperator:

public class ApiErrorOperator<T> implements ObservableOperator<T, Response<T>> {


    @Override
    public Observer<? super Response<T>> apply(Observer<? super T> observer) throws Exception {
        return new DisposableObserver<Response<T>>() {
            @Override
            public void onNext(Response<T> tResponse) {
                if(!tResponse.isSuccessful()){
                    try {
                        if (tResponse.errorBody() != null) {
                            observer.onError(new ApiException(tResponse.errorBody().string()));
                        }else{
                            observer.onError(new ApiException(C.ERROR_UNKNOWN));
                        }
                    } catch (IOException e) {
                        observer.onError(new ApiException(C.ERROR_IO));
                    }
                }
                else if (tResponse.body() == null) {
                    observer.onError(new ApiException(C.ERROR_NOT_FOUND));
                }else{
                    observer.onNext(tResponse.body());
                    observer.onComplete();
                }
            }

            @Override
            public void onError(Throwable e) {
                observer.onError(e);
            }

            @Override
            public void onComplete() {
                observer.onComplete();
            }
        };
    }
}

Answer 1

Мы не рекомендуем писать пользовательское поведение таким образом.Вы должны следовать протоколу Observable, например так:

public class EmptyResponseOperator implements ObservableOperator<MyResponse, MyResponse> {
    @Override
    public Observer<? super MyResponse> apply(Observer<? super MyResponse> observer)
    throws Exception {
        return new DisposableObserver<MyResponse>() {

            // -------------------------------------
            // vvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvvv
            @Override
            public void onStart() {
                observer.onSubscribe(this);
            }
            // ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
            // -------------------------------------

            @Override
            public void onNext(MyResponse myResponse) {
                dispose(); // <-------------------------------------------------------
                if (myResponse.getTotalResultsCount() == 0) {
                    observer.onError(new ApiException("Search found 0 results"));
                } else {
                    observer.onNext(myResponse);
                    observer.onComplete();
                }
            }

            @Override
            public void onError(Throwable e) {
                if (!isDisposed()) {   // <---------------------------------------
                    observer.onError(e);
                }
            }

            @Override
            public void onComplete() {
                if (!isDisposed()) {   // <---------------------------------------
                    observer.onComplete();
                }
            }
        };
    }
}

Answer 2

Ваша реализация неверна, и старайтесь избегать цепных перемещений вниз по течению в вашем коде.

проверьте ниже образец и просмотрите документацию здесь.

 Single.just(1)
    .delaySubscription(Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter e) throws Exception {
            if (!e.isDisposed()) {
                e.onError(new TestException());
            }
        }
    }))
    .test()
    .assertFailure(TestException.class);

- onSubscribe соединяет их и все готово.

Другое решение

создать пользовательский оператор. Как?