드디어 flow 공식문서 스터디의 마지막이다..
기초중의 기초이지만.. 아직도 이해하기 힘든 부분들이 있는 건 맞는 듯 ㅜㅜ 그래도 같이 공부하니까 훨씬 낫다..
사설은 그만하고 flow 공문 마지막 읽기 ㄱㄱㄱㄱ
Flow Completion
flow 수집이 정상적이던, 예외를 터뜨리던 다음 action을 취하고 싶을 것이다.
액션을 취하기 위해서는 imperative(명령형) 혹은 declarative(선언형)으로 코드를 작성할 수 있다.
Imperative finally block
수집이 완료된 후에 수행을 하고 싶다면, try-catch 이외에도 finally를 사용할 수 있다.
Declarative handling
선언형으로 코드를 작성하고 싶다면, onCompletion 연산자를 이용하면 된다.이 연ㅅ나자는 flow가 완전히 수집이 되고나면 실행된다.
fun simple(): Flow<Int> = (1..3).asFlow()
fun main() = runBlocking<Unit> {
simple()
.onCompletion { println("Done") }
.collect { value -> println(value) }
}onCompletion 의 중요 이점은 람다의 nullable Throwable parameter이다.
이 parameter로 flow 수집이 정상적으로 혹은 예외를 발생했는지 확인가능하다.
fun simple(): Flow<Int> = flow {
emit(1)
throw RuntimeException()
}
fun main() = runBlocking<Unit> {
simple()
.onCompletion { cause -> if (cause != null) println("Flow completed exceptionally") }
.catch { cause -> println("Caught exception") }
.collect { value -> println(value) }
}
// ---Output```
// 1
// Flow completed exceptionally
// Caught exceptiononCompletion은 catch 와는 다르게, exception을 처리하지는 않는다. 위의 코드를 보면, exception이 downstream으로 진행되고, catch에서 처리가 된다.
Successful completion
onCompletion 과 catch의 또다른 차이점은, onCompletion은 모든 예외를 확인하고 취소 또는 실패 없이 upstream이 성공적으로 완료된 경우에만 null 예외를 수신가능하다는 점이다.
Imperative versus declarative
선언형을 사용할 것이냐 명령형을 사용할 것 이냐는 본인의 선호도와 코드 스타일에 따라 선택하면 된다고 한다
Launching Flow
비동기적으로 들어오는 값을 처리하기 위해 onEach 연산자를 사용할 수 있다.
onEach 연산자는 중간 연산자여서 이 자체만으로 기능을 할 수 없기 때문에 collect와 같은 terminal 연산자가 필요하다.
// Imitate a flow of events
fun events(): Flow<Int> = (1..3).asFlow().onEach { delay(100) }
fun main() = runBlocking<Unit> {
events()
.onEach { event -> println("Event: $event") }
println("Done")
}
// ---Output---
// Done
fun main() = runBlocking<Unit> {
events()
.onEach { event -> println("Event: $event") }
.collect() // <--- Collecting the flow waits
println("Done")
}
// ---Output---
// Event: 1
// Event: 2
// Event: 3
// Done위의 코드를 확인해보면 알겠지만, collect 터미널 연산자를 사용하게 되면 collect 구문 이후의 실행문들은 flow가 종료된 후에야 실행이 된다. (Done을 print 하는 코드가 대기) 비동기 이벤트를 계속해서 기달고 싶지 않다면 launchIn 터미널 연산자를 사용할 수 있다.
fun main() = runBlocking<Unit> {
events()
.onEach { event -> println("Event: $event") }
.launchIn(this) // <--- Launching the flow in a separate coroutine
println("Done")
}
// ---Output---
// Done
// Event: 1
// Event: 2
// Event: 3위와 같이 launchIn 터미널 연산자를 사용하게 되면 flow collection은 다른 코루틴에서 동작하여 다른 코드블럭을 동시에 실행시킬 수 있다.
launtchIn 을 사용할 때는 파라미터로 collect 작업을 수행할 코루틴의 scope를 지정해줘야한다.
launchIn은 job을 반환하기 때문에 이를 이용하여 flow를 조작할 수 있다.
Flow cancellation checks
편의성을 위해서 flow builder는 각 방출된 값에 취소를 위한 추가적인 ensureActive 체크를 한다.
이 말은 즉슨, 값을 계속해서 방출하는 flow를 cancel 가능하다는 것인데, 다음 예제를 확인해보자
fun foo(): Flow<Int> = flow {
for (i in 1..5) {
println("Emitting $i")
emit(i)
}
}
fun main() = runBlocking<Unit> {
foo().collect { value ->
if (value == 3) cancel()
println(value)
}
}위 코드는 4를 방출할 때 터진다.
fun main() = runBlocking<Unit> {
(1..5).asFlow().collect { value ->
if (value == 3) cancel()
println(value)
}
}이 코드는 5까지 방출 후 runBlocking으로 부터 취소를 확인한다.
이렇게 다른 이유는, 아래의 코드는 suspend할 시점이 없어서 취소 체크를 할 시간이 없다..
→ 이거 flow 스터디 첫시간에 나눴던 이야기.. cancel을 하기 위해서는 suspend 부분이 필요
busyflow를 cancellable하게 바꾸는 방법이 두가지가 있다.
.onEach{ } 내에서 ensureActive를 붙여주거나 flow 자체를 cancellable 하게 바꿔주는 것이다.
// 각각 마다 ensureAcitive 보장
fun main() = runBlocking<Unit> {
(1..5).asFlow()
.onEach { currentCoroutineContext().ensureActive() }
.collect { value ->
if (value == 3) cancel()
println(value)
}
}
// cancellable() 사용
fun main() = runBlocking<Unit> {
(1..5).asFlow().cancellable().collect { value ->
if (value == 3) cancel()
println(value)
}
}Flow and Reactive Streams
공식문서에 달려있던 링크를 첨부해본다..
Reactive Streams and Kotlin Flows
