Flow를 공부해보자 2편

부제 : Flow 공식문서 제대로 읽기

오늘의 목표

스터디를 하기 위해... 오늘도 영어를 읽어본다 (으아악..)

참고 자료

공식문서

Asynchronous Flow

일시 중단 함수는 하나의 값을 비동기적으로 반환한다.
비동기적으로 계산된 여러 값을 반환하려면 어떻게 해야 할까?
-> 이때 Kotlin Flows가 등장한다.
(내 말투 아니고.. 공식문서 말투입니다.)

Representing multiple values

Sequences

sequence는 eager(즉시) evaluation을 하는 collection과는 달리 lazy evaluation으로 처리한다. Sequence에서 최종 연산이 호출될 때, 모든 layz 연산들이 수행된다.

multi value를 sequence를 사용하여 출력하고자 한다면 다음과 같이 작성할 수 있다.

fun simple(): Sequence<Int> = sequence { // sequence builder
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it
        yield(i) // yield next value
    }
}

fun main() {
    simple().forEach { value -> println(value) } 
}

Suspending functions

sequence를 활용한 연산은 main thread를 block한다.
이 값들이 suspend를 사용하여 비동기적으로 계산하면, blocking 없이 작업을 수행할 수 있다.

suspend fun simple(): List<Int> {
    delay(1000) // pretend we are doing something asynchronous here
    return listOf(1, 2, 3)
}

fun main() = runBlocking<Unit> {
    simple().forEach { value -> println(value) } 
}

Flows

반환 타입으로 List<Int>를 가진 다는 것은 값을 한번에만 반환할 수 있음을 의미한다. 비동기적으로 계산되는 값들의 stream을 표현하기 위해서는, Flow<Int> 타입을 사용할 수 있다. (Sequence<Int>와 같이 비동기 적으로 동작함)

fun simple(): Flow<Int> = flow { // flow builder
    for (i in 1..3) {
        delay(100) // pretend we are doing something useful here
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> {
    // Launch a concurrent coroutine to check if the main thread is blocked
    launch {
        for (k in 1..3) {
            println("I'm not blocked $k")
            delay(100)
        }
    }
    // Collect the flow
    simple().collect { value -> println(value) } 
}

Flow를 설명하기 위해 여러 예제를 가져다 붙인 것 같은데..
본격적으로 Flow를 파헤쳐보자

Flows are cold

Flow는 기본적으로 cold stream이다. (sequence와 유사)
flow builder 내부적으로는 collect가 되기 전까지 작업을 수행하지 않는다.

fun simple(): Flow<Int> = flow { 
    println("Flow started")
    for (i in 1..3) {
        delay(100)
        emit(i)
    }
}

fun main() = runBlocking<Unit> {
    println("Calling simple function...")
    val flow = simple()
    println("Calling collect...")
    flow.collect { value -> println(value) } 
    println("Calling collect again...")
    flow.collect { value -> println(value) } 
}

위의 코드는 simple()함수가 suspend가 아니기 때문에 호출 자체는 빠르게 반환되며, 아무것도 기다리지 않는다. flow는 수집될 때마다 새로 시작된다.

Flow cancellation basics

Flow는 Coroutine의 일반적인 취소를 준수한다.
취소가능한 suspending 함수 안에서 flow가 취소되면 collect를 취소할 수 있다..

아래 코드는 withTimeoutOrNull 블록에서 실행 중일 때 타임아웃이 발생하면 flow가 취소되고 코드 실행이 중지되는지를 보여준다.

fun simple(): Flow<Int> = flow { 
    for (i in 1..3) {
        delay(100)          
        println("Emitting $i")
        emit(i)
    }
}

fun main() = runBlocking<Unit> {
    withTimeoutOrNull(250) { // Timeout after 250ms 
        simple().collect { value -> println(value) } 
    }
    println("Done")
}

결과를 예측한대로

Emitting 1
1
Emitting 2
2
Done

Flow builders

아래와 같은 것들이 있다. 딱 보면 아시겠쥬?

flow { ... }
flowOf
asFlow()

Intermediate flow operators

중간 연산자들은 upstream flow에 적용되고 downstream flow로 반환된다. 이 연산자들은 flow와 마찬가지로 cold하다. 각 연산자들은 suspend하지 않기 때문에 바로 실행되고, flow로 반환한다.
(이래서 중간연산자 붙으면.. flow로 반환되는거였구나 싶다..)

sequences들의 연산자와 중요한 차이점은 이 연산자들은 코드 내부적으로 suspending 함수들을 호출할 수 있다는 것이다.

suspend fun performRequest(request: Int): String {
    delay(1000) // imitate long-running asynchronous work
    return "response $request"
}

fun main() = runBlocking<Unit> {
    (1..3).asFlow() // a flow of requests
        .map { request -> performRequest(request) }
        .collect { response -> println(response) }
}

Transform operator

가장 일반적인 transform 연산자는 .transform

Size-limiting operators

.take와 같은 연산자는 지정한 한계에 도달하면 flow의 실행을 취소할 수 있다. 코루틴에서의 cancellation은 항상 exception을 던지기 때문에, 이와 같은 경우에는 항상 try { ... } finally { ... }와 같은 resource-management functions로 관리해야 한다.

fun numbers(): Flow<Int> = flow {
    try {                          
        emit(1)
        emit(2) 
        println("This line will not execute")
        emit(3)    
    } finally {
        println("Finally in numbers")
    }
}

fun main() = runBlocking<Unit> {
    numbers() 
        .take(2) // take only the first two
        .collect { value -> println(value) }
}

출력 : 
1
2
Finally in numbers

Terminal flow operators

이 연산자들은 flow의 collection을 시작시키는 suspending function이다.

.collect는 가장 기본적인 연산자이다.
그 외에도 toSet, toList, first, single, reduce, flod가 있다.
기능은 생각하는 그대로 이다. ㅇ_ㅇ

Flows are sequential

여러 flow에서 동작하는 특수한 연산자를 사용하지 않는한, flow는 순차적으로 실행된다.각 방출된 값은 중간 연산자에 의해 가공되어 upstream에서 downstream으로 전달되어 최종적으로 terminal 연산자에게 전달된다.

(1..5).asFlow()
    .filter {
        println("Filter $it")
        it % 2 == 0              
    }              
    .map { 
        println("Map $it")
        "string $it"
    }.collect { 
        println("Collect $it")
    }   

실행 결과 : 
Filter 1
Filter 2
Map 2
Collect string 2
Filter 3
Filter 4
Map 4
Collect string 4
Filter 5

---

나는 왜 공식문서 번역밖에 못하는가..
멧돼지처럼 deep dive 하는 사람이 되고 싶다..