Asynchronous Flow - Intermediate flow operators~

https://kotlinlang.org/docs/flow.html#intermediate-flow-operators

Asynchronous Flow | Kotlin

위 링크를 공부하며 남기는 기록이다. Flow 중간연산자부터 하겠다. 이전의 공부기록은 작성을 깜박했는데 추후 기록해둘 예정이다.

Intermediate flow operators

플로우 중간 연산자다. 플로우 자체를 중간 연산자로 사용할 수 있으며 업스트림 플로우에 적용시킨다음 다운스트림 플로우를 반환 한다. 이때 동작은 콜드 플로우다. 

기본연산자로는 .map, .filter등이 있는데 시퀀스와 차이점은 suspend 함수를 호출할 수 있다는 것이다.

inline fun <T, R> Flow<T>.map(crossinline transform: suspend (T) -> R): Flow<R>

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) }
}

map에서 1..3인 콜드 플로우를 하나씩 받아서  performRequest에 넣는다. 그러면 collect는 return된 response $request를 구독해서 가져간다.

Transform operator

변환 연산자는 map과 filter와 같이 생겼다. transform 연산자를 사용하며 임의의 값을 임의의 시간에 방출할 수있다.

(1..3).asFlow() // a flow of requests
        .transform { request ->
            emit("Making request $request") 
            emit(performRequest(request)) 
        }
        .collect { response -> println(response) }

이 경우 한 플로우 당 Making~ 을 먼저 출력하고 performRequest로 받은 값을 뒤이어 출력한다.

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

map과 비슷하다고 하니 이렇게 바꿔봤다. 출력값을 예상해봤을때 map에서 반환되는 값이 없음을 알 수 있는데, 따라서 아래와 같은 출력문이 이어진다. 이에 반에 emit으로 넘겨주는 transform은 emit된 값을 차례대로 response에 넣어주니까 값이 반환됨을 알수있다.

Making request 1
response 1
kotlin.Unit
Making request 2
response 2
kotlin.Unit
Making request 3
response 3
kotlin.Unit

collect의 respose에 들어갈 println은 값을 반환하지 않기 때문에 kotlin.Unit이 출력된다.

Size-limiting operators

cancel을 일으키는 take함수를 이용해 받아오는 값 개수를 조절하는 방식이다.

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) }
}

코루틴 안에서 cancel을 일으키기 때문에 꼭 try-finally로 예외처리를 해줘야한다. take(2)는 emit(2)까지만 받고 끊어버린다. 받은 값을 collect해서 가져올수있다는 점이 그냥 cancel과 다르다고 할 수 있겠다.

Terminal flow operators(종단 연산자)

플로우를 수집하는 suspend 함수들이다. collect가 대표적이고 toList, toSet과 같이 바로 컬렌션으로 바꿀 수도 있다.

val sum = (1..3).asFlow()
        .map { it * it } // squares of numbers from 1 to 5                           
        .reduce { a, b -> a + b } // sum them (terminal operator)
    println(sum)

reduce, fold도 해당한다. 코드를 보면 a에 1이 들어가고 b에 2*2가 들어간다. 합친게 a로 다시 들어가고 b에 3*3이 들어오는 누적합 컬렉션 내장함수이다.

Flows are sequential

플로우의 수집은 기본적으로 순차적으로 수행된다. 각 연산자마다 emit된 값들은 업스트림의 모든 중간연산자에 의해 처리되고 최종적으로 다운스트림에 전달되어 종단연산자에서 수집하는게 과정이다.

Flow Context

플로우 수집은 항상 수집을 호출한 코루틴의 context에서 이루어진다. -> context 보존이라고 부른다.

fun simple(): Flow<Int> = flow {
    log("Started simple flow")
    for (i in 1..3) {
        emit(i)
    }
}  

fun main() = runBlocking<Unit> {
    simple().collect { value -> log("Collected $value") } 
    val job = launch{
        simple().collect { value -> log("Collected $value") } 
    }
}

