플로우의 컬렉션 처리는 어떻게 작동할까?

플로우에서 플로우 처리 함수를 사용했을 때, 내부적으론 어떤 프로세스로 흘러가는지 살펴보자.

suspend fun main() {
    flowOf(1, 2, 3)     // 1, 2, 3
        .map { it * it }          // 1, 4, 9
        .collect { println(it) } // 149
}

fun <T, R> Flow<T>.map(transform: suspend (value: T) -> R
): Flow<R> = flow {
	collect { value ->
    	emit(transform(value))
    }
}

fun <T> flowOf(vararg elements: T): Flow<T> = flow {
	for (element in elements) {
    	emit(element)
    }
}

map 뿐만 아니라, 다른 함수를 인라인으로 구현하면 다음과 같은 코드가 된다.

1 suspend fun main() {
2     flow @map {
3         flow @flowOf{
4            for (element in arrayOf(1, 2, 3)) {
5                this@flowOf.emit(element)
6            }
7        }.collect { value ->
8            this@map.emit(value * value)
9        }
10    }.collect {
11        println(it)
12    }
13 }

단계별로 하나씩 분석해 보자면

1. 2번 라인에서(map { it * it }) 플로우를 시작하고, 10번 라인에서 원소들을 모음.
2. 10번라인 collect가 시작될 때, 2번 라인의 @map 람다식을 수행함.
3. 2번라인의 @map 람다식은 또 다른 빌더(3번라인의 @flowOf → flowOf(1, 2, 3))를 호출하고 7번라인의 collect에서 원소를 모음.
4. 7번라인의 collect가 시작될 때, 3번라인의 @flowOf 람다식을 실행함.
5. 첫 번째 값으로 1을 내보내며 7번라인(collect)의 람다식이 실행됨.
6. 8번라인의 this@map.emit이 내보내진 값을 value * value 연산 후 @map 플로우로 내보낸 뒤 10번 라인(collect)의 람다식이 실행됨.
7. 값이(1) 출력된 후 10번 라인의 람다식이 종료되고 8번 라인의 람다식이 재개됨.
8. 8번 라인의 람다식이 끝났으므로(emit후 일시중단 후 재개된 것.), 5번라인의 람다식이 재개됨.
9. 위 과정을 반복함.

위 프로세스를 보기전에는 flowOf() 부터 시작하여 map {it * it }를 거쳐서 1, 4, 9라는 결괏값이 나왔다고 생각했다. (인라인된 코드를 살펴보면 결과는 일단 맞다.)

하지만 플로우 처리의 시작을 map { it * it } 부터 시작한다는 것이다.
map { it * it } 부터 시작을해서 collect가 호출이 되면 flowOf(1, 2, 3)이 실행되고
배열을 순환해서 값을 방출하며, 방출된 값을 this@map.emit(value * value) 변환해서 다시 방출하는 것.

대부분 플로우 처리와 생명주기 함수에서 이와 같은 과정이 일어남.

결론은 종단 함수가 붙어 있는 플로우나, 플로우 연산 함수가 실행되는 곳 부터 플로우가 시작되는 것.