How to stop observing old RxJava response statuses without dispose?

Question

I have such case. User starts chain of request by clicking a button. Then he will make two photos uploading. However after starting uploading photos, he might return and start process over.

My CompositeDisposable() is attached to the viewModel it will only get be cleared after onCleared(). This is why strange issue occurs: user might start uploading photos, go back, start again and responses from old requests will be delivered, before new ones will be uploaded!

How should I modify all my regular RxJava requests and zip operator to look only after requests which are new, not old ones.

Again, I can't call CompositeDisposable.dispose(), before each button event, because that would terminate upload process.

I need to only dispose possible old responses.

​

Here is my sample:

//called two times, for uploading
fun uploadPhoto(){
compositeDisposable.add(
            apiService.networkRequest(linkedHashMap, url)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeWith(object: DisposableSingleObserver<retrofit2.Response<String>>() {
                        override fun onSuccess(t: retrofit2.Response<String>) {
                            // will provide result even if two new uploadPhoto() methods gets called
                            handleResponse(t)
                        }

                        override fun onError(e: Throwable) {

                        }
                    }))

}
}

​

fun handleResponse(response: retrofit2.Response<String>)
{
    responseList.add(response) //saves number of responses
    if(responseList.size == 2)
    {
        //calls new network request and creates a new logic. 
    }
}

The problem is that handleResponse() gets called, after uploadPhoto() returns previous result

Answer 1

OK, if I understood correctly your case, you want to discard the response from the first upload and take in consideration the response from the second, or to generalize: ignore any previous responses and take in consideration only the latest.

If this is so then one simple solution would be to check the compositeDisposable every time before starting a new upload. If the list is not empty then discard everything and add the new disposable to it.

Something like this:

fun uploadPhoto(){
   if(compositeDisposable.size() > 0){
       compositeDisposable.clear()
   }

   // ...
}

Pay attention to use compositeDisposable.clear() not .dispose().

Regarding your follow up question:

So, calling compositeDisposable.clear() will dispose each and every item in the list, more specifically this means that the working thread will be interrupted and, yes, in your case it means the upload process will be terminated.

If you want the upload to continue, then you will have to come up with a different mechanism other than clearing the disposables.

I am not sure if you could do that in Rx, but one idea not involving Rx would be to have some kind of uploadId, like a random hash generated, associated with each upload. This id should be given to your networking layer and then passed back in the response.

Then in the ViewModel you would keep track of the currentUploadId, and:

• whenever the user performs a new upload, update currentUploadId with new generated id
• whenever handleResponse(...) is received, you check response.uploadId with currentUploadId, if they don't match then you simply discard this response.