我需要运行一个任务,它会发出一些数据。我想订阅这些数据,例如 PublishSubject。但是我无法解决单实例流的问题。如果我再次尝试调用它,它将创建另一个实例,并且该工作将完成两次。我尝试在内部运行流程并将值发布到 BroadcastChannel,但这个解决方案似乎不正确。这种任务的最佳实践是什么?
问问题
53 次
1 回答
0
这会变魔术:
fun <T> Flow<T>.refCount(capacity: Int = Channel.CONFLATED, dispatcher: CoroutineDispatcher = Dispatchers.Default): Flow<T> {
class Context(var counter: Int) {
lateinit var job: Job
lateinit var channel: BroadcastChannel<T>
}
val context = Context(0)
fun lock() = synchronized(context) {
if (++context.counter > 1) {
return@synchronized
}
context.channel = BroadcastChannel(capacity)
context.job = GlobalScope.async(dispatcher) {
try {
collect { context.channel.offer(it) }
} catch (e: Exception) {
context.channel.close(e)
}
}
}
fun unlock() = synchronized(context) {
if (--context.counter == 0) {
context.job.cancel()
}
}
return flow {
lock()
try {
emitAll(context.channel.openSubscription())
} finally {
unlock()
}
}
}
于 2020-07-15T13:05:07.273 回答