코틀린의 반응형 프로그래밍 #3 - SharedFlow

 

이전 포스팅에서 Flow에 대하여 살펴보았습니다. 이번 포스팅에서는 Flow의 child에 속하는 SharedFlow에 대하여 살펴보려고 합니다. 이 타입은 FRP(Functional Reactive Programming) 스타일로 프로그래밍을 할때 더욱 많은 가능성을 열어줍니다. 이전 포스팅에서 다루었던 많은 내용들이 이 포스팅에서 이어지겠지만, SharedFlow는 Flow와 비교할 때 몇가지 주요 차이점이 있습니다. 아래 다이어그램을 바로 이해하지 못하더라도 본 게시글을 모두 읽어 보았다면 충분히 이해 할 수 있을것으로 기대합니다.

 

https://elizarov.medium.com/shared-flows-broadcast-channels-899b675e805c

 

Hot Flow

첫 번째이자 가장 중요한 차이점은 SharedFlow가 콜드(cold)가 아니라 핫(hot) 스트림이라는 점입니다. 이는 SharedFlow에서 컬렉션이 발생하지 않더라도 Flow가 여전히 값을 내보내고 있음을 의미합니다. 이는 또한 SharedFlow가 구독될 때 활성화된 인스턴스가 반환됨을 의미합니다. 이를 통해 여러 구독자가 동일한 Flow에서 값을 받을 수 있습니다. 이는 구독될 때마다 새로운 인스턴스를 생성하는 콜드 Flow와는 다릅니다.

 

 

값 전달(Sending Values)

내부적으로 데이터를 생성하는 Flow와 달리 SharedFlow(특히 MutableSharedFlow)는 외부에 기록됩니다. 이 작업은 suspending emit 함수 또는 non-suspending tryEmit 함수를 사용하여 수행됩니다. 중단 함수 버전은 버퍼에 값이 추가될 때까지 일시 중단되지만, non-suspending 함수 버전은 대신 버퍼에 값을 추가하려고 시도하고 불가능한 경우 false를 반환합니다.

 

 

재생 및 버퍼(Replay and Buffer)

핫 Flow를 사용하면 새로운 구독자에게 값을 전송하여 새로운 값이 도착할 때까지 기다릴 필요가 없습니다. 이 작업은 마지막 X값을 새로운 가입자에게 다시 보내는 재생 캐시(replay cache)를 사용하여수행됩니다. 기본적으로 이 값은 랑데부 채널(Rendezvous Channel)과 동일한 0입니다.

여러 구독자의 특성상 일부는 다른 구독자보다 빠르게 방출된 값을 소비할 수 있습니다. 모든 구독자를 차단하지 않도록 SharedFlow에는 버퍼가 있습니다. 기본적으로 이 버퍼는 재생 캐시 크기이지만 extraBufferCapacity 매개 변수를 사용하여 확장할 수 있습니다. 또한 buffer overflow정책을 설정할 수 있습니다. 기본적으로 오버 플로우는 송시자/발신자를 일시 중단시키지만, 일시 중단을 방지하기 위해 다른 정책을 선택하여 버퍼에서 값을 삭제할 수도 있습니다. 일시 중단이 발생하면 UI가 멈추거나 즉시 실행해야할 다른 로직의 수행이 늦어질수 있습니다.

 

 

SharedFlow 사용

SharedFlow를 만들기 위하여, MutableSharedFlow를 만들고 asSharedFlow()를 사용하여 SharedFlow로 변환하거나 콜드 Flow를 핫 SharedFlow로 변환하기 위하여 shareIn()을 사용합니다.

private fun convert_cold_to_hot_flow() {
    // 1. asSharedFlow를 사용하여 Hot Flow로 변환
    val mutableSharedFlow = MutableSharedFlow<Int>()
    val sharedFlow = mutableSharedFlow.asSharedFlow()

    // 2. shareIn을 사용하여 Hot Flow로 변환
    val myFlow = flowOf(1, 2, 3)
    val flowSharedFlow = myFlow.shareIn(
        scope = GlobalScope,
        started = SharingStarted.Lazily,
        replay = 3, // 디폴트 값은 0
    )
}

 

SharedFlow를 수집하려면 Flow에서와 같이 터미널 연산자를 적용해야 합니다.

val sharedFlow = MutableSharedFlow<Int>().asSharedFlow()

// 데이터 수집하기 위하여 collect 터미널 연산자를 적용한 구독자1
val subscriber1 = sharedFlow.collect { ... }

// 첫번째 데이터를 가져오기위하여 first 터미널 연산자를 적용한 구독자2
val subscriber2 = sharedFlow.first()

 

마지막으로 MutableStateFlow로 값을 보내려면 emit 또는 tryEmit을 사용 할수 있습니다.

val sharedFlow = MutableSharedFlow<Int>()

coroutineScope.launch {
  sharedFlow.emit(1)
}

sharedFlow.tryEmit(2)

 

 

결론

세 가지 Flow 타입 중 두 가지를 다루었습니다. 단일 구독자가 콜드 스트림의 데이터를 구독할 수 있도록 하는 Flow, 여러 구독자가 핫 스트림의 데이터를 구독할 수 있는 SharedFlow가 바로 그것입니다. 다음 포스팅에서는 이 시리즈를 마무리하기 위해 훨씬 더 전문화된 SharedFlow의 하위 프로그램인 StateFlow를 다룰 것입니다.

 

---