RxJava扩展函数封装网络请求结果实现技术问询

没问题，我来帮你搞定这个RxJava+Retrofit的网络请求结果封装！结合你已经定义好的NetworkConsumer，我们可以给RxJava的流类型（比如Observable或Single）写扩展函数，让你不用每次都手动创建NetworkConsumer子类，直接用lambda处理逻辑，代码会简洁很多。

核心就是利用Kotlin的扩展函数特性，给Observable<NetworkResponse<T>>（或Single<NetworkResponse<T>>）添加自定义订阅方法，内部自动实例化你定义的NetworkConsumer，把lambda回调传递进去，复用你已经写好的成功/失败判断逻辑。

针对Observable的基础扩展 #

如果你的Retrofit接口返回Observable类型，就用这个扩展：

import io.reactivex.rxjava3.core.Observable
import io.reactivex.rxjava3.disposables.Disposable

// 基础版：只处理成功和失败回调
fun <T> Observable<NetworkResponse<T>>.subscribeNetwork(
    onSuccess: (T) -> Unit,
    onFailure: () -> Unit
): Disposable {
    return subscribe(object : NetworkConsumer<T>() {
        override fun onSuccess(data: T) {
            onSuccess.invoke(data)
        }

        override fun onFailure() {
            onFailure.invoke()
        }
    })
}

// 进阶版：如果你的NetworkConsumer还有onError处理Throwable，就加这个重载
fun <T> Observable<NetworkResponse<T>>.subscribeNetwork(
    onSuccess: (T) -> Unit,
    onFailure: () -> Unit,
    onError: (Throwable) -> Unit
): Disposable {
    return subscribe(object : NetworkConsumer<T>() {
        override fun onSuccess(data: T) = onSuccess(data)
        override fun onFailure() = onFailure()
        override fun onError(e: Throwable) = onError(e)
    })
}

针对Single的扩展（如果用Single替代Observable） #

很多时候我们会用Single（只发射一次数据）来做网络请求，对应的扩展函数：

import io.reactivex.rxjava3.core.Single
import io.reactivex.rxjava3.disposables.Disposable

fun <T> Single<NetworkResponse<T>>.subscribeNetwork(
    onSuccess: (T) -> Unit,
    onFailure: () -> Unit
): Disposable {
    return subscribe(object : NetworkConsumer<T>() {
        override fun onSuccess(data: T) = onSuccess(data)
        override fun onFailure() = onFailure()
    })
}

现在你可以直接在Retrofit返回的流上调用这个扩展函数，代码非常清爽：

// 假设你的Retrofit接口是这样的：
// interface ApiService {
//     @GET("user/info")
//     fun getUserInfo(): Observable<NetworkResponse<User>>
// }

apiService.getUserInfo()
    .subscribeNetwork(
        onSuccess = { user ->
            // 处理成功拿到的用户数据
            updateUserProfile(user)
        },
        onFailure = {
            // 处理请求失败（比如HTTP错误、业务逻辑错误）
            showErrorDialog("请求失败，请稍后重试")
        }
    )

如果你的NetworkResponse还包含错误码、错误提示文本这类信息，建议把NetworkConsumer的onFailure方法升级一下，让它能传递错误详情：

// 先修改你的NetworkConsumer抽象类
abstract class NetworkConsumer<T> : Consumer<NetworkResponse<T>> {
    override fun accept(response: NetworkResponse<T>) {
        if (response.isSuccessful()) {
            onSuccess(response.data)
        } else {
            // 把错误信息传递给onFailure
            onFailure(response.errorCode, response.errorMessage)
        }
    }

    abstract fun onSuccess(data: T)
    abstract fun onFailure(errorCode: Int, errorMessage: String)
    // 可选：处理RxJava抛出的异常（比如网络连接异常）
    open fun onError(e: Throwable) {}
}

// 对应的扩展函数也同步更新
fun <T> Observable<NetworkResponse<T>>.subscribeNetwork(
    onSuccess: (T) -> Unit,
    onFailure: (Int, String) -> Unit,
    onError: (Throwable) -> Unit = {}
): Disposable {
    return subscribe(object : NetworkConsumer<T>() {
        override fun onSuccess(data: T) = onSuccess(data)
        override fun onFailure(errorCode: Int, errorMessage: String) = onFailure(errorCode, errorMessage)
        override fun onError(e: Throwable) = onError(e)
    })
}

这样你就能在失败回调里拿到具体的错误信息，做更精准的处理（比如根据错误码显示不同的提示）。

内容的提问来源于stack exchange，提问作者advice