Flow 연산자

Flow 연산자

Flow 연산자는 flow 파이프라인에서 값을 변환하고 처리하게 해줘요. Kotlin은 두 가지 주요 종류의 flow 연산자를 제공해요.

kotlinx.coroutines 라이브러리는 폭넓은 flow 연산자를 제공하지만, 내장 연산자가 제공하지 않는 동작이 필요하다면 사용자 정의 연산자를 직접 정의할 수도 있어요.

출처: Kotlin 공식 문서

본문

이어지는 섹션들은 대응하는 내장 연산자와 함께 사용자 정의 구현의 예시를 담고 있어요.

중간 연산자

중간 연산자는 업스트림 flow의 값을 소비하는 새 다운스트림 flow를 돌려줘요. 여러 중간 연산자를 연결해서 최종 결과를 수집하기 전에 flow 파이프라인을 만들 수 있어요.

중간 연산자는 목적에 따라 다음 범주로 분류할 수 있어요.

변환 연산자

변환 연산자는 업스트림 flow가 방출한 값을 변환해요. 값의 타입을 바꾸거나, 값을 건너뛰거나, 추가 값을 다운스트림으로 방출하는 데 쓸 수 있어요.

변환 연산자는 중단(suspending) 람다를 받으므로, 그 람다는 각 방출 값을 처리하면서 중단 함수를 호출할 수 있어요. flow 파이프라인이 동시성을 도입하는 연산자를 쓰지 않는다면, 그들은 여전히 값을 순차적으로 처리해요.

.transform() 연산자는 일반적인 변환 연산자로, .map()이나 .filter()처럼 더 구체적인 변환 연산자의 기반으로 쓸 수 있어요.

다음은 .transform() 연산자로 각 업스트림 값을 그 값만큼 여러 번 방출하는 예시예요.

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

