RxJava：TestObserver上的AssertError在抛出预测时没有声明为true

Question

我正在使用Retrofit测试真正的API调用，如下所示：

@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将它们连接起来然后你就去了。

另一种方法

创建自定义运算符How?