3부 채널과 플로우 - 4

StateFlow

StateFlow의 정의부터 알고 가자!

Kotlin StateFlow는 프로그램에서 상태를 관리하는데 사용되는 Kotlin 라이브러리이다. StateFlow는 여러 코루틴에서 공유될 수 있는 State Holder이다.

Android의 LiveData와 비슷하며 Kotlin의 Flow API에 기반을 두고 있다.
반응형과 Thread-Safe 방식으로 상태 관리하는 것을 아주 쉽게 해준다.
여러 Coroutine이 상태를 관찰할 수 있으며 하나의 Coroutine에서만 상태 변화를 발생시킬 수 있다.

StateFlow는 안정적이고 반응형으로 상태를 관리해야 하는 프로그램에서 유용하다.

StateFlow는 어떻게 구현이 돼 있을까?

우선 알아야할게 StateFlow는 SharedFlow의 한 종류로 상태를 공유하기 위한 상태 객체일 뿐이다.

public interface StateFlow<out T> : SharedFlow<T> {
    /**
     * The current value of this state flow.
     */
    public val value: T
}

// 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

SharedFlow를 StateFlow처럼 쓸려면 위와 같이 사용할 수 있다.
이처럼 StateFlow는 단지 SharedFlow을 그대로 계승 됐다.
- 단 초기 값을 가진다는 것, 단 한 개의 최신 상태 값을 가지는것, resetReplayCache가 지원되지 않는 것이 차이점이다.
결론적으로 StateFlow는 객체의 최신 상태를 얻어오기위한 특수목적 이외에는 SharedFlow를 사용하면 된다.

그럼 StateFlow의 Data는 어떻게 갱신이 되길래 Thread-Safe 하다는 걸까?

우선 emit, value, update 방식이 존재한다.

emit

코드링크

emit()은 call을 wrap하여 value 값을 설정하는 suspend function이다.
그렇기에 해당 function은 CoroutineBuilder 내부에서만 호출할 수 있다.

override suspend fun emit(value: T) {
    this.value = value
}

value

코드링크

@Suppress("UNCHECKED_CAST")
public override var value: T
    get() = NULL.unbox(_state.value)
    set(value) { updateState(null, value ?: NULL) }

NULL.unbox라는 함수를 통해 현재 _state 가 관리되고 있다.
_state 는 atomic으로 set() 내의 updateState() 함수를 통해 계속 변화되고 있다.

value로 업데이트 하는 예시를 들어보자!

_uiState.value = _uiState.value.copy(title = "Something")

uiState value를 copy해서 업데이트 하는 코드이다.
겉보기에는 딱히 문제가 될 것은 없어 보인다.

하지만 copy가 완료 되고 StateFlow의 새 값이 전송될 때까지 다른 Thread가 접근하지 않을 것이라는 보장이 있나?

// Label: Launch A
viewModelScope.launch(Dispatchers.IO) {
    _uiState.value = _uiState.value.copy(doneButtonEnabled = true)
}

// Label: Launch B
viewModelScope.launch(Dispatchers.Default) {
    _uiState.value = _uiState.value.copy(title = "New Title")
}

확실하게 접근을 못한다는 보장은 없다.
그렇기에 해당 방법을 사용할 때 Lock을 걸거나 하는 방법도 존재하지만 우린 더 쉽게 하는 방법을 알고 있다.

update를 활용하자!

public inline fun <T> MutableStateFlow<T>.update(function: (T) -> T) {
    while (true) {
        val prevValue = value
        val nextValue = function(prevValue)
        if (compareAndSet(prevValue, nextValue)) {
            return
        }
    }
}

내부적으로 compareAndSet() 함수 즉 CAS알고리즘을 사용하고 잇다.
현재 상태 value와 이전 value를 저장을 한 뒤 compareAndSet 함수를 실행해서 값의 변경이 있는지 확인한다.
false를 반환한다면 상태를 업데이트할 수 있을 때까지 while 루프를 실행하게 된다!

shardIn 함수 내부

public fun <T> Flow<T>.shareIn(
    scope: CoroutineScope,
    started: SharingStarted,
    replay: Int = 0
): SharedFlow<T> {
    val config = configureSharing(replay)
    val shared = MutableSharedFlow<T>(
        replay = replay,
        extraBufferCapacity = config.extraBufferCapacity,
        onBufferOverflow = config.onBufferOverflow
    )
    @Suppress("UNCHECKED_CAST")
    val job = scope.launchSharing(config.context, config.upstream, shared, started, NO_VALUE as T)
    return ReadonlySharedFlow(shared, job)
}

scope.launchSharing(config.context, config.upstream, shared, started, NO_VALUE as T)에 대해서 살펴보자

코루틴 시작 전략 설정
```
val start = if (started == SharingStarted.Eagerly) CoroutineStart.DEFAULT else CoroutineStart.UNDISPATCHED
```
- started가 Eagerly인 경우, 코루틴은 기본 방식 (DEFAULT)으로 시작합니다. 그렇지 않으면 UNDISPATCHED 방식으로 시작하여 즉시 실행되지 않도록 한다.
코루틴 시작
```
return launch(context, start = start) { // the single coroutine to rule the sharing
```
- 주어진 컨텍스트와 시작 전략을 사용하여 새로운 코루틴을 시작한다.
공유 시작 전략에 따른 데이터 수집 코루틴 내부에서 started 전략에 따라 데이터 수집 방식을 결정한다.

Eagerly 전략

when {
started === SharingStarted.Eagerly -> {
    upstream.collect(shared)
}

Lazily 전략

started === SharingStarted.Lazily -> {
    shared.subscriptionCount.first { it > 0 }
    upstream.collect(shared)
}

기타 및 커스텀 전략


else -> {
    started.command(shared.subscriptionCount)
        .distinctUntilChanged()
        .collectLatest {
            when (it) {
                SharingCommand.START -> upstream.collect(shared)
                SharingCommand.STOP -> { /* do nothing */ }
                SharingCommand.STOP_AND_RESET_REPLAY_CACHE -> {
                    if (initialValue === NO_VALUE) {
                        shared.resetReplayCache()
                    } else {
                        shared.tryEmit(initialValue)
                    }
                }
            }
        }
}
}


다른 및 커스텀 전략은 command를 통해 수신된 명령을 처리한다. 
명령에 따라 다르게 동작한다:

START: 데이터를 수집.
STOP: 아무 작업도 하지 않다.
STOP_AND_RESET_REPLAY_CACHE: 재생 캐시를 초기화. initialValue가 설정된 경우, 초기값으로 캐시를 설정.

Jwhyee / kotlin-coroutine-study

3부 채널과 플로우 - 4 #8