Is it possible in Rx java kotlin to remove timeout or abort observable on success

Question

I have the following code, and when succesfully onNext is received after 5 minutes I am receiving onError as there is timeout, so how can I disable timeout or abort disposable to not call doSomethingElseUnSuccesfull() after doSomethingSuccesfull() is called

getObservable().timeout(5, TimeUnit.MINUTES)
            .onErrorReturnItem(Optional.of(false))
            .distinctUntilChanged()
            .subscribe {
                if (it.get()) {
                    doSomethingSuccesfull()
                    //Remove timer or abort disposable
                } else {
                     doSomethingElseUnSuccesfull()
                   }
            }.also { disposableList.add(it) }

In disposableList = CompositeDisposable() there are several different disposable

Answer 1

I don't know what are you trying to do but if you just need to do a single request, then you should use Single instead of a Observable.

If that is not the case, you can try to use the .doOnNext{} and get your disposable instance in a variable too to dispose it when you need, something like that:

var myDisposable: Disposable? = null
var shouldDispose = false

getObservable().timeout(5, TimeUnit.MINUTES)
            .onErrorReturnItem(Optional.of(false))
            .distinctUntilChanged()
            .doOnNext {
                if (shouldDispose) {
                    myDisposable?.dispose()
                    myDisposable = null
                    shouldDispose = false
                }
            }
            .subscribe {
                if (it.get()) {
                    doSomethingSuccesfull()
                    shouldDispose = true
                } else {
                     doSomethingElseUnSuccesfull()
                   }
            }.also {
              disposableList.add(it)
              myDisposable = it
            }