CODEWITHSUNDEEP
androidrx-javakotlin
apouche
apouche

Reputation: 9973

Android RxJava onNext called even when observer is unsubscribed

first of I'm new to Android, but going from the iOS world I used RxSwift extensively and found a behavior which I find very odd.

I encapsulate a retrofit call into a RxJava Observable (I'm not using the RxJavaCallAdapterFactory but I really don't think that's the issue here).

fun SearchService.rx_myObjects(query: String?): Observable<List<MyClass>> {
    return Observable.create { observer ->
        val request = getMyObjects(query = query)
        request.enqueue(object : Callback<MyClass>() {
            override fun onSuccess(result: MyClass) {
                Log.v("TEST", "is unsubscribed ${observer.isUnsubscribed} for query $query")
                observer.onNext(result)
                observer.onCompleted()
            }

            override fun onError(error: APIError) {
                observer.onError(error)
            }
        })
    }
}

And while I suspect the onNext NOT to be called when that observer is unsubscribed this doesn't appear to be the case

08-04 11:50:39.767 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query par
08-04 11:50:44.264 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed true for query pari
08-04 11:50:44.653 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query par
08-04 11:50:46.358 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query pari
08-04 11:50:47.990 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed false for query pari
08-04 11:50:48.033 5107-5107/com.mydomain.myapp V/TEST: is unsubscribed true for query par

as you can see the last call is unsubscribed and (that you can't see here but it's happening because I have the par results instead of the pari ones) the doOnNext closure is executed on my observable even if it's unsubscribed.

I've digged into the internet and found this which seems to describe the same issue except i'm not using the cache at all.

Here's my subscription call:

private fun retrieveMyObjects(query: String?) {
        subscription = service.rx_myObjects(query = query)
            .doOnError { showError(it) }
            .doOnNext { this.airports = it }
            .subscribe()
    }

and the objects call is triggered on a textView update

search_view.setOnQueryTextListener(object : SearchView.OnQueryTextListener {
    override fun onQueryTextChange(newText: String?): Boolean {
        subscription?.unsubscribe()
        retrieveMyObjects(newText)

        return true
    }
})

Any advice ? Am I missing something here ?

Upvotes: 0

Views: 655

Answers (1)

dwursteisen
dwursteisen

Reputation: 11515

In the worst case : You perform a http call. Then, if the text change, you unsubscribe from the previous subscription, then perform another http call. But, the response from the first http call just spawn.

That's why you've got this sort of log orders. As you have unsubscribre from your Observable, it should not emit a new value. For this, you can add a simple if before the onNext / oncompleted block : 

request.enqueue(object : Callback<MyClass>() {
        override fun onSuccess(result: MyClass) {

            if(!observer.isUnsubscribed()) {
                  Log.v("TEST", "is unsubscribed ${observer.isUnsubscribed} for query $query")
                  observer.onNext(result)
                  observer.onCompleted()
             }
        }

        override fun onError(error: APIError) {
            if(!observer.isUnsubscribed()) {
                  observer.onError(error)
            }
        }
    })

Like this, you avoid to emit a value that won't be latter processed (as you unsubscribe from it)

Please note that you're using Observable.create : you may avoid using it as you'll have to manage a lot of things, like Rx Contract, unscubsccription, backpressure. So you may use RxJavaCallAdapterFactory as you have suggested, or give it a try to Observable.fromAsync.

Upvotes: 1

Related Questions