[Kotlin] Coroutines Flow - Intermediate operator

#1 개요

#1-1 중간 연산자 (Intermediate operator)

Asynchronous Flow | Kotlin

중간 연산자에 대해 살펴본다.
 

#1-2 중간 연산자의 구조

중간 연산자는 이름 그대로의 역할을 수행한다. 이름에 있는 '중간'은, emit과 collect 사이 '중간'을 의미한다. 시작점이 emit이고 도착점이 collect인 배수관 속 물의 흐름을 intermediate operator가 하이재킹하는 느낌이라고 보면 된다. 여기서 중요한 점은, 앞으로(=미래에) emit될 데이터를 변형하는 연산자는 결코 아니라는 것이다. 이미(=과거에) emit된 데이터에 특정 연산을 가하여, collect에 전달하는 연산자다. 중간 연산자이니 말이다.
 

#2 Flow에 가용한 중간 연산자 목록

#2-1 베이스 코드

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

fun main() {
    // 1, 2, 3, ...을 방출하는 Flow
    val numberFlow: Flow<Int> = flow {
        for (i in 1..10) {
            delay(100)
            println("emitted: $i")
            emit(i)
        }
    }

    // Flow를 collect
    runBlocking {
        numberFlow.collect { value ->
            println("collected: $value")
        }
    }
}

/* ↑ ↑ ↑ 출력 결과
emitted: 1
collected: 1
emitted: 2
collected: 2
emitted: 3
collected: 3
emitted: 4
collected: 4
emitted: 5
collected: 5
emitted: 6
collected: 6
emitted: 7
collected: 7
emitted: 8
collected: 8
emitted: 9
collected: 9
emitted: 10
collected: 10
*/

먼저, 아무런 중간 연산자를 사용하지 않은 베이스 코드다. 이 코드를 중간 연산자를 이용하는 코드로 수정해보겠다.
 

#2-2 map()

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.runBlocking

fun main() {
    // 1, 2, 3, ...을 방출하는 Flow
    val numberFlow: Flow<Int> = flow {
        for (i in 1..10) {
            delay(100)
            println("emitted: $i")
            emit(i)
        }
    }

    // Flow를 collect
    runBlocking {
        numberFlow
            .map { value -> // 원본 데이터를 변환
                "Hello ${value}!"
            }
            .collect { value ->
                println("collected: $value")
            }
    }
}

/* ↑ ↑ ↑ 출력 결과
emitted: 1
collected: Hello 1!
emitted: 2
collected: Hello 2!
emitted: 3
collected: Hello 3!
emitted: 4
collected: Hello 4!
emitted: 5
collected: Hello 5!
emitted: 6
collected: Hello 6!
emitted: 7
collected: Hello 7!
emitted: 8
collected: Hello 8!
emitted: 9
collected: Hello 9!
emitted: 10
collected: Hello 10!
*/

map()은 emit된 데이터를 변형하여 collect()에 전달한다.
 

#2-3 filter()

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

fun main() {
    // 1, 2, 3, ...을 방출하는 Flow
    val numberFlow: Flow<Int> = flow {
        for (i in 1..10) {
            delay(100)
            println("emitted: $i")
            emit(i)
        }
    }

    // Flow를 collect
    runBlocking {
        numberFlow
            .filter { value -> // 원본 데이터를 규칙에 맞춰 걸러냄
                value % 3 == 0
            }
            .collect { value ->
                println("collected: $value")
            }
    }
}

/* ↑ ↑ ↑ 출력 결과
emitted: 1
emitted: 2
emitted: 3
collected: 3
emitted: 4
emitted: 5
emitted: 6
collected: 6
emitted: 7
emitted: 8
emitted: 9
collected: 9
emitted: 10
*/

filter()는 boolean 연산을 인수로 받는다. 그 연산이 true면 collect에 전달하고, 아니라면 전달하지 않는다.
 

#2-4 기타

Asynchronous Flow | Kotlin

take, toList, toSet, first, single, reduce, fold 등 다른 중간 연산자들에 대한 내용은 위 링크에서 확인할 수 있다.
 

#3 요약

중간 연산자는 하이재킹이다.