Coroutines Flow 3편 - Flow 터미널 연산자, Flow는 순차적이다

Flow 터미널 연산자

Flow의 터미널 연산자는 flow를 수집을 시작하는 일시정지 함수*1이다. collect 연산자는 가장 기본 연산자이지만, 사용을 더 쉽게 만드는 다른 터미널 연산자들도 있다.

• 다양한 Collection으로의 변환을 수행하는 toList 와 toSet 같은 연산자.*2
• 첫 값만 가져오기 위한 first 연산자와 하나의 값만 방출되는 것을 확인하는 single 연산자.*3
• flow를 값으로 줄이는 reduce나 fold를 연산자.

예를 들어 :

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

📌 전체 코드는 이곳에서 확인할 수 있습니다.

이는 하나의 숫자를 출력한다 :

55

📖 아래 내용은 독자의 이해를 위해 번역자가 추가한 글입니다.

*1. Flow는 차가운 스트림이기 때문에 터미널 연산자가 실행되어야 방출되는 값들을 수집이 시작된다.

*2. toList()는 다음과 같이 사용할 수 있다. 

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

        println(flow.toList()) // [1, 2, 3] 출력
    }
}

*3. first() 는 여러 개가 방출되더라도 한 개만 출력된다. 여러개가 emit 되더라도 관계없다.

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

        println(flow.first()) // 1 출력
    }
}

하지만 한 개만 방출되어야 할 때 여러개가 방출되는 경우를 잡고 싶을 수 있다.

이 때 single() 을 사용한다. single은 아무 값도 방출되지 않는 flow에 대해서는 NoSuchElementException을 발생시키며, 2개 이상의 값이 방출되는 flow에 대해서는 Illegal Argument Exception을 발생시킨다.

/**
 * The terminal operator that awaits for one and only one value to be emitted.
 * Throws [NoSuchElementException] for empty flow and [IllegalStateException] for flow
 * that contains more than one element.
 */
public suspend fun <T> Flow<T>.single(): T {

flow에서 하나만 방출될 때 :

suspend fun main() {
    coroutineScope {
        val flow = flow {
            emit(1)
        }

        println(flow.single()) // 1 출력
    }
}

flow에서 아무 값도 방출되지 않을 때

suspend fun main() {
    coroutineScope {
        val flow = flow<Int> {
        }

        println(flow.single())
        // Exception in thread "main" java.util.NoSuchElementException: Flow is empty
    }
}

둘 이상의 값들이 방출될 때

suspend fun main() {
    coroutineScope {
        val flow = flow<Int> {
            emit(1)
            emit(2)
        }

        println(flow.single())
        // Exception in thread "main" java.lang.IllegalArgumentException: Flow has more than one element
    }
}

이 글은 Coroutines 공식 문서를 번역한 글입니다.

원문 : Asynchronous Flow - Terminal flow operators

원문 최종 수정 :  2022년 9월 28일

Flow는 순차적이다

여러 Flow들에서 작동하는 특수한 연산자를 사용하지 않는 한 각 개별 Flow의 컬렉션은 순차적으로 동작한다. 컬렉션은 터미널 연산자를 호출하는 Coroutine에서 직접 동작한다. 기본적으로 어떠한 새로운 Coroutines도 실행되지 않는다. 방출된 각 값들은 중간 연산자들에 의해 업스트림에서 다운스트림으로 처리된 후 터미널 연산자에게 전달된다.

정수 중 짝수를 필터링 한 후 문자열에 매핑하는 다음 예제를 참조하자 : 

(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

이 글은 Coroutines 공식 문서를 번역한 글입니다.

원문 : Asynchronous Flow - Flows are sequential

원문 최종 수정 :  2022년 9월 28일

목차로 돌아가기