flatMapIterable does not resume emitting items with Maybe.error()

Question

I have a requirement wherein I have to send saved API requests on a button click. These API requests are added to a list and this list is saved to SharedPreferences if the device is offline. Once the device regains connectivity, the saved requests should be sent on a click of a button. If one of the requests get a HTTP status code of 401, the whole process should stop. However, in case of other Exceptions, the process should not interrupted and the next saved request on the list should be sent. If a request succeeds, it is removed from the list of saved requests. At the end of the process, any requests that remain unsent are saved to SharedPreferences.

Now I have a special case for an Exception that I call InvalidRequestException. I want to remove the request from the list when it encounters this particular error, and at the same time I want to carry on sending the remaining requests in the list.

I modeled my code from this post. Here is the code for the method that kicks off the whole process:

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;

}

The actual sending of saved requests happens in uploadProcess. This is where I attempt to catch the occurrence of InvalidRequestException and delete the request that encounters it:

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);
        });
}

Now when I tested this, I found out that the whole stream stops whenever InvalidRequestException is encountered, which is not the behavior I want. I want to continue sending the other requests in the list. I actually removed the part where the request is removed from the list (requestList.remove(request);), and the stream continued and the next request was sent via apiService.transact().

Am I mistaken in assuming that returning Maybe.empty() would resume the emission of Observable<Request> from the flatMapIterable?

EDIT: It seems I am encountering a ConcurrentModificationException, that's why the stream terminates immediately and the other Requests are not sent. I will have to study this exception first.

score 0 · Accepted Answer · answered Jul 29 '19 at 07:54

As I noted in my edit, I wasn't able to catch the ConcurrentModificationException, thus the entire stream was interrupted. I was in fact, modifying the List<Request> that is being emitted into individual Observable<Request>, since requestList is only a shallow copy of the List<Request> emitted by getSavedOfflineRequest.

One solution I tried was to first serialize the list via Moshi, so that the unserialzied list will not hold any reference to the original list. I found out soon enough that requestList.remove(request) will return false, since I had yet to properly override the equals() and hashCode() method for the Request class. In the end, I just settled to create a "failed requests" list, adding Requests to this list whenever an error is encountered.

flatMapIterable does not resume emitting items with Maybe.error()

1 Answers1