RxJava — функция расширения для подписки

Я пытаюсь использовать RxJava / Retrofit и хочу написать функцию расширения, чтобы обернуть некоторую логику с сетевым результатом.

У меня есть класс NetworkConsumer, который расширяет 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
}

Я хочу создать функцию расширения, чтобы позволить мне использовать синтаксис Lambda, как вы можете делать с обычным Consumers.

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

Однако, если я хочу использовать свой NetworkConsumer, я должен сделать:

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

Итак, я пытаюсь написать функцию расширения, например:

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

Но не компилируется, ошибка:

> Type mismatch. 

> Required: Consumer<in T!>! 

> Found: NetworkConsumer<in T>