RxJava -订阅扩展功能

Question

我正在尝试使用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
}

我想要创建一个扩展函数，允许我像使用普通Consumers那样使用Lambda语法。

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>

Answer 1

可能是因为您的NetworkConsumer类扩展了NetworkResponse<T>泛型类型。

您的扩展函数与Observable<T>对象一起工作，该对象要求订阅Consumer<T>对象。当您将NetworkConsumer用于订阅它时，它等于Consumer<NetworkResponse<T>>对象。

因此，这是不匹配的：Consumer<T>与Consumer<NetworkResponse<T>>。您应该像NetworkConsumer<T> : Consumer<T>一样重写您的类。

添加了

如果只对使用者使用NetworkResponse类型很重要，则可以执行以下操作：

1)为NetworkConsumer类添加泛型类型的扩展语句。

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

在这里，T -类型在父类Consumer类中“放置”，这只能是NetworkResponse。和TResponce类型的真实反应，应该放在NetworkResponse中。

2)将所有retrofit api接口方法作为返回NetworkResponse<...>对象。在此之后，不应该出现任何编译错误，但这还不够-- retrofit不知道如何创建NetworkResponse类的对象。

3)我认为这是最困难的时刻。您需要创建你自己的转炉厂来表示retrofit如何为您创建responces对象。也许是这个问题可以帮上忙，或者你可以在谷歌找到一些类似的解决方案。