Coroutine Flow

Flow는 데이터 스트림의 일종으로 Coroutine 기반으로 데이터를 수집하는 기능이다.

 

안드로이드 공식문서

• Producer: Flow Builder를 통해 데이터를 생산
• Intermediary: 생산 된 데이터를 중간에서 가공하는 역할 (선택)
• Consumer: 데이터가 넘어올 때마다 수집

Flow의 특징

1. Flow는 데이터 요청 시점부터 발행 시작

fun main() {
    val scope = CoroutineScope(Dispatchers.IO)

    val flow = flow {
        var number = 0
        while (true) {
            delay(1000L)
            emit(number++)
        }
    }

    scope.launch {
        delay(3000L)
        flow.collect {
            println("first collect $it")
        }
    }

    mainTreadRun()
}

• Flow는 선언부터 발행하지 않고 collect가 호출되는 시점부터 발행하기 때문에 Collect하기 전엔 number가 증가하지 않는다.

2. 1대1 관계

fun main() {
    val scope = CoroutineScope(Dispatchers.IO)

    val flow = flow {
        var number = 0
        while (true) {
            delay(1000L)
            emit(number++)
        }
    }

    scope.launch {
        flow.collect {
            println("first collect $it")
        }
    }

    scope.launch {
        delay(3000)
        flow.collect {
            println("second collect: $it")
        }
    }

    mainTreadRun()
}

• First Collect 시점부터 Number 증가
• 3초 뒤 Second Collect 시점에는 Number의 값이 3
• 그러나 Flow는 1대1 관계이기에 Second Collect는 First Collect와는 별개로 다시 0부터 발행한다.

결과

 

이와 같은 특징을 같은 Flow를 Cold Flow라한다.

---

StateFlow

StateFlow의 경우 Hot Flow라 하며, 다음과 같은 특징을 같고 있다.

1. 값을 저장

@OptIn(ExperimentalSubclassOptIn::class)
@SubclassOptInRequired(ExperimentalForInheritanceCoroutinesApi::class)
public interface StateFlow<out T> : SharedFlow<T> {
    /**
     * The current value of this state flow.
     */
    public val value: T
}

StateFlow 내에는 value 값을 갖고 있는데 이는 최신 데이터를 저장한다.

따라서 초기화 시에 초기 값을 지정해주어야 한다.

 

2. 중복 데이터가 넘어올 시 무시

현재 Value와 동일한 값이 Emit 되는 경우에는 무시하는 특성을 갖고 있다.

fun main() {
    val scope = CoroutineScope(Dispatchers.IO)
    val test = MutableStateFlow(0)

    scope.launch {
        launch {
            while(true){
                test.emit(0)
                println("emit 0")
                delay(1000)
            }
        }

        launch {
            test.collect{
                println("StateFlow Collect: $it")
            }
        }
    }

    mainTreadRun()        
}

3. 현재 발행 시점부터 데이터를 수집

StateFlow는 Flow와는 다르게 공유의 개념으로 한 번 발행이 시작되면 수집 시 처음부터 발행하지 않고 최신 발행 데이터부터 수집하게 된다.

fun main() {
    val scope = CoroutineScope(Dispatchers.IO)

    val flow = flow {
        var number = 0
        while (true) {
            delay(1000L)
            emit(number++)
        }
    }

    val stateFlow: StateFlow<Int> = flow.stateIn(
        scope = scope,
        started = SharingStarted.WhileSubscribed(5000L),
        initialValue = 0
    )

    scope.launch {
        stateFlow.collect {
            println("first collect $it")
        }
    }

    scope.launch {
        delay(3000)
        stateFlow.collect {
            println("second collect: $it")
        }
    }

    mainTreadRun()
}

---

SharedFlow

SharedFlow는 따로 값을 저장해두지 않는다.

따라서 emit된 데이터를 바로 보내주며, 이때 데이터 중복 데이터를 포함하여 전송한다.

replay

SharedFlow에는 매개변수로 replay를 지정할 수 있다.

데이터가 빠르게 넘어오다보면 누락되는 데이터가 있기 마련이다.

replay는 이 누락된 데이터를 가장 최근 데이터부터 가져올 수 있다.

fun main() {
    val scope = CoroutineScope(Dispatchers.IO)
    val sharedFlow = MutableSharedFlow<Int>(replay = 0)

    scope.launch {
        launch {
            for(i in 0..10){
                sharedFlow.emit(i)
                delay(1000)
            }
        }

        launch {
            delay(3000)
            sharedFlow.collect {
                println("$it")
            }
        }
    }

    mainTreadRun()
}

위의 경우 3부터 출력되지만 replay를 1로 설정하면 누락된 데이터 중 가장 최근 값인 2부터 출력된다.