java - 在 android 中使用 rxjava 停止长轮询

Question

我正在使用 retrofit 和 rxjava 从 android 客户端对 API 进行long poll。在这种情况下，我们等待来自 API 的 200 或 408 超时响应并处理响应或重新连接以再次等待更多数据。这很好用。我需要阻止 rx 重试某些错误代码(如 500) 或者如果我想中断该过程，例如我的应用程序在后台，所以让我们停止长轮询。

 retrofitInterface.startPolling()  //returns an Observable 
                    .repeat()
                    .retry()
                    .subscribe(new Subscriber<List<Stuff>>() {
                        @Override
                        public void onCompleted() {

                        }

                        @Override
                        public void onError(Throwable e) {

                        }

                        @Override
                        public void onNext(List<Stuff> updates) {
                             //process stuff
                            }

                        }
                    });

我不确定 repeatWhen 和 retryWhen 是否是正确的解决方案，我想继续重复和重试对 API 的 http 调用，但在某些情况下停止重复条件 (比如我将类中的 boolean 值翻转为假) 或者如果状态代码是 500 而不是 408 则停止重试。

Answer 1

如果将请求答案包装在类型为 <Response<?>> 的对象中会更容易，这使您可以控制错误代码。

我为那个用例所做的是在我有一些特定的错误代码时抛出一个特定的异常:

public <T> T throwExceptionIfFailure(T res) {
    Response result = (Response<?>) res;
    if (!result.isSuccessful()) {
        try {
            String msg = result.errorBody().string();
            if (result.code() == 401 || result.code() == 403) {
                invalidateToken();
                msg = context.getString(R.string.invalid_credential);
            } else if (result.code() == 502) {
                msg = context.getString(R.string.server_down);
            }

            throw Exceptions.propagate(new IOException(msg));
        } catch (Throwable e) {
            throw Exceptions.propagate(e);
        }
    } else {
        return res;
    }
}

我在 RX 的 map 函数中添加了这个方法:

    serviceRetrofit.getContacts()
            .map(result -> serviceRetrofit.throwExceptionIfFailure(result))
            .map(result -> createOrUpdateContact(result))
            .retry(4)
            .onErrorReturn(error -> handleErrorEvent(error))
            .doOnCompleted(() -> emitStoreChange(new Store.StoreChangeEvent()))
            .subscribe();