// A simplified custom implementation of the default .transform() operator
inline fun <T, R> Flow<T>.myTransform(
    // Accepts a suspending lambda that can emit values downstream
    crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow {
    // Collects values from the upstream flow
    [email protected] { value ->
        // Applies the transformation and emits values to the downstream flow
        [email protected](value)
    }
}

// Uses the default .transform() operator
suspend fun main() = withContext(Dispatchers.Default) {
    val flow = (0..4).asFlow().transform { value ->
        // Emits each value as many times as its value
        repeat(value) {
            emit(value)
        }
    }
    println(flow.toList())
    // [1, 2, 2, 3, 3, 3, 4, 4, 4, 4]
}

.map() 연산자로 각 업스트림 값을 하나의 다운스트림 값으로 변환할 수 있어요.

다음은 .map()으로 각 값을 4배로 곱하는 예시예요.

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

// A simplified custom implementation of the default .map() operator
inline fun <T, R> Flow<T>.myMap(
    crossinline transform: suspend (value: T) -> R
): Flow<R> = transform { value ->
    emit(transform(value))
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Multiplies each upstream value by four
    val flow = (0..4).asFlow().map { it * 4 }
    println(flow.toList())
    // [0, 4, 8, 12, 16]
}
//sampleEnd

조건과 일치하는 업스트림 값만 방출하려면 .filter() 연산자를 쓰세요.

다음은 3으로 나눈 나머지가 1인 값들을 방출하는 예시예요.

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

// A simplified custom implementation of the default .filter() operator
inline fun <T> Flow<T>.myFilter(
    crossinline predicate: suspend (value: T) -> Boolean
): Flow<T> = transform { value ->
    // Emits only values that match the condition
    if (predicate(value))
        emit(value)
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Emits only values where dividing by 3 leaves a remainder of 1
    val flow = (0..10).asFlow().filter { it % 3 == 1 }
    println(flow.toList())
    // [1, 4, 7, 10]
}
//sampleEnd

일부 연산자는 값을 변환하고 조건과 일치하는 결과만 방출해서, .map()과 .filter() 같은 다른 변환 연산자의 동작을 결합할 수 있어요.

예를 들어 .mapNotNull()으로 각 업스트림 값을 변환하고 non-null 결과만 방출할 수 있어요.

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

// A simplified custom implementation of the default .mapNotNull() operator
inline fun <T, R: Any> Flow<T>.myMapNotNull(
    crossinline transform: suspend (value: T) -> R?
): Flow<R> = transform { value ->
    transform(value)?.let { transformed ->
        emit(transformed)
    }
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Converts each string to Double and skips values that can't be converted
    val flow = flowOf("1.2", "10", "11", "error", "0.000")
        .mapNotNull { it.toDoubleOrNull() }
    
    println(flow.toList())
    // [1.2, 10.0, 11.0, 0.0]
}
//sampleEnd

필터링 및 크기 제한 연산자

필터링 및 크기 제한 연산자는 어떤 값이 flow에서 다운스트림으로 계속 갈지 제어해요. 반복되는 연속 값을 제거하거나, flow 시작 부분의 값을 건너뛰거나, 지정된 개수의 값 이후에 수집을 취소하는 데 쓸 수 있어요.

반복되는 연속 값을 무시하려면 .distinctUntilChanged() 연산자를 쓰세요. 이전에 방출된 값과 다를 때만 값을 방출해요.

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

// A simplified custom version of the default .distinctUntilChanged() operator
fun <T> Flow<T>.myDistinctUntilChanged(): Flow<T> = flow {
    var lastEmitted: Any? = Any() // A value that's equal only to itself
    [email protected] { value ->
        if (lastEmitted != value) {
            [email protected](value)
            lastEmitted = value
        }
    }
}

suspend fun main() = withContext(Dispatchers.Default) {
    // Removes repeated consecutive values from the upstream flow
    val flow = flowOf(1, 2, 3, 3, 3, 4, 5, 5, 1).distinctUntilChanged()
    println(flow.toList())
    // [1, 2, 3, 4, 5, 1]
}

.drop() 연산자로 업스트림 flow가 방출하는 첫 값들을 건너뛸 수 있어요. 예를 들어 .drop(2)는 처음 두 값을 건너뛰고 나머지 값을 다운스트림으로 방출해요.

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

// A simplified custom version of the default .drop() operator
fun <T> Flow<T>.myDrop(count: Int): Flow<T> = flow {
    require(count >= 0)
    var elementsAlreadyDropped = 0
    [email protected] { value ->
        if (elementsAlreadyDropped == count) {
            [email protected](value)
        } else {
            ++elementsAlreadyDropped
        }
    }
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Skips the first two values from the upstream flow    
    val flow = flowOf(1, 2, 3, 4, 5).drop(2)
    println(flow.toList())
    // [3, 4, 5]
}
//sampleEnd

고정된 개수의 값 이후에 수집을 취소하려면 .take() 연산자를 쓰세요. 다음은 .take() 연산자로 처음 세 값만 수집하는 예시예요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.*
import java.io.IOException
import kotlin.time.Duration.Companion.milliseconds

// A simplified custom version of the default .take() operator
fun <T> Flow<T>.myTake(count: Int): Flow<T> = flow {
    require(count > 0)
    val cancellationException = CancellationException()
    var elementsRemaining = count
    try {
        [email protected] {
            emit(it)
            --elementsRemaining
            if (elementsRemaining == 0) {
                // Cancels the upstream flow after the requested number of values
                throw cancellationException
            }
        }
    } catch (e: Throwable) {
        if (e === cancellationException) {
            // Handles the CancellationException used to cancel the upstream flow
            // Completes the flow after the set number of values in .myTake()
        } else {
            // Rethrows unexpected exceptions
            throw e
        }
    }
}

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Collects only the first three values from the upstream flow
    val flow = (0..1000).asFlow().take(3)

    println(flow.toList())
    // [0, 1, 2]
}
//sampleEnd

동시 처리 연산자

기본적으로 flow 파이프라인은 값을 순차적으로 처리해요. 업스트림 flow가 값을 방출하면, 다음 값이 방출되기 전에 수집자가 그 값을 처리하죠.

업스트림 flow를 다운스트림 수집과 동시에 실행하려면 동시 처리 연산자로 버퍼(buffer)를 도입하세요. 버퍼는 업스트림 flow가 방출했지만 수집자가 아직 처리하지 않은 값을 저장해요.

