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) 我认为这一刻是最困难的。您需要创建your own converter factory 来说明retrofit 如何为您创建响应对象。可能是this question可以帮忙，或者你可以在google找到一些类似的解决方案。