Flow를 공부해보자 4편

Yerin·2023년 11월 22일

코틀린

목록 보기

10/11

flow를 공부하고 있지만.. 프로젝트에서는 stateFlow를 적용했다...
심지어 굳이 flow를 쓰는데의 이점이 없는 부분이라 아직 이점이 뭔지는 모르겠지만.. 그래도 cold flow 정복 한번 가보자고 ~

저번시간에 이어서 공식문서를 읽어보자 .........

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

// ---Output---
// 1 -> one
// 2 -> two
// 3 -> three

같이 처리할 값이 있어야만 처리가 된다.
-> 동시 처리 되는 값이 없으면 한쪽이 씹히게 된다.

Combine

각 flow에서 가장 최근에 방출된 값을 결합하여 생성된 값을 가진 flow을 반환한다.

val flow = flowOf(1, 2).onEach { delay(10) }
val flow2 = flowOf("a", "b", "c").onEach { delay(15) }
flow.combine(flow2) { i, s -> i.toString() + s }.collect {
    println(it) // Will print "1a 2a 2b 2c"
}

둘 중 하나만 값이 collect 되어도 처리된다.
하지만, 둘 중 하나가 트리거 되기 전이라면, 그 전의 값들은 모두 씹힌다.

Flattening flows

<Flow<Flow<String>>> 과 같은 flow를 가공하기 위해서는 single flow로 flatten 해야 한다.

flatMapConCat

이 함수는 inner flow가 완료될 때까지 기다린 후 다음 collect를 시작한다.

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First") 
    delay(500) // wait 500 ms
    emit("$i: Second")    
}

fun main() = runBlocking<Unit> { 
    val startTime = currentTimeMillis() // remember the start time 
    (1..3).asFlow().onEach { delay(100) } // emit a number every 100 ms 
        .flatMapConcat { requestFlow(it) }                                                                           
        .collect { value -> // collect and print 
            println("$value at ${currentTimeMillis() - startTime} ms from start") 
        } 
}

// ---Output---
// 1: First at 153 ms from start
// 1: Second at 654 ms from start
// 2: First at 755 ms from start
// 2: Second at 1255 ms from start
// 3: First at 1355 ms from start
// 3: Second at 1856 ms from start

flatMapMerge

flow를 동시에 수집하고 그 값을 single flow로 합쳐서 가능한 빨리 값을 방출하는 방법.

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First") 
    delay(500) // wait 500 ms
    emit("$i: Second")    
}

fun main() = runBlocking<Unit> { 
    val startTime = currentTimeMillis() // remember the start time 
    (1..3).asFlow().onEach { delay(100) } // a number every 100 ms 
        .flatMapMerge { requestFlow(it) }                                                                           
        .collect { value -> // collect and print 
            println("$value at ${currentTimeMillis() - startTime} ms from start") 
        } 
}

// ---Outout---
// 1: First at 136 ms from start
// 2: First at 231 ms from start
// 3: First at 333 ms from start
// 1: Second at 639 ms from start
// 2: Second at 732 ms from start
// 3: Second at 833 ms from start

기본적으로 실행 흐름이 16개...
멧돼지가 flow를 엄청 많이 만드는 실행 코드를 보여줬는데.. 16개까지만 계속 만들고 그 이후부터는 대기를 타더라는 소문..

flatMapLatest

새로운 flow가 방출되는 즉시 이전 flow의 수집이 취소된다.

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First") 
    delay(500) // wait 500 ms
    emit("$i: Second")    
}

fun main() = runBlocking<Unit> { 
    val startTime = currentTimeMillis() // remember the start time 
    (1..3).asFlow().onEach { delay(100) } // a number every 100 ms 
        .flatMapLatest { requestFlow(it) }                                                                           
        .collect { value -> // collect and print 
            println("$value at ${currentTimeMillis() - startTime} ms from start") 
        } 
}

// ---Output---
// 1: First at 142 ms from start
// 2: First at 322 ms from start
// 3: First at 425 ms from start
// 3: Second at 931 ms from start

Flow Exceptions

Collector try and catch

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

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

// ---Output---
// Emitting 1
// 1
// Emitting 2
// 2
// Caught java.lang.IllegalStateException: Collected 2

위 코드를 통해 try-catch를 이용하면 오류를 잡고나서 방출이 멈추는 것을 알 수 있다.

Everything is caught

emitter나 중간 또는 터미널 오퍼레이터에서 발생하는 모든 예외를 포착한다.

Exception transparency : 예외 투명성?

방출하는 쪽에서 예외 처리를 캡슐화나는 방법은 무엇이 있을까?

모든 Flows 구현체는 exception에 투명해야한다.

flow 빌더를 try/catch 블럭 안에서 사용하는 것은 exception에 투명하지 못한 행위이다.

exception에 투명하다
downstream에서 발생한 에러를 미리 처리하여 collector가 알 수 없으면 안된다는 의미이다.
에러가 났더라도 어떤 형태로든 collector가 알아차릴 수 있어야 한다.

emitter는 catch 연산자를 통하여 exception transparency를 유지할 수 있고 exception 처리를 캡슐화 할 수 있다. catch 연산자 안에서 예외를 분석하여 어떤 예외가 포착되었는지에 따라 다른 방식으로 대응할 수 있다.

simple()
    .catch { e -> emit("Caught $e") } // emit on exception
    .collect { value -> println(value) }

위 코드와 같이 try-catch 구문이 없어도 예외 처리가 가능하다.

Transparent catch

catch 중간 연산자를 사용하면 upstream에서 생긴 exception만 처리 가능하다.

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

fun main() = runBlocking<Unit> {
    simple()
        .catch { e -> println("Caught $e") } // does not catch downstream exceptions
        .collect { value ->
            check(value <= 1) { "Collected $value" }                 
            println(value) 
        }
}