이 버퍼를 도입하는 연산자 중 하나가 .buffer() 연산자예요. 버퍼 용량과 버퍼가 가득 찼을 때 어떻게 할지를 설정하게 해줘요. 예를 들어:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
            println("Emitted $it!")
        }
    }
        // Lets the upstream flow emit up to four values ahead of the collector
        .buffer(4)
        .collect {
            println("Processed $it!")
            delay(20.milliseconds)
        }
}
//sampleEnd

수집자가 업스트림 flow보다 느릴 때, 파이프라인은 수집자가 아직 처리하지 않은 값을 다루는 방법을 필요로 해요.

기본적으로 수집자는 업스트림 flow에 *백프레셔(backpressure)*를 적용해요. 이 전략에서 업스트림 flow는 버퍼가 가득 차면 중단되고, 수집자가 공간을 비우면 재개돼요.

업스트림 flow를 중단하는 대신 값을 버리려면 onBufferOverflow 매개변수를 설정하세요.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.BufferOverflow
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
            println("Emitted $it!")
        }
    }
        // Stores up to four values before applying the overflow behavior
        // Drops the oldest buffered value when the buffer is full
        .buffer(4, onBufferOverflow = BufferOverflow.DROP_OLDEST)
        .collect { value ->
            println("Processed $value!")
            delay(20.milliseconds)
        }
}
//sampleEnd

.conflate() 연산자도 쓸 수 있어요. 이는 buffer(1, onBufferOverflow = BufferOverflow.DROP_OLDEST)의 축약형이에요. 이전 값이 수집되는 동안 방출된 값을 건너뛰고 최신 값만 처리하고 싶을 때 쓰세요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
            println("Emitted $it!")
        }
    }.conflate().collect {
        println("Processed $it!")
        delay(20.milliseconds)
    }
}
//sampleEnd

.conflate() 연산자는 수집자가 처리하는 버퍼 값에만 영향을 줘요. 이미 시작된 처리는 취소하지 않죠. 그러려면 대신 collectLatest()를 쓰세요.

앞선 예시들에서 .buffer()와 .conflate() 연산자는 코루틴 컨텍스트를 바꾸지 않고 업스트림 flow를 별도의 코루틴에서 동시에 실행해요.

업스트림 flow를 다른 코루틴 컨텍스트에서 실행하려면 .flowOn() 연산자를 쓰세요. 디스패처가 바뀐다면 .flowOn()은 업스트림 flow를 별도의 코루틴에서 수집하고, 업스트림 방출과 다운스트림 수집 사이에 버퍼를 사용해요.

다음은 .flowOn()으로 업스트림 flow를 Dispatchers.IO에서 실행하는 단순화된 예시예요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    flow {
        repeat(10) {
            emit(it)
        }
        println("Finished emitting!")
    }.flowOn(Dispatchers.IO).collect {
        println("Received $it!")
        delay(10.milliseconds)
    }
}
//sampleEnd

이 예시에서 .flowOn() 연산자는 동시 업스트림 처리를 도입할 수 있지만, 버퍼 동작은 명시적으로 설정되지 않아요.

업스트림 flow의 코루틴 컨텍스트와 버퍼 동작을 모두 설정하려면 .flowOn()을 .buffer()나 .conflate()와 결합하세요. 이 연산자들을 함께 사용하면 *연산자 융합(operator fusion)*이 일어나 단일 버퍼를 공유해요.

다음은 .flowOn(Dispatchers.IO)로 업스트림 flow를 Dispatchers.IO에서 실행하고 .conflate()로 가장 새로운 버퍼 값을 유지하는 예시예요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.Random
import kotlin.time.Duration.Companion.milliseconds
import kotlin.math.round

//sampleStart
fun awaitSensorSignal(): SensorSignal {
    Thread.sleep(10)
    val reading =
        round(Random.nextDouble(25.0, 100.0) * 100.0)/100.0
    println("Measured $reading as the temperature")
    return SensorSignal(temperatureCelsius = reading)
}

data class SensorSignal(
    val temperatureCelsius: Double
)

suspend fun sendLatestTemperature(temperatureCelsius: Double) {
    println("Starting to send $temperatureCelsius...")
    delay(50.milliseconds)
    println("Sent $temperatureCelsius.")
}

