在sharedFlow上的collect call永远不会正常complete, 还有Flow.launchIn.
可以配置replay and buffer overflow strategy.
如果subscriber suspend了, sharedflow会suspend这个stream, buffer这个要发射的元素, 等待subscriber resume.
Because onBufferOverflow is set with BufferOverflow.SUSPEND, the flow will suspend until it can deliver the event to all subscribers.
默认参数:
public fun <T> MutableSharedFlow(
replay: Int = 0,
extraBufferCapacity: Int = 0,
onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
)
// MutableStateFlow(initialValue) is a shared flow with the following parameters:
val shared = MutableSharedFlow(
replay = 1,
onBufferOverflow = BufferOverflow.DROP_OLDEST
)
shared.tryEmit(initialValue) // emit the initial value
val state = shared.distinctUntilChanged() // get StateFlow-like behavior
RxJava的等价替代:
PublishSubject -> SharedFlow.
BehaviorSubject -> StateFlow.
Use Flow in Android 发送事件(Event或Effects): SharedFlow
因为SharedFlow没有conflated特性, 所以适合发送事件, 即便值变化得快也是每个都发送.
private val _sharedViewEffects = MutableSharedFlow<SharedViewEffects>() // 1
val sharedViewEffects = _sharedViewEffects.asSharedFlow() // 2
Flow的安全收集
关于收集Flow的方法, 主要还是关注一下生命周期的问题, 因为SharedFlow和StateFlow都是hot的.
在这个文章里有详细的讨论: A safer way to collect flows from Android UIs
在UI层收集的时候注意要用repeatOnLifecycle:
class LatestNewsActivity : AppCompatActivity() {
private val latestNewsViewModel = // getViewModel()
override fun onCreate(savedInstanceState: Bundle?) {
//...
// Start a coroutine in the lifecycle scope
lifecycleScope.launch {
// repeatOnLifecycle launches the block in a new coroutine every time the
// lifecycle is in the STARTED state (or above) and cancels it when it's STOPPED.
repeatOnLifecycle(Lifecycle.State.STARTED) {
// Trigger the flow and start listening for values.
// Note that this happens when lifecycle is STARTED and stops
// collecting when the lifecycle is STOPPED
latestNewsViewModel.uiState.collect { uiState ->
// New value received
when (uiState) {
is LatestNewsUiState.Success -> showFavoriteNews(uiState.news)
is LatestNewsUiState.Error -> showError(uiState.exception)
}
}
}
}
}
}
public suspend fun Lifecycle.repeatOnLifecycle(
state: Lifecycle.State,
block: suspend CoroutineScope.() -> Unit
) {
require(state !== Lifecycle.State.INITIALIZED) {
"repeatOnLifecycle cannot start work with the INITIALIZED lifecycle state."
}
if (currentState === Lifecycle.State.DESTROYED) {
return
}
// This scope is required to preserve context before we move to Dispatchers.Main
coroutineScope {
withContext(Dispatchers.Main.immediate) {
// Check the current state of the lifecycle as the previous check is not guaranteed
// to be done on the main thread.
if (currentState === Lifecycle.State.DESTROYED) return@withContext
// Instance of the running repeating coroutine
var launchedJob: Job? = null
// Registered observer
var observer: LifecycleEventObserver? = null
try {
// Suspend the coroutine until the lifecycle is destroyed or
// the coroutine is cancelled
suspendCancellableCoroutine<Unit> { cont ->
// Lifecycle observers that executes `block` when the lifecycle reaches certain state, and
// cancels when it falls below that state.
val startWorkEvent = Lifecycle.Event.upTo(state)
val cancelWorkEvent = Lifecycle.Event.downFrom(state)
val mutex = Mutex()
observer = LifecycleEventObserver { _, event ->
if (event == startWorkEvent) {
// Launch the repeating work preserving the calling context
launchedJob = this@coroutineScope.launch {
// Mutex makes invocations run serially,
// coroutineScope ensures all child coroutines finish
mutex.withLock {
coroutineScope {
block()
}
}
}
return@LifecycleEventObserver
}
if (event == cancelWorkEvent) {
launchedJob?.cancel()
launchedJob = null
}
if (event == Lifecycle.Event.ON_DESTROY) {
cont.resume(Unit)
}
}
this@repeatOnLifecycle.addObserver(observer as LifecycleEventObserver)
}
} finally {
launchedJob?.cancel()
observer?.let {
this@repeatOnLifecycle.removeObserver(it)
}
}
}
}
}
class LocationRepository(
private val locationDataSource: LocationDataSource,
private val externalScope: CoroutineScope
) {
val locations: Flow<Location> =
locationDataSource.locationsSource.shareIn(externalScope, WhileSubscribed())
}
class LocationRepository(
private val locationDataSource: LocationDataSource,
private val externalScope: CoroutineScope
) {
val locations: Flow<Location> =
locationDataSource.locationsSource.stateIn(externalScope, WhileSubscribed(), EmptyLocation)
}
class UserRepository(
private val userLocalDataSource: UserLocalDataSource,
private val externalScope: CoroutineScope
) {
// DO NOT USE shareIn or stateIn in a function like this.
// It creates a new SharedFlow/StateFlow per invocation which is not reused!
fun getUser(): Flow<User> =
userLocalDataSource.getUser()
.shareIn(externalScope, WhileSubscribed())
// DO USE shareIn or stateIn in a property
val user: Flow<User> =
userLocalDataSource.getUser().shareIn(externalScope, WhileSubscribed())
}
StateFlow使用总结
从ViewModel暴露数据到UI, 用StateFlow的两种方式:
暴露一个StateFlow属性, 用WhileSubscribed加上一个timeout.
class MyViewModel(...) : ViewModel() {
val result = userId.mapLatest { newUserId ->
repository.observeItem(newUserId)
}.stateIn(
scope = viewModelScope,
started = WhileSubscribed(5000),
initialValue = Result.Loading
)
}
移动开发
发布于：2021-08-30 13:22
|
阅读数：448
|
评论：0
QQ好友和群
QQ空间
