介绍
在 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 收集器已经完成，回调可能仍然会继续运行。