没有onError的SocketException取消订阅请求

Question

我正在尝试使用 RxJava 和 Retrofit(2.3) 发出请求。我希望如果出现任何错误，我可以重试或向客户端显示消息。

但是，我注意到有时我有一个 SocketException 导致不调用 onError，显然请求的订阅者只是取消订阅而不调用其他任何东西（不是 onComplete 也不是 onError）。任何人都知道为什么会发生这种情况，以及如何以通用方式解决这个问题（而不是简单地执行 onUnsubscribe() 并检查 observable 是否没有发送任何 onError 或 onComplete）？

在我的界面上我有这样的东西：

@GET("userInfo")
Observable<List<UserInfo>> getUserInfo(@Header("token") String token);

这就是我创建 observable 的方式：

    public Observable<UserModel> requestUserInfo(final String token) {
        return mService.getUserInfo(token)
                .retryWhen(new RetryWithDelay(HTTP_RETRIES), HTTP_TIME_BETWEEN_RETRIES)))
                .flatMap(new Func1<List<UserInfo>, Observable<UserModel>() {
                    @Override
                    public Observable<UserModel> call(List<UserInfo> userInfo) {
                        return Observable.just(new UserModel(userInfo));
                    }
                });
    }

------ 更新 --------

这就是我在演示者上调用 requestUserInfo 方法的方式

private CompositeSubscription mCompositeSubscription = null;

 public PresenterX(ViewX view) {
    ...
    mCompositeSubscription = new CompositeSubscription();
 }

public void getUserModel() {
       String userToken = new AccessModel().getUserToken();

        mCompositeSubscription.add(mNetworkRequestModel.requestUserInfo(userToken)
                .flatMap(new Func1<UserModel, Observable<UserModel>>() {
                    @Override
                    public Observable<UserModel> call(UserModel userModel) {
                        if (userModel != null) {
                            saveUserModel(userModel); //sync saving
                            return Observable.just(userModel);
                        } else {
                            return Observable.error(new SaveException());
                        }
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Subscriber<UserModel>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "Subscriber was completed")
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "Subscriber called onError")
                        mView.handleErrors(e);
                    }

                    @Override
                    public void onNext(UserModel userModel) {
                        Log.i(TAG, "Subscriber called onNext")
                        mView.populateUserInfo(userModel);
                    }
                }));
    }

    //called by activity when onDestroyMethod is called
    //I assume this is not called as I have other requests running parallelly to this getUserModel() and they are not terminated, despite having other compositeSubscription to manage those
    public void onDestroy(){
       mCompositeSubscription.clear();
    }

由于我有一个 HttpLoggingInterceptor，这是请求突然停止时打印给我的唯一日志。

HttpLoggingInterceptor logging = new HttpLoggingInterceptor(new HttpLoggingInterceptor.Logger() {
            @Override
            public void log(String message) {
                LOG.info(Thread.currentThread(), String.format("%s", message));
        }
    });

这里是日志异常：

I/ExampleApp-ApiClient(21338): : Thread: 1343 |  <-- HTTP FAILED: java.net.SocketException: Socket closed

Answer 1

因为你试图从服务器获取 rx 对象（Observable）。你的功能不正确 移除 Observable

@GET("userInfo")
List<UserInfo> getUserInfo(@Header("token") String token);