问题描述
我在 Spring Boot 上有一个带有 rSocket 服务的服务器:
@MessageMapping("get-messages")
fun getMessageById(): Flow<Set<Message>> {
return flow { emit(service.getLatestMessages()) }
}
因为 repo 不是响应式的,我想定期去数据库中获取数据并将其提供给订阅者(如果存在)。
我想像这样使用 StateFlow:
private val stateFlowMessages = MutableStateFlow<Set<Message>>(emptySet())
init {
Coroutinescope(dispatchers.IO).launch {
while(true){
if (stateFlowProducts.subscriptionCount.value > 0)
stateFlowProducts.value = service.getLatestMessages()
delay(6 * 1000)
}
}
}
但订阅者总是 0,我认为“while”和“delay”不是最佳做法?
解决方法
0. subscriptionCount: `0 1 2 0 2`
1. Map to true/false: `false true true false true`
2. Distinct. : `false true false true`
3. Filter. : ` true true`
3. MapLatest. : ` list list`.
stateFlowProducts.subscriptionCount
.map { it > 0 }
.distinctUntilChanged()
.filter { it }
.mapLatest { service.getLatestMessages() }
.onEach { stateFlowProducts.value = it }
.launchIn(scope)