[Kotlin/Android] SharedFlow의 속성 파해치기

SharedFlow

SharedFlow는 Hot Flow로, SharedFlow를 수집하는 다수의 구독자가 동시에 같은 데이터를 수신할 수 있고, 한번에 여러 값을 발행할 수 있다는 특징이 있다.

Hot Flow, Cold Flow

기존의 Flow는 Cold Flow이다. Cold Flow는 구독자가 값을 수집하기 시작할 때 데이터를 생성하며, 하나의 구독자만 존재한다. Flow를 collect를 할때마다 flow의 block이 새롭게 실행되며, 이전 구독과는 독립적이다. (Cd 플레이어를 생각해보면 재생 버튼을 누를 때 음악이 나온다.)

Hot Flow는 구독자가 값을 수집하는지 여부와 상관 없이 데이터를 생산한다. Flow를 collect 하면 그 당시 방출되는 값부터 수집한다. (라디오를 생각해보면 라디오는 계속 흘러나오고, 내가 주파수를 맞추면 그 때부터 들려오는 라디오를 청취할 수 있다.)

그럼 SharedFlow 사용 방법과 속성에 대해서 알아보자.

val sharedFlow = MutableSharedFlow<Int>(
        replay = 1,
        extraBufferCapacity = 3,
        onBufferOverflow = BufferOverflow.SUSPEND
    )

SharedFlow는 데이터 방출 시 내부 버퍼를 사용하여 데이터를 관리한다. 버퍼의 동작은 다음 세 가지 주요 옵션에 의해 결정된다.

• replay
  구독자가 flow를 수집했을 당시에 과거에 방출된 데이터를 몇 개까지 재전송할지 결정.
  기본값: 0 (과거 데이터를 재전송하지 않음).

val sharedFlow = MutableSharedFlow<Int>(replay = 2)

runBlocking {
    sharedFlow.emit(1)
    sharedFlow.emit(2)
    sharedFlow.emit(3)

    sharedFlow.collect { value ->
        println("Collected: $value") // 2, 3 출력
    }
}

replay = 2 설정으로 인해 구독자는 마지막 두 개의 데이터(2, 3)를 수신.

• extraBufferCapacity
  replay 외에 추가로 버퍼에 저장할 수 있는 데이터 개수.
  기본적으로 replay에 정의된 데이터가 버퍼의 일부를 차지하고, extraBufferCapacity는 그 외 추가 공간을 제공한다.
  기본값: 0 (추가 버퍼 없음).

val sharedFlow = MutableSharedFlow<Int>(replay = 1, extraBufferCapacity = 1)

runBlocking {
    val success1 = sharedFlow.tryEmit(1) // 성공
    val success2 = sharedFlow.tryEmit(2) // 성공
    val success3 = sharedFlow.tryEmit(3) // 실패
    println("Emit results: $success1, $success2, $success3") // true, true, false
}

replay = 1 : 한 개의 데이터는 항상 재전송 가능.
extraBufferCapacity = 1 : 추가 공간 한 개 제공.
총 버퍼 용량 = replay + extraBufferCapacity = 2.

• onBufferOverflow
  버퍼가 가득 찼을 때 어떻게 처리할지 결정.

BufferOverflow.SUSPEND (기본값): 대기하여 버퍼에 공간이 생길 때까지 기다림.
BufferOverflow.DROP_OLDEST: 가장 오래된 데이터를 삭제하고 새 데이터를 저장.
BufferOverflow.DROP_LATEST: 가장 최근 데이터를 삭제하고 새 데이터를 저장하지 않음.

val sharedFlow = MutableSharedFlow<Int>(
    replay = 1,
    extraBufferCapacity = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST
)

runBlocking {
    sharedFlow.emit(1) // replay에 저장
    sharedFlow.emit(2) // extraBufferCapacity에 저장
    sharedFlow.emit(3) // 오래된 데이터 삭제 후 저장

    sharedFlow.collect { value ->
        println("Collected: $value") // 2, 3 출력
    }
}

처음 방출된 데이터(1)는 삭제되고, 나중 데이터(2, 3)가 유지된다.

SharedFlow에 데이터를 발행하는 방법을 알아보자.

fun emitFlow() {
	_sharedFlow.tryEmit(1)
}

suspend fun emitFlow() {
	_sharedFlow.emit(1)
}

두 함수의 차이는 suspend 유무이다.
tryEmit은 중단 함수가 아니며, 일반 함수에서 flow에 값을 방출할 수 있다.
하지만 이렇게만 인지하고 있으면 예상하지 못하는 상황이 발생할 수 있다.

public fun tryEmit(value: T): Boolean

tryEmit은 Boolean 값을 반환한다. 값이 방출 성공되면 true, 실패하면 false를 반환한다.
그러면 값이 방출 실패되는 경우는 무엇일까?

val sharedFlow = MutableSharedFlow<Int>(
    replay = 0,
    extraBufferCapacity = 1, 
    onBufferOverflow = BufferOverflow.SUSPEND
)

runBlocking {
    val success = sharedFlow.tryEmit(1)
    println("Try emit success: $success") // true (성공 시)
    val failed = sharedFlow.tryEmit(2)
    println("Try emit success: $failed") // false (버퍼가 가득 찬 경우)
}


Try emit success: true
Try emit success: false

두번째 tryEmit 함수의 값은 false를 반환한다.
현재 구독자가 없는 상태이기 때문에 값이 바로 사용되지 못하고 위에 설명한 extraBufferCapacity라는 SharedFlow의 버퍼 공간에 이미 값 1이 저장되어 있어, 값 2는 정상적으로 방출되지 못한 것이다.

emit() 함수를 사용할 때 주의점도 존재한다.
onBufferOverflow의 기본값은 BufferOverflow.SUSPEND이며, emit() 중단 함수를 호출할 때 버퍼가 가득 찬 경우 버퍼에 공간이 생길 때까지 대기하게 되어, 호출한 코루틴이 정지 상태로 유지되는 상황이 발생할 수 있다.