Flow의 collect은 언제 suspend 될까?

사진: Unsplash의John Matychuk

 

 

이번 게시글에서는 Flow의 collect에 대해 알아보고, collect가 코루틴을 언제 중단시키는지 문제와 함께 알아보고자 합니다.

---

 

 

 

Flow의 collect 메서드

---

Flow는 비동기 데이터 스트림으로, 순차적으로 값을 방출하고 정삭적으로 완료하거나 예외와 함께 완료됩니다. 

 

Flow inferface에는 `collect` 메서드가 존재하는데, 이 메서드는 FlowCollector를 입력받아 데이터를 emit할 수 있도록 합니다.

public interface Flow<out T> {

    public suspend fun collect(collector: FlowCollector<T>)
}

 

 

collect는 suspend 함수이다.

여기서 알아두어야 할 것은 `collect` 메서드는 suspend 함수라는 것입니다.

 

Flow는 비동기 데이터 스트림으로 데이터를 방출(emit)하는 측과 수집(collect)하는 측은 서로 독립적으로 동작합니다. 이러한 비동기 작업을 처리하기 위해서는 suspend가 필요했습니다.

 

collect는 데이터를 처리하는 동안 Flow가 비동기적으로 동작하는 것을 방해하지 않음을 의미합니다.

Flow의 데이터 방출 과정(예: 네트워크 작업, 데이터베이스 읽기 등)은 별도로 진행되며, collect는 이를 기다렸다가 처리할 뿐입니다.

collect는 Flow가 데이터를 방출(emit)할 때까지 기다리며, 방출된 데이터를 수집하는 함수다.

 

 

 

Cold Flow의 collect

---

 예시에서는 Flow를 만들고, 이를 collect한 다음 "end"를 출력하게 만들었습니다.

fun main() {
    runBlocking {
        val flow = flow { 
            emit(1)
            emit(2)
            emit(3)
        }

        launch {
            flow.collect {
                println(it)
            }

            println("end")
        }
    }
}

// 결과
1
2
3
end

 

위의 예시를 알 수 있는 점은, collect는 Flow가 emit하는 데이터가 끝날 떄까지 기다리며, 그 동안 suspend된다는 점입니다. 이는 "end" 메시지가 마지막에 출력된다는 점으로 알 수 있습니다.

 

 

 

Hot Flow의 collect

---

 이번에는 cold flow를 `stateIn`을 사용해서 hot flow로 바꿔 보겠습니다.

fun main() {
    runBlocking {
        val flow = flow {
            emit(1)
            emit(2)
            emit(3)
        }.stateIn(
            scope = this,
            started = SharingStarted.Eagerly,
            initialValue = 0
        )


        launch {
            flow.collect {
                println(it)
            }

            println("end")
        }
    }
}

 

위의 결과가 어떻게 출력될 지 한번 예상해보시기 바랍니다. 

 

 

 

 

 

 

 

 

 

결과

3
// 그리고 영원히 지속됩니다. end는 출력되지 않습니다.

 

실행 결과로 알 수 있는 것은 2가지입니다.

1. emit한 결과중 `3`만 collect되었다.

2. "end"가 출력되지 않았다.

 

 

1. emit한 결과중 `3`만 collect되었다.

이 결과에 대한 이유는 많은 분들이 알고계실 것 같습니다. 이는 바로 hot flow의 특성 때문입니다. 

 

hot flow는 collector의 존재 여부와 관게없이 데이터를 방출합니다. `stateIn`의 started를 뭘로 설정하느냐에 따라서 결과가 달라질 수는 있겠지만, collector와 상관없이 emit이 이뤄진다고 생각하면 됩니다.

 

 

2. "end"가 출력되지 않았다.

사실 이 결과에 대한 이유는 위에서 이미 설명했었습니다. 다시 한번 더 설명하자면 Flow는 emit하는 데이터가 끝날 떄까지 기다리며, 그 동안 suspend 되는 특성을 가지고 있습니다.

 

그럼 hot flow의 경우에는 어떨까요?? 이러한 특성은 그대로 유지되면서, Flow는 언제든지 emit될 수 있습니다. 그렇기 때문에 hot flow를 collect 할 때는 모든 데이터가 emit되었다는 것을 보장할 수가 없게 됩니다.

 

다시 예제로 돌아오면, 예제의 collect가 진행되는 코루틴은 hot flow를 collect할 때의 특성 때문에 무한대로 중단되게 됩니다. 이 때문에 `println("end")`은 동일한 코루틴 내에 위치하기에 실행될 수가 없었습니다.

launch {
    flow.collect {
        println(it)
    }

    println("end")
}

 

 

IDE에서도 이를 인지하여 아래와 같이 경고 메시지를 띄워주기도 합니다.

 

 

 

마무리하며

---

이번 게시글은 제가 hot flow를 collect 할 때 발생했던 여러가지 상황을 이해하지 못했기에 작성하게 되었습니다. 이번 게시글을 통해 collect이 코루틴을 어떻게 중단하는지 이해하게 되었다고 생각합니다.