여러 Flow 하나로 합치기

복수의 Flow를 합치는 다양한 방법이 있다.

• zip
• combine

zip

• Sequence.zip 확장 함수처럼, Flow는 두 개의 Flow의 값을 결합하는 zip 연산자를 가지고 있음.

fun <T1, T2, R> Flow<T1>.zip(
    other: Flow<T2>, 
    transform: suspend (T1, T2) -> R
): Flow<R>

• 두 개의 Flow를 받아 새로운 Flow 를 리턴함.

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

fun main() = runBlocking<Unit> {
    val nums = (1..3).asFlow()
    val str = flowOf("one", "two", "three")
    nums.zip(str) { a, b ->
        "$a -> $b"
    }.collect {
        println(it)
    }
}

실행결과
1 -> one
2 -> two
3 -> three

zip은 두 플로우증 하나의 플로우가 끝나면, 플로우가 취소됨.

combine

• Flow가 가장 최신의 값이나 연산을 표현할 때, 해당 Flow의 가장 최신 값에 의존적인 연산의 수행을 필요로 하거나, 업 스트림이 새로운 값을 방출(emit)할 때 다시 연산할 수 있음.
  • 해당 연산을 수행하는 연산자의 집합이 combine임.
    • 즉, 여러 Flow의 동적인 변경사항들을 즉시 반영하여 결합 결과를 생성하고 방출함.

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

fun main() = runBlocking<Unit> {
    val nums = (1..3).asFlow().onEach { delay(300) }
    val strs = flowOf("one", "two","three").onEach { delay(400) }
    val startTime = System.currentTimeMillis()

    nums.zip(strs) { a, b ->
        "$a -> $b"
    }.collect {
        println("$it at ${System.currentTimeMillis() - startTime} ms from start")
    }
}

실행결과
1 -> one at 422 ms from start
2 -> two at 828 ms from start
3 -> three at 1235 ms from start

fun <T> Flow<T>.onEach(action: suspend (T) -> Unit): Flow<T>

• onEach 는 다운 스트림으로 방출(emit)되기 전에 지정된 작업을 수행하는 Flow를 리턴함.
• zip 을 사용했을 때 400ms마다 출력되고, 동일한 결과가 생성됨.

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

fun main() = runBlocking<Unit> {
    val nums = (1..3).asFlow().onEach { delay(300) }
    val strs = flowOf("one", "two","three").onEach { delay(400) }
    val startTime = System.currentTimeMillis()

    nums.combine(strs) { a, b ->
        "$a -> $b"
    }.collect {
        println("$it at ${System.currentTimeMillis() - startTime} ms from start")
    }
}

실행결과
1 -> one at 428 ms from start
2 -> one at 634 ms from start
2 -> two at 833 ms from start
3 -> two at 940 ms from start
3 -> three at 1239 ms from start

• combine을 사용한 경우 두 개의 Flow들이 각 방출에 따라 프린트 됨.
  • 두 플로우 중 하나라도 새 값을 방출하면 다른 플로우의 마지막 값을 사용하여 결합함.
  • 그러므로, nums Flow에서 1을 방출해도, 아직 strs Flow가 방출되지 않은 상태임.
    • nums Flow가 one을 방출했을 때, 결합되어 1 -> one at 428 ms from start이 출력됨.
    • 즉, 단일 Flow가 방출해도 결합 대상이 없다면 결합되지 않음.

Flow의 결합을 추적하는 경우엔 combine, 최종 결과만 필요한 경우는 zip을 사용해야 하는 것 같음.
<https://velog.io/@sana/KotlinFlow-Flow-%EA%B2%B0%ED%95%A9-%EC%97%B0%EC%82%B0%EC%9E%90-zip-combine>

Flow를 Flatten하기