Kotlin callbackFlow - 将基于回调机制的 API 转换为 Flow/数据流

介绍

在 Android 中使用协程开发时经常遇到调用的 API 或第三方库是通过回调的机制提供的，而不是 Flow/数据流 API 的方式。比如：通过 callbackFlow 和 ConnectivityManager 以 Flow 的形式监听网络状态变化：

// 通过回调机制 API 监听网络变化
val networkCallback = object : NetworkCallback() {
    override fun onAvailable(network: Network) {
        TODO("A new network ready for use")
    }

    override fun onCapabilitiesChanged(
        network: Network,
        networkCapabilities: NetworkCapabilities
    ) {
        TODO("The Network whose capabilities have changed")
    }

    override fun onLost(network: Network) {
        TODO("The network lost")
    }
}

ContextCompat.getSystemService(context, ConnectivityManager::class.java)?.let {
    it.registerDefaultNetworkCallback(networkCallback)
}

如果我们希望利用 Kotlin Flow 来实现更具声明式的方式来使用回调 API 时，可以通过 callbackFlow 将回调机制的 API 转换为 Flow/流。

在 Kotlin 协程中，callbackFlow 是一个流构建器函数，允许您将基于回调的 API 转换为流。
callbackFlow 生成的冷流，没有接收者时不产生数据。

通过 callbackFlow 监听网络状态

如下以通过 callbackFlow 将 ConnectivityManager 的调用转换为 Flow

package cn.itmob.demo.callbackflow

fun isConnected(): Flow<Boolean> = callbackFlow {
    ContextCompat.getSystemService(context, ConnectivityManager::class.java)?.let { cm ->
        val callback = object : NetworkCallback() {
            override fun onCapabilitiesChanged(
                network: Network,
                networkCapabilities: NetworkCapabilities,
            ) {
                trySendBlocking(networkCapabilities.hasCapability(NET_CAPABILITY_INTERNET))
            }

            override fun onLost(network: Network) {
                trySendBlocking(false)
            }
        }

        cm.registerDefaultNetworkCallback(callback)
        awaitClose { cm.unregisterNetworkCallback(callback) }
    } ?: flowOf(false)
}

本例中，我们创建一个返回 Flow<Boolean> 的函数，以 Flow API 的方式使用 ConnectivityManager，在它的回调函数中将网络变化的数据发送到 Flow 中。

现在可以使用上面的函数以 Flow 的形式获取网络状态，如下所示：

launch {
    isConnected()
    .collect { isConnected ->
        TODO("Network status changed")
    }
}

awaitClose 参数在 Flow 取消收集时或回调式 API 手动调用 SendChannel.close 时调用，通常用于在完成后清理资源，例如注销回调。

为了防止 Flow 收集器被取消时发生内存泄漏，必须使用 awaitClose，否则即使 Flow 收集器已经完成，回调可能仍然会继续运行。