suspend fun main() = withContext(Dispatchers.Default) {
    val smartHomeTemperatureFlow = flow {
        while (true) {
            val signal = awaitSensorSignal()
            emit(signal.temperatureCelsius)
            println("Emitted $signal")
        }
    }
        // Runs the upstream flow in Dispatchers.IO
        .flowOn(Dispatchers.IO)
        // Keeps the newest buffered value and drops older ones
        .conflate()
        // Collects the first two values from the upstream flow
        .take(2)
        .collect { temperature ->
            println("Received $temperature!")
            sendLatestTemperature(temperature)
        }
}
//sampleEnd

결합 연산자

결합 연산자는 여러 업스트림 flow에서 값을 소비하고 단일 다운스트림 flow를 돌려줘요. 수집자가 둘 이상의 flow에서 값을 필요로 할 때 써요.

두 업스트림 flow의 값을 짝지으려면 .zip() 연산자를 쓰세요. 각 flow의 첫 번째 값, 그다음 두 번째 값을 결합하는 방식이죠. 결과 flow는 업스트림 flow 중 하나가 완료되는 즉시 완료돼요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.random.Random
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
suspend fun main() = withContext(Dispatchers.Default) {
    // Emits a ticker value every 100 milliseconds
    val tickerFlow = flow {
        while (true) {
            emit(Unit)
            delay(100.milliseconds)
        }
    }

    val start = TimeSource.Monotonic.markNow()
    tickerFlow
        // Combines each ticker emission with the next number
        .zip(flowOf(1, 2, 3)) { _, value ->
            value
        }.collect {
            println("${start.elapsedNow()}: received $it")
        }
}
//sampleEnd

여러 flow의 최신 값을 결합하려면 .combine() 연산자를 쓰세요. 어떤 업스트림 flow가 값을 방출할 때, 각 업스트림 flow의 최신 값을 사용해 새 값을 방출해요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
enum class Theme {
    Dark,
    Light,
}

data class UiState(
    val messages: List<String>,
    val theme: Theme,
)

val messagesFlow = MutableStateFlow(
    listOf(
        "Hello!",
        "Is anyone here?",
    )
)

val themeFlow = MutableStateFlow(
    Theme.Light
)

// Combines the latest values from both upstream flows
val uiStateFlow = combine(messagesFlow, themeFlow) { messages, theme ->
    UiState(messages, theme)
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        // Uses UNDISPATCHED to subscribe before the first update happens
        val uiUpdateJob = launch(start = CoroutineStart.UNDISPATCHED) {
            uiStateFlow.collect {
                // Draws the UI
                println(it)
            }
        }
        messagesFlow.update { messages -> messages + "I'll be back!" }
        delay(100.milliseconds)
        
        themeFlow.value = Theme.Dark
        delay(100.milliseconds)
        
        uiUpdateJob.cancel()
    }
}
//sampleEnd

이 예시에서 combine()은 messagesFlow와 themeFlow의 최신 값으로 uiStateFlow를 만들어요. 어느 업스트림 flow를 업데이트해도 최신 messages와 theme를 가진 새 UiState가 방출돼요.

여러 flow의 값을 동시에 수집해 하나의 다운스트림 flow로 방출하려면 .merge() 연산자를 쓰세요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
interface UiEvent

class ClickEvent: UiEvent

class RightClickEvent: UiEvent

suspend fun main() {
    withContext(Dispatchers.Default) {

        val clickFlow = MutableSharedFlow<ClickEvent>()
        val rightClickFlow = MutableSharedFlow<RightClickEvent>()

        coroutineScope {
            // Uses UNDISPATCHED to subscribe before the first update happens
            val collectJob = launch(start = CoroutineStart.UNDISPATCHED) {

                // Collects both upstream flows concurrently and emits their values downstream
                merge(clickFlow, rightClickFlow).collect {
                    println("Observed an event: $it")
                }
            }
            clickFlow.emit(ClickEvent())
            delay(100.milliseconds)
            
            clickFlow.emit(ClickEvent())
            delay(100.milliseconds)
            
            rightClickFlow.emit(RightClickEvent())
            delay(100.milliseconds)
            
            collectJob.cancel()
        }
    }
}
//sampleEnd