job으로 묶은 코루틴과 runBlocking 바로 밑에있는 코루틴에서 각각 실행한 플로우는 각각의 코루틴 컨텍스트에서 돌아간다. simple()로 만든 플로우 함수가 수집을 호출한 코루틴 컨텍스트를 따른다는 걸 기억하면 된다.

[main @coroutine#1] Started simple flow
[main @coroutine#1] Collected 1
[main @coroutine#1] Collected 2
[main @coroutine#1] Collected 3
[main @coroutine#2] Started simple flow
[main @coroutine#2] Collected 1
[main @coroutine#2] Collected 2
[main @coroutine#2] Collected 3

문제는 장기적인 사용이 필요한 작업은 따로 별도의 스레드에서 돌려야되는데 이때 코루틴 컨텍스트를 전환해서 사용하는 withContext를 flow빌더 내에서 하면 컨텍스트 보존을 지키지 못하기 때문에 오류가 발생한다. 이럴 때 필요한 연산자가 flowOn연산자이다.

fun foo(): Flow<Int> = flow {
    for (i in 1..3) {
        Thread.sleep(100) // pretend we are computing it in CPU-consuming way
        log("Emitting $i")
        emit(i) // emit next value
    }
}.flowOn(Dispatchers.Default) // RIGHT way to change context for CPU-consuming code in flow builder

fun main() = runBlocking<Unit> {
    foo().collect { value ->
        log("Collected $value") 
    } 
}

이렇게 flowOn을 사용해주면 flow빌더부분은 DefaultDispatcher에서 돌고 collect는 메인 스레드 코루틴에서 돌게된다. 컨텍스트 전환이 안전하게 이루어졌다. 즉 flowOn의 목적은 방출-수집을 순서대로 처리하는 특성인 순차성을 못지키는 대신 컨텍스트 보존을 지키면서 코루틴 디스패처를 바꾸는 데 있다.

Buffering

delay대신에 buffer연산자를 사용하는 방법이다. 

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
    val time = measureTimeMillis {
        simple()
            .buffer() // buffer emissions, don't wait
            .collect { value -> 
                delay(300) // pretend we are processing it for 300 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")
}

여기서 버퍼를 빼버리면 시간이 각 수집당 400씩 총 1200ms이 걸리는데 버퍼를 사용하면 처음 100만 걸려서 100+300+300+300인 1000ms이 대충 걸린다. buffer연산자를 사용하면 수집때 발생하는 delay(300) 때 미리 방출시 delay를 처리해서 더 효율적인 실행이 일어난다. 그렇다면 수집이 더 빠르면 어떻게 될까?

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
    val time = measureTimeMillis {
        simple()
            .buffer() // buffer emissions, don't wait
            .collect { value -> 
                delay(50) // pretend we are processing it for 300 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")
}

처음 100+50+100+100 에 오차범위 약 50ms이므로 350~400ms이 나온다. 수집때가 더 빠르다면 버퍼도 크게 효과가 없다는 것을 알 수 있다.

Conflation & Processing the latest value

conflate연산자를 사용해서 수집이 너무 느릴경우 값을 하나하나 다 방출하지않고 중간값을 건너뛰고 제일 최신값을 가져온다. 

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
    val time = measureTimeMillis {
        simple()
            .conflate() // conflate emissions, don't process each one
            .collect { value -> 
                delay(190) // pretend we are processing it for 300 ms
                println(value) 
            } 
    }   
    println("Collected in $time ms")
}

2까지 수집되는데 200이 걸리는데, 1을 수집하면서 200을 넘어버리면 conflate가 넘겨버린다. 하지만 190까지 수집한다면 2도 같이 수집하는 것을 볼 수 있다.

 

최신값 처리하는 방법 중에 conflation(병합)도 있고, 연산자뒤에 Latest가 붙은 연산자가 존재한다. 이 연산자들은 새값이 방출되면 수집하던 코드블록을 취소하고 새로 방출된 값을 가져온다.

fun simple(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100) // pretend we are asynchronously waiting 100 ms
        emit(i) // emit next value
    }
}

fun main() = runBlocking<Unit> { 
    val time = measureTimeMillis {
        simple()
            .collectLatest { value -> // cancel & restart on the latest value
                println("Collecting $value") 
                delay(300) // pretend we are processing it for 300 ms
                println("Done $value") 
            } 
    }   
    println("Collected in $time ms")
}

설명이 겹치니 그냥 넘어가도록 하겠다.

Composing multiple flows

다중 플로우를 활용하는 방법들이다.

Zip

val nums = (1..3).asFlow() // numbers 1..3
    val strs = flowOf("one", "two", "three") // strings 
    nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string
        .collect { println(it) } // collect and print

nums 플로우 방출값과 strs플로우 방출값을 zip으로 묶어서 return으로 String을 반환하는 코드다.  시퀀스에서 쓰는 zip과 거의 동일하다.

Combine

이미 방출된 값으로 또 추가연산을 수행해야하는 경우 combine연산자를 이용한다.

fun main() = runBlocking<Unit> { 

    val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
    val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms          
    val startTime = System.currentTimeMillis() // remember the start time 
    nums.combine(strs) { a, b -> "$a -> $b" } // compose a single string with "combine"
        .collect { value -> // collect and print 
            println("$value at ${System.currentTimeMillis() - startTime} ms from start") 
        } 
}

이러면 순서성을 유지한채로 가는데 strs가 각 방출이 끝나는 시점, nums 각 방출이 끝나는 시점이 계속 맞물려서 아래 출력과 같이 나온다. 처음에는 strs까지 기다려야하니 400ms, nums는 다음 방출이 시작됐으니 nums는 갱신되고 str는 좀 더 걸리고 하는 과정이 지속된다.

1 -> one at 444 ms from start
2 -> one at 648 ms from start
2 -> two at 846 ms from start
3 -> two at 949 ms from start
3 -> three at 1246 ms from start

즉 방출이 일어날 때마다 다른 플로우의 최신값을 가지고 병합해서 출력하는게 combine 연산자다.

Flattening Flows

플로우의 방출값을 또 플로우에 넣을 때 나중의 처리를 위해서 평탄화가 필요하다. flatten, flatMap 연산자가 이때 사용된다. 

flatMapConcat는 그냥 이중 플로우가 순차적으로 끝나는걸 기다려준다.

flatMapMerge는 각 플로우를 동시에 돌려버린다(동시 수집해서 단일 플로우로 가능한 빨리 방출시킨다). 

flatMapLatest는 collectLatest와 비슷한데, 수집이 느리면 건너 뛰어버린다.

 

Flow exceptions

플로우를 take와 같은 걸로 cancel이 나게하면 emit에서 예외를 던지는데 이 예외는 수집쪽에서 받는다. 이 말은 예외처리를 수집쪽에서 해줘야한다는 뜻이다.

    try {
        simple().collect { value ->         
            println(value)
            check(value <= 1) { "Collected $value" }
        }
    } catch (e: Throwable) {
        println("Caught $e")
    }

value가 2가되면 catch로 넘어가는 코드다. 이 경우 중간/종단/방출연산자 중 어느부분에서 예외가 발생했는지 알 수 없는데, flow에 map으로 예외를 넘겨주면 종단연산자에서 받는걸 확실하게 인지할 수 있다. 

fun simple(): Flow<String> = 
    flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i) // emit next value
        }
    }
    .map { value ->
        check(value <= 1) { "Crashed on $value" }                 
        "string $value"
    }

Exception transparency 예외 투명성

플로우를 사용할 때는 예외 처리를 잘해야한다. flow{}빌더 안에서 try~catch로 예외처리해서 예외 없는 값만 방출하는 것은 예외 투명성을 위반하는 것이라고 할 수 있다. 

 

emit이 catch연산으로 이런 예외 투명성을 지킬 수 있다. catch에서 emit을 사용해 방출하거나, throw를 또 사용해서 다시 예외를 던질 수 있다.

fun simple(): Flow<String> = 
    flow {
        for (i in 1..3) {
            println("Emitting $i")
            emit(i) // emit next value
        }
    }
    .map { value ->
        check(value <= 1) { "Crashed on $value" }                 
        "string $value"
    }

fun main() = runBlocking<Unit> {
    simple()
        .catch { e -> emit("Caught $e") } // emit on exception
        .collect { value -> println(value) }
}

예외가 발생하는 9번 라인에서 던져진 e를 catch가 받아서 emit하고 이걸 collect가 받는다. try~catch를 쓰지않아도 동일한 결과를 보여줌을 알 수 있다.

 

catch 연산자는 업스트림에서 발생하는 예외만 잡아준다. 위 코드 블럭에서 check가 main함수 안 collect에 위치했다면 예외처리를 하지못하고 오류를 내뿜었을 것이다.

 

Flow completion

정상이던 예외던 플로우가 종료되면 처리하는 방식이 2가지가 있다. imperative와 declarative방식이다.

imperative한 방식은 무조건 try~finally를 거치게 만들어서 수집이 종료되고 실행할 코드를 정의해두는 것이다.

try {
        simple().collect { value -> println(value) }
    } finally {
        println("Done")
    }

declarative한 방식으로는 플로우빌더에 onCompletion을 붙여서 종료시 수행할 행동을 선언해두는 것이다.

simple()
        .onCompletion { println("Done") }
        .collect { value -> println(value) }

그러나 onCompletion은 예외를 잡지 못하기때문에 catch와 같이 사용해줘야한다. 또 한가지 catch와 onCompletion의 차이는 onCompletion이 예외를 종료시점으로 해서 발견할 수는 있어도 플로우 취소나 실패를 같이 호출하는 게 아니기 때문에 실행 중 오류가 나버린다.

 

다음 코드가 그 예시인데, 예외결과를 받아서 출력은 하지만 다운스트림에서 발생하는 예외 탓에 실행오류가 난다.

fun simple(): Flow<Int> = (1..3).asFlow()

fun main() = runBlocking<Unit> {
    simple()
        .onCompletion { cause -> println("Flow completed with $cause") }
        .collect { value ->
            check(value <= 1) { "Collected $value" }                 
            println(value) 
        }
}

Imperative(명령형)와 declarative(선언형)가 각각 특징이 있는데 필요에 따라 맞춰사용하는게 좋다.

Launching flow

onEach는 addEventListener와 같은 행동을 보여준다. onEach는 중간연산자이기 때문에 값을 받아 무조건 수집해줘야한다.

fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

fun main() = runBlocking<Unit> {
    events()
        .onEach { event -> println("Event: $event") }
        .collect() // <--- Collecting the flow waits
    println("Done")
}

onEach를 사용할 때 해당 플로우를 분리된 코루틴으로 보내버릴 수 있는 종단연산자가 있는데 launchin이 그것이다.

fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }

fun main() = runBlocking<Unit> {
    events()
        .onEach { event -> println("Event: $event") }
        .launchIn(this) // <--- Launching the flow in a separate coroutine
    println("Done")
}

launchIn을 붙여 버려서  events 플로우는 분리된 코루틴이 되어, 아래 라인의 println이 먼저 실행된다. addEventListener와 더 유사한건 launchIn이다. 수집해버리기 때문에 event회수도 된다고 볼 수 있어 removeEventListener 역할도 겸한다.

 

launchIn은 코루틴을 생성하기 때문에 Job객체를 반환한다. 따라서 job에 걸수있는 연산도 다 가능하다.

 

플로우를 중간에 cancel하려면 cancellable()연산자를 붙여주면 된다. 이러면 cancel이 일어날때 예외를 반환한다.

 

 

 

"댓글, 공감 버튼 한 번씩 누르고 가주시면 큰 힘이 됩니다"