콜드 스트림을 핫 스트림으로 변경하기 (shareIn, stateIn)

MutableSharedFlow를 사용하여 값을 발행하고 다른 한편에서는 이를 수집할 수 있는 수단을 제공한다. 그러나 다른 코드에 있는 콜드 플로우를 가져와서 이 업스트림을 수집하고 SharedFlow로 내보낼 수 있는 편리한 방법도 있다. 바로 ShareIn 연산자를 사용하는 것이다.

ShareIn 연산자

shareIn 연산자는 다음과 같이 생겼다.

fun <T> Flow<T>.shareIn(
    scope: CoroutineScope,
    started: SharingStarted,
    replay: Int = 0,
): SharedFlow<T>

이 연산자를 사용하면, Flow(콜드스트림)를 주어진 코루틴 스코프(scope) 내에서 시작되는 SharedFlow(핫스트림)로 변경한다. 단일로 실행되는 업스트림 플로우에서 발행되는 데이터를 여러개의 다운스트림 구독자들이 공유하는 형태이고, 주어진 replay 값에 따라 새로운 구독자들에게 값을 다시 발행(replay)하게 된다. 

공유하는 코루틴의 시작은 started 매개변수에 의해 제어되며, 다음과 같은 옵션을 지원한다.

• Eagerly : 첫번째 구독자가 생기기도 전에 업스트림이 시작된다. 이 경우 replay 매개변수에 지정된 대로 가장 최근 값을 초과하여 업스트림에서 방출된 모든 값이 즉시 삭제된다.
  
• Lazily : 첫번째 구독자가 생긴 후에 업스트림 플로우가 시작된다. 그러므로 첫번째 구독자는 발행되는 모든 값을 얻을 수 있다는 점을 보장한다. 이후의 구독자들은 단지 최근에 발행 또는 replay되는 값만 얻게 된다. 업스트림 플로우는 구독자가 심지어 사라지더라도 계속해서 활성화되지만, 구독자 없이 가장 최근의 리플레이된 값만 캐시된다.
  
• WhileSubscribed() : 첫번째 구독자가 생긴 후에 업스트림이 시작된다. 마지막 구독자가 사라지면 즉시 멈추게 되며, replay 된 캐시를 영원히 보존하게 된다. 다음과 같은 추가적인 매개변수를 선택적으로 지정할 수 있다.
  stopTimeoutMillis : 마지막 구독자가 사라지는 시점과 공유 코루틴이 중단되는 시점 사이의 지연시간을 지정한다. 기본값은 0으로 즉시 중단된다.
  replayExpirationMillis : 공유 코루틴이 중단되는 시점과 리플레이 캐시가 리셋되는 시점 사이의 지연시간을 지정한다. 기본값은 Long.MAX_VALUE이다. 0으로 지정하면 캐시를 즉시 만료 시키게 된다.
  stopTimeoutMillis 및 replayExpirationMillis를 음수로 지정하는 경우 IllegalArguementException이 발생한다.

shareIn 연산자는 생성 및 유지 관리 비용이 많이 드는 콜드 플로우인데 구독자가 여러개인 상황에서 유용하게 사용할 수 있다.

다음의 예제코드를 살펴보자. 

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

private suspend fun initHeavyLogic() {
    delay(1000)
    println("initHeavyLogic")
}

val coroutineScope = CoroutineScope(Dispatchers.IO)
val counter: Flow<Int> = flow<Int> {
    initHeavyLogic()
    var count = 0
    while(true) {
        emit(count++)
        delay(200)
    }
}
    .shareIn(
        scope = coroutineScope,
        started = SharingStarted.Eagerly
    )

fun main(args: Array<String>) {
    println("Started")
    runBlocking {
        launch {
            println("#1 Launched")
            counter
                .take(10)
                .collect {
                    println("#1:$it")
                }
        }
        launch {
            delay(2000)
            println("#2 Launched")
            counter
                .take(10)
                .collect {
                    println("#2:$it")
                }
        }
    }
    println("Finished")
}

결과:

Started
#1 Launched
initHeavyLogic
#1:0
#1:1
#1:2
#1:3
#1:4
#2 Launched
#1:5
#2:5
#1:6
#2:6
#1:7
#2:7
#1:8
#2:8
#1:9
#2:9
#2:10
#2:11
#2:12
#2:13
#2:14
Finished

• shareIn 연산자를 사용으로 initHeavyLogic이 여러번 호출되지 않고 한번만 호출된다.
• shareIn 연산자를 사용하여 Cold Flow를 Hot Flow로 변경한다. #2가 출력될 때 0이아닌 5부터 출력하는 것을 확인할 수 있다.

StateIn 연산자

StateFlow는 SharedFlow를 확장하는 특수한 버전으로 stateIn 연산자 또한 shareIn 연산자의 특수한 버전으로 생각하면 된다. replay 매개변수가 없고(StateFlow의 경우 항상 1이다), 필수로 지정해야하는 initialValue가 있다. stateIn 연산자의 형태는 다음과 같다.

fun <T> Flow<T>.stateIn(
    scope: CoroutineScope,
    started: SharingStarted,
    initialValue: T
): StateFlow<T>

이 연산자를 사용하면, Flow(콜드스트림)를 주어진 코루틴 스코프에서 시작되는 StateFlow(핫스트림)로 변경한다. 단일로 실행되는 업스트림 플로우에서 가장 최근에 내보낸 값을 여러 다운스트림 구독자들이 공유하는 형태다.

후원하기