RxJava - Extension Function for Subscribe

Question

I'm trying to use RxJava / Retrofit, and I want to write an extension function to wrap some logic with the network result.

I have a class, NetworkConsumer that extends Consumer.

abstract class NetworkConsumer<T> : Consumer<NetworkResponse<T>> {

    override fun accept(response: NetworkResponse<T>) {
        if (response.isSuccessful()) {
            onSuccess(response.data)
        } else {
            onFailure()
        }
    }

    // other functions such as onSuccess and onFailure
}

I want to create an extension function to allow me to use the Lambda syntax like you can do with normal Consumers.

service.login(email, password)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({
            result -> 
                // result is the correct object
        }, {
           // handle the error
        })

However, if I want to use my NetworkConsumer, I must do:

.subscribe(object : NetworkConsumer<LoginResponse>() {
                    override fun onSuccess(response: LoginResponse) {
                        // ...

So I'm trying to write an extension function, like:

fun <T> Observable<T>.subscribe(onNext: NetworkConsumer<in T>, onError: Consumer<in Throwable>): Disposable {
    return subscribe(onNext, onError, Functions.EMPTY_ACTION, Functions.emptyConsumer<Any>())
}

But it doesn't compile, the error is:

> Type mismatch. 

> Required: Consumer<in T!>! 

> Found: NetworkConsumer<in T>

Have you considered using RxKotlin (https://github.com/ReactiveX/RxKotlin) which is basically just a collection of Kotlin extension methods for RxJava? You may also just look at their code how the subscriber extensions are implemented https://github.com/ReactiveX/RxKotlin/blob/2.x/src/main/kotlin/io/reactivex/rxkotlin/subscribers.kt — Strelok, Mar 20 '18 at 02:34
I've been looking into that, and tried a few things, and while I was able to get their code to work for the extension functions, I'm still unsure how to modify it to get me the result back how I want it. I don't understand how to use my `NetworkConsumer` in that context, and allow me to route all errors to the onError consumer. — advice, Mar 20 '18 at 05:16

Ircover · Answer 1 · 2018-03-21T06:05:48.367

Guess it is because your NetworkConsumer class extends with NetworkResponse<T> generic type.

Your extension function works with Observable<T> object which expects Consumer<T> object for subscribe. When you put your NetworkConsumer for subscribing it equals to Consumer<NetworkResponse<T>> object.

So here is mismatch: Consumer<T> against Consumer<NetworkResponse<T>>. You should rewrite your class like NetworkConsumer<T> : Consumer<T>.

Added

If it is important to use only NetworkResponse types for consumer, you can do following:

1) Add extending statement of generic type for your NetworkConsumer class.

abstract class NetworkConsumer<T : NetworkResponse<TResponce>, TResponce> : Consumer<T> {...

Here T - type to "put" inside parent Consumer class, that can be only NetworkResponse. And TResponce - type of real responce that should be placed in NetworkResponse.

2) Make all of your retrofit api interface methods as returning NetworkResponse<...> object. After this there should not be any compile errors, but it is not enough - retrofit doesn't know how to create objects of your NetworkResponse class.

3) I think this moment would be most difficult. You need to create your own converter factory to say retrofit how to create responces objects for you. May be this question can help, or you can find some similar solutions at google.

Sadly, I cannot. I need to wrap `NetworkResponse` because all the responses from the server return a 200, but contain a status themselves. So I'm wrapping that logic so I can easily get the actual status and payload. — advice, Mar 20 '18 at 18:22
@Advice-Dog I added solution with converter factory to my answer. Wish it can help you. — Ircover, Mar 21 '18 at 06:07

RxJava - Extension Function for Subscribe

1 Answers1