라이프사이클 연산자

라이프사이클 연산자는 flow 수집 중 특정 시점에 실행되는 중단 람다를 받아요. flow가 수집되기 전, 각 값이 방출되기 전, 수집 완료 후, 또는 flow가 값을 방출하지 않고 완료될 때 로직을 배치하는 데 쓸 수 있어요.

.onStart() 연산자는 업스트림 flow가 수집되기 전에 그 람다를 실행해요. 각 방출 값 전에 실행되어야 하는 코드에는 .onEach()를 쓰세요.

.onStart()와 유사하게, 핫 flow용 .onSubscription()을 쓰면 구독자가 flow 수집을 시작한 뒤, 방출된 값을 수집하기 전에 코드를 실행할 수 있어요.

다음은 이 연산자들로 수집이 시작되기 전과 각 값이 다운스트림으로 방출되기 전에 메시지를 출력하는 예시예요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .onStart() operator
fun <T> Flow<T>.myOnStart(
    action: suspend FlowCollector<T>.() -> Unit
): Flow<T> = flow {
    [email protected]()
    [email protected](this@flow)
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf("Page 1", "Page 2", "Page 3").onStart {
            println("Processing pages!")
        }.onEach {
            println("Emitted $it")
        }.collect {
            println("Collected $it")
        }
    }
}

수집 완료 후 코드를 실행하려면 .onCompletion() 연산자를 쓰세요. 그 람다는 업스트림 flow가 성공적으로 완료될 때 다운스트림으로 값을 방출할 수 있어요. 예를 들어:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .onCompletion() operator
fun <T> Flow<T>.myOnCompletion(
    action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit
): Flow<T> = flow {
    var exception: Throwable? = null
    try {
        [email protected](this@flow)
    } catch (e: Throwable) {
        // Run `action`, but if `action` calls `emit`, throw `e` from it
        FlowCollector<T> { throw e }.action(e)
        throw e
    }
    [email protected](null)
}

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf("Page 1", "Page 2", "Page 3").onCompletion {
            println("Almost done...")
            // Emits an additional value after the upstream flow completes
            emit("Last Page!")
        }.collect {
            println("Collected $it")
        }
    }
}
//sampleEnd

업스트림 flow가 값을 방출하지 않고 완료될 때 코드를 실행하려면 .onEmpty() 연산자를 쓰세요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .onEmpty() operator
fun <T> Flow<T>.myOnEmpty(
    action: suspend FlowCollector<T>.() -> Unit
): Flow<T> = flow {
    var emittedSomething = false
    [email protected] { value ->
        emittedSomething = true
        [email protected](value)
    }
    if (!emittedSomething) {
        action()
    }
}

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf("Page 1", "Page 2", "Page 3").onEmpty {
            // Doesn't print anything, because the upstream flow emits values
            println("No pages to load!")
        }.collect()
        flowOf<Int>().onEmpty {
            println("No pages to load!")
            // No pages to load!
        }.collect()
    }
}
//sampleEnd

종단 연산자

종단 연산자는 flow를 수집해요. 방출된 값을 소비하거나, 수집된 값에 기반한 결과를 돌려주거나, 특정 CoroutineScope에서 flow를 수집하는 데 쓸 수 있어요.

flow를 수집하려면 collect() 연산자를 쓰세요. collect()에 람다를 전달하면 각 방출 값을 받아요.

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

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf(1, 2, 3).collect {
            println("Collected $it!")
        }
    }
}
//sampleEnd

collect()를 람다 없이 호출할 수도 있어요.

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

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flowOf(1, 2, 3).onEach {
            println("Collected $it!")
        }.collect()
    }
}
//sampleEnd

flow를 수집하되 새 값이 방출될 때 미완료 작업을 취소하고 싶다면 collectLatest() 연산자를 쓰세요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        flow {
            println("Emitting Page 1")
            emit("Page 1")
            delay(50.milliseconds)
            println("Emitting Page 2 in quick succession")
            emit("Page 2")
            delay(200.milliseconds)
            println("Emitting Page 3")
            emit("Page 3")
        }.flowOn(Dispatchers.IO).collectLatest {
            println("Starting to process $it!")
            try {
                delay(100.milliseconds)
            } catch (e: CancellationException) {
                println("Canceled processing $it.")
                throw e
            }
            println("Done processing!")
        }
    }
}
//sampleEnd

일부 종단 연산자는 flow를 수집하고 수집된 값에 기반한 결과를 돌려줘요. 예를 들어 .first() 연산자로 첫 방출 값을 돌려받고 수집을 취소할 수 있어요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

suspend fun main() { 
    withContext(Dispatchers.Default) {
        val firstValue = flowOf(1, 2, 3).first()

        println(firstValue)
        // 1
    }
}

방출된 값을 컬렉션으로 수집하려면 .toList() 또는 .toSet() 연산자를 쓰세요.

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

// A simplified custom implementation of the default .toList() operator
suspend fun <T> Flow<T>.myToList(): List<T> = buildList {
    [email protected] { value ->
        // Adds each emitted value to the resulting list
        add(value)
    }
}

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        val list = flowOf(1, 2, 3).toList()
        println(list)
        // [1, 2, 3]

        val set = flowOf(1, 2, 2, 3).toSet()
        println(set)
        // [1, 2, 3]
    }
}
//sampleEnd

방출된 값을 단일 결과로 결합하려면 .reduce() 또는 .fold() 연산자를 쓰세요. .fold() 연산자는 여러분이 제공한 값을 시작 값으로 쓰고, .reduce() 연산자는 첫 방출 값을 대신 사용해요.

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

//sampleStart
suspend fun main() {
    withContext(Dispatchers.Default) {
        // Uses the first emitted value as the starting value
        val reduced = flowOf(1, 2, 3).reduce { accumulator, value ->
            accumulator + value
        }

        // Starts with the provided starting value
        val folded = flowOf(1, 2, 3).fold(2) { accumulator, value ->
            accumulator + value
        }

        println(reduced)
        // 6

        println(folded)
        // 8
    }
}
//sampleEnd

특정 CoroutineScope에서 flow 수집하기

화면이나 다른 오래 사는 객체가 flow에서 값을 필요로 할 때는, 그 객체의 CoroutineScope에서 수집자를 시작하세요. 그러면 객체가 파괴될 때 그 CoroutineScope를 취소하는 것이 수집도 취소하게 돼요.

특정 CoroutineScope에서 flow를 수집하려면 .launchIn() 종단 연산자를 쓰세요. 이 연산자는 수집 코루틴의 Job을 돌려줘요.

다음은 화면이 StateFlow에서 값을 수집하고, 화면이 닫힐 때 수집 코루틴을 멈추는 예시예요.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.TimeSource

// A simplified custom version of the default .launchIn() operator
fun <T> Flow<T>.myLaunchIn(scope: CoroutineScope): Job = scope.launch {
    [email protected]()
}

//sampleStart
data class Coordinate(val x: Int, val y: Int)

class MyScreen(val scope: CoroutineScope) {
    private val _mousePosition =
        MutableStateFlow<Coordinate>(Coordinate(0, 0))
    val mousePosition get() = _mousePosition.asStateFlow()

    init {
        // Starts collecting the StateFlow in the screen's CoroutineScope
        mousePosition.onEach {
            updateStatusBar()
        }.launchIn(scope)
    }

    fun moveMouse(newCoordinate: Coordinate) {
        _mousePosition.value = newCoordinate
    }

    private fun updateStatusBar() {
        println("Mouse is at ${_mousePosition.value}")
    }
}

suspend fun main() {
    withContext(Dispatchers.Default) {
        val childScope = CoroutineScope(
            currentCoroutineContext() + Job(currentCoroutineContext()[Job])
        )
        val screen = MyScreen(childScope)
        delay(100.milliseconds)
        
        screen.moveMouse(Coordinate(10, 15))
        delay(100.milliseconds)
        
        screen.moveMouse(Coordinate(1, 3))
        delay(100.milliseconds)
        
        childScope.cancel()
    }
}
//sampleEnd

더 알아보기 (Learn more)