flatMapIterable 不会使用 Maybe.error() 恢复发射项目

Question

我有一个要求，我必须在单击按钮时发送已保存的 API 请求。这些 API 请求被添加到一个列表中，如果设备处于离线状态，该列表将保存到 SharedPreferences。设备重新连接后，应单击按钮发送保存的请求。如果其中一个请求的 HTTP 状态代码为 401，则整个过程应该停止。但是，如果出现其他异常，则不应中断该过程，并且应发送列表中的下一个已保存请求。如果请求成功，则将其从已保存请求列表中删除。在流程结束时，任何未发送的请求都将保存到 SharedPreferences。

现在我有一个例外情况，我称之为InvalidRequestException。当遇到此特定错误时，我想从列表中删除该请求，同时我想继续发送列表中的剩余请求。

我从this 帖子中建模了我的代码。这是启动整个过程的方法的代码：

public LiveData<UploadStatus> startUploading() {
MutableLiveData<UploadStatus> uploadStatus = new MutableLiveData<>();

compositeDisposable.add(paramRepository.getSavedOfflineRequest()    // returns Observable<List<Request>>
    .doOnComplete(() -> uploadStatus.setValue(UploadStatus.NO_ITEMS))
    .flatMapIterable( requests -> {
        requestList = requests;
        requestListSizeText.set(Integer.toString(requestList.size()));
        return requestList;
    })                                                              // observable should now be Observable<Request>         
    .flatMapCompletable(this::uploadProcess)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(() ->{
            paramRepository.setOfflineRequestString("");            // clear saved offline requests from shared preferences
            uploadStatus.setValue(UploadStatus.SUCCESS);
        },
        error -> {
            if (error instanceof SessionExpiredException) {
                uploadStatus.setValue(UploadStatus.LOGGED_OUT);
            } else {
                if(!requestList.isEmpty()) {
                    paramRepository.saveRequestsToPrefs(requestList);
                } else { 
                    paramRepository.deleteSavedRequests();
                }
                uploadStatus.setValue(UploadStatus.FAIL);
            }
        }
    )
);

return uploadStatus;

}

保存请求的实际发送发生在uploadProcess。这是我尝试捕捉InvalidRequestException 并删除遇到它的请求的地方：

private Completable uploadProcess(Request request) {

    return apiService.transact(saleUrl, BuildConfig.ApiKey,request)
        .doOnSubscribe(disposable -> {
            uploadAttempts++;
        })
        .toMaybe()
        .onErrorResumeNext(error -> {
            if(error instanceof InvalidRequestException) {
                requestList.remove(request);

                if(requestList.isEmpty()) {
                    return Maybe.error(new OfflineTxnsNotUploadedException());
                }
            }
            else if (error instanceof SessionExpiredException)                // inform UI that session has expired
                return Maybe.error(error);
            else if (requestList.size() == uploadAttempts)  {                 // nothing was uploaded
                return Maybe.error(new OfflineTxnsNotUploadedException());
            }
            return Maybe.empty();
        })
        .flatMapCompletable(response -> {
            requestList.remove(request);
            successCount++;
            successCountText.set(Integer.toString(successCount));
            return createTransaction(request, response);
        });
}

现在，当我对此进行测试时，我发现每当遇到InvalidRequestException 时，整个流都会停止，这不是我想要的行为。我想继续发送列表中的其他请求。我实际上删除了从列表中删除请求的部分（requestList.remove(request);），然后流继续，下一个请求通过apiService.transact()发送。

我是否错误地假设返回 Maybe.empty() 将恢复从 flatMapIterable 发射 Observable<Request>？

编辑：看来我遇到了ConcurrentModificationException，这就是为什么流立即终止并且不发送其他请求的原因。我得先研究一下这个异常。

Answer 1

正如我在编辑中指出的那样，我无法捕捉到 ConcurrentModificationException，因此整个流被中断了。事实上，我正在修改正在发出的List<Request> 到单个Observable<Request>，因为requestList 只是getSavedOfflineRequest 发出的List<Request> 的浅拷贝。

我尝试的一个解决方案是首先通过 Moshi serialize the list，这样未序列化的列表就不会包含对原始列表的任何引用。我很快就发现requestList.remove(request) 将返回false，因为我还没有正确覆盖Request 类的equals() 和hashCode() 方法。最后，我只是决定创建一个“失败的请求”列表，每当遇到错误时将Requests 添加到此列表中。