[Kotlin] Flow - Latest

Flow의 메소드를 더 알아보자

suspend fun main() = coroutineScope {

    val flow = flow {
        repeat(5) {
            val pancakeIndex = it + 1
            println("Emitter:    Start Cooking Pancake $pancakeIndex")
            delay(100)
            println("Emitter:    Pancake $pancakeIndex ready!")
            emit(pancakeIndex)
        }
    }

    flow.collectLatest {
        println("Collector:  Start eating pancake $it")
        delay(300)
        println("Collector:  Finished eating pancake $it")
    }
}

flow에는 collectLatest라는게 있는데, 가장 최신화된 데이터만 받아오게 된다.

이 함수를 실행하게 된다면

Emitter:    Start Cooking Pancake 1
Emitter:    Pancake 1 ready!
Collector:  Start eating pancake 1
Emitter:    Start Cooking Pancake 2
Emitter:    Pancake 2 ready!
Collector:  Start eating pancake 2
Emitter:    Start Cooking Pancake 3
Emitter:    Pancake 3 ready!
Collector:  Start eating pancake 3
Emitter:    Start Cooking Pancake 4
Emitter:    Pancake 4 ready!
Collector:  Start eating pancake 4
Emitter:    Start Cooking Pancake 5
Emitter:    Pancake 5 ready!
Collector:  Start eating pancake 5
Collector:  Finished eating pancake 5

collect내부에 작업이 수행중임에도 emit되는 순간 새로운 기존의 수행중인 로직을 무시하고 새로 포착된 로직을 수행한다

데이터를 최신화하기 위해서 사용된다.

suspend fun main() = coroutineScope {

    val flow = flow {
        repeat(5) {
            val pancakeIndex = it + 1
            println("Emitter:    Start Cooking Pancake $pancakeIndex")
            delay(100)
            println("Emitter:    Pancake $pancakeIndex ready!")
            emit(pancakeIndex)
        }
    }.mapLatest {
        println("Add topping onto the pancake $it")
        delay(200)
        it
    }

    flow.collect {
        println("Collector:  Start eating pancake $it")
        delay(300)
        println("Collector:  Finished eating pancake $it")
    }
}

mapLatest 메소드를 이용하면 emit된 요소를 변환하기 시작한다.

그런데 작업중 새로운 emit이 들어오게 되면 기존의 작업을 무시하고 최신화된 작업을 수행한다.

이때 변환이 완료된다음 collect에서 포착된다.

suspend fun main() = coroutineScope {

    val flow = flow {
        repeat(5) {
            println("Emitter:    Start Cooking Pancake $it")
            delay(100)
            println("Emitter:    Pancake $it ready!")
            emit(it)
        }
    }.conflate()

    flow.collect {
        println("Collector:  Start eating pancake $it")
        delay(300)
        println("Collector:  Finished eating pancake $it")
    }
}

conflate를 사용하면 데이터 스트림 처리에서 중간의 데이터를 건너뛰고 가장 최근의 데이터만 유지하는 전략을 사용한다.

콜렉터가 아직 처리 중인 데이터가 있을 때 새 데이터가 방출되면, 기다리고 있던 중간 데이터들은 버려지고 가장 최근의 데이터만 유지된다.

즉, 콜렉터가 준비될때 최신 데이터만 전달받아 처리하게 된다.

빠른 데이터 스트림에서 최신 상태를 유지하면서 리소스 사용을 최적화하고 싶을 때 적합한 메소드