How to cancel request using merge

Question

Actually I am using RxAndroid, Retrofit2.0 and Okhttp 3.2.0 in my project. I am using Observable.merge to call multiple request in order to upload files. Is there any solution to cancel a unique request ? I know I could use subscription.unsubscribe() but it will cancel all tasks.

I Used okhttp method like client.dispatcher.get(0).cancel() but it throw me an Io Exception like Canceled and it will cancel all tasks.

Do you have a proper way to cancel only a selected request ? Thank you

Answer 1

It's working well. This is the code:

                PublishSubject control = PublishSubject.create();
                publishSubjectMap.put(position, control);
                Observable<Response<Upload>> cancellableRestrofitObservable = dropboxapi.uploadImage(requestBody, params, position)
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .takeUntil(control.asObservable());
                observables.add(cancellableRestrofitObservable);


        Observable<Response<Upload>> mergedObservable = Observable.merge(observables);

//To cancel publishSubjectMap.get(event.getPosition()).onNext("cancel");