코틀린의 반응형 프로그래밍 #2 - Flow

 

이전 포스팅에서 반응형 프로그래밍이 무엇인지에 대하여 다루었습니다. 이번 포스팅에서는 코틀린의 Flow 타입을 활용하여 설명을 이어가보려고 합니다. 이 포스팅에서 독자를 Flow 전문가로 만들 의도는 없지만, 독자가 Flow를 사용하는 이유와 시기를 이해하고 Flow와 함께 작성된 코드를 읽을 수 있을 만큼의 수준의 내용은 제공하려고 합니다. 다음 샘플 코드는  글 전체에서 참조, 반복 및 리믹스하게 됩니다. 마지막에는 독자 모두가 완전히 이해할 수 있기를 바랍니다.

flow {
  emit(1)
  emit(2)
}.map { value ->
  "방출된 값 : $value"
}.collect { value -> println(value) }

 

정의

Flow는 비동기적으로 생성되는 순차적인 값들의 스트림을 나타냅니다. 이것은 단일 비동기적인 값을 반환하는 중단 함수와 순차적인 값 집합을 보유하지만 비동기적으로 생성되지 않는 컬렉션과 병치됩니다. Flow의 값들은 일반적으로 터미널 연산자(terminal operator, 아래쪽다시 다시 설명)를 통해 값을 "수집" 함으로써 액세스됩니다. Flow는 값들을 방출(emit)하는데, 이는 스트림에 추가함으로써 값들을 수집할 수 있게 해줍니다. Flow가 생성되고 종료되기 전에 "operators"의 "chain"을 Flow에 적용하여 Flow에 대한 방출에 반응하고 조작할 수 있습니다. Flow는 "cold" 상태인데, 이는 터미널 연산자가 적용될 때까지 값들이 방출되지 않음을 의미합니다. 즉, Flow 객체가 자원 낭비에 대한 걱정을 덜면서 전달될 수 있음을 의미합니다.

    @Test
    fun `flow_emit_test`() = runBlocking {
        flow_emit()
    }

    private suspend fun flow_emit() {
        flow {
            emit(1)
            emit(2)
        }.map { value ->
            "방출된 값 : $value"
        }.collect { value -> println(value) }

        /**
         * 결과
         *
         * 방출된 값 : 1
         * 방출된 값 : 2
         */
    }

 

Flow 빌더(Flow Builders)

Flow는 어떤 값이 어떻게 방출할지 정의하는 빌더 함수로 시작이 됩니다. 가장 간단한 두 가지 방법은 가변인자(varargs)를 flowOf(...) 를 통해서 내보내는것과 호출자로부터 모든값(e.g. 컬렉션 또는 범위)을 내보내는 .asFlow()를 사용하는 것입니다. 또 다른 일반적이고 더 강력한 방법은 flow {} 함수를 사용하는것입니다. 이 함수는 클로저(closure)를 매개 변수로 사용하여 emit() 함수를 호출하여 값을 방출할 수 있습니다. 값을 방출하는것뿐만 아니라 내부에 원하는 코드도 작성할 수 있습니다. 한가지 주목할 점은, Flow는 콜드 스트림(cold stream)이기 때문에 터미널 연산자가 적용될때까지 클로저가 실행되지 않습니다. 이러한 빌더는 매우 유용하지만 SQL 쿼리 결과를 Flow 타입으로 전달 할 수 있는 Room과 같은 라이브러리에서 생성된 Flow를 사용하는 것이 더 일반적입니다.

flow { // <== Flow 빌더
  emit(1)
  emit(2)
}.map { value -> 
  "방출된 값: $value"
}.collect { value -> println(value) } 

(0..10).asFlow()

flowOf("Hello", "Flow", "World", "!")

 

 

중간 연산자(Intermediate Operators)

이제 Flow가 생성되었으니, 그 Flow의 값을 조작하고 연산자를 사용하여 반응할 수 있습니다. 터미널 연산자로 알려진 특별한 종류의 연산자는 다음에 다룰 것입니다. 이 연산자들은 값을 제한하고, 값을 변환하고, 추가적인 값을 방출하며, Flow에 대한 다른 중요한 연산을 수행할 수 있습니다. 이들은 연산자의 다른 체인을 통해 다른 결과값을 생성함으로써 기본적인 Flow를 재사용할 수 있습니다.

예

• Limit - filter, 조건을 통과하지 못하는 Flow에서 값을 제거하는 필터
• Transform - map, T형에서 S형으로 값을 변환하는 맵, Flow를 Flow<S>로 바꿀수 있음
• Emit - onStart, Flow가 수집되기 전에 지정된 action을 호출하는 flow를 반환
• React - onEach, 값을 받을 때마다 실행되는 각각의 반응

flow {
  emit(1)
  emit(2)
}.map { value -> // <== 중간 연산자
  "방출된 값 : $value"
}.collect { value -> println(value) } 

(0..10).asFlow()
  .onStart { emit(11) }
  .filter { it % 2 != 0 }
  .onEach { println("홀수 $it") }
  .collect { ... }

 

 

터미널 연산자(Termianl Operators)

최종 연산자는 터미널 연산자로, 터미널 연산자는 Flow를 가져오고 Cold Flow를 활성화하여 값을 사용할 수 있도록 합니다. 터미널 연산자의 첫 번째 타입은 값 또는 값 집합을 가져온 다음 진행중인 Flow를 취소(방출 종료)하는 연산자입니다. 이 타입의 가장 일반적인 연산자는 첫 번째 요소가 방출된 후 Flow를 취소하고 더 이상의 값을 받지 않는 연산자입니다. 다른 타입은 취소가 완료되거나 다른 메커니즘에 의해 취소될 때까지 Flow에서 값을 계속 검색하는 비취소 연산자입니다. 이 타입의 가장 일반적인 연산자는 수신된 각 값에 대해 클로저를 실행하는 collect입니다.

flow {
  emit(1)
  emit(2)
}.map { value -> 
  "방출된 값 : $value"
}.collect { value -> println(value) } // 터미널 연산자

// x는 0
val x = (0..10).asFlow()
  .first()

 

 

디스패처(Dispatchers)

Flow 연산자 클로저와 터미널 연산자는 Flow의 비동기 특성을 처리하기 위하여 중단 함수(suspending function)을 사용합니다. Flow 체인에서 Dispatcher를 명시하지 않으면 터미널 연산자를 호출할때 사용되는 Dispatcher를 Flow 체인 전체에서 사용됩니다. 현재의 Dispatcher를 변경하기 위해서는 flowOn 연산자가 사용됩니다. flowOn 연산자에 대한 중요한 사항은 체인 아래의 모든 호출에 적용하는 대신 다음 상위 flowOn 호출까지 체인의 모든 이전 연산자에 적용된다는 것입니다.

flow { ... }
  .map { ... // IO Dispatcher에서 실행됨 }
  .flowOn(Dispatchers.IO)
  .filter { ... // Main Dispatcher에서 실행됨 }
  .flowOn(Dipatchers.Main)

 

 

예외 처리(Exception Handling)

Flow 체인에서 예외가 발생하면 Flow가 취소되고, 예외가 터미널 연산자가 위치한 코드 블럭에 노출이 됩니다. 이를 통해 종료를 처리하는 코드 블럭은 Flow를 try/catch하고 원하는 대로 이어서 진행할 수 있습니다. Flow가 예외를 처리해야 하는 경우 catch 연산자를 사용하거나 다른 연산자 내부에서 try/catch를 사용하여 처리 할 수 있습니다. catch 연산자를 사용하는 경우에는 Flow는 종료되지만 클로저에서 다른 값을 방출할 수도 있습니다. 만약 다른 연산자에서 try/catch가 호출되어 예외가 처리되면 Flow는 취소 대신 정상적으로 계속해서 사용할 수 있습니다.

    @Test
    fun `flow_exception_do_not_use_try-catch`() = runBlocking {
        `flow_exception_handling-use_only_catch`()
    }

    private suspend fun `flow_exception_handling-use_only_catch`() {
        flow {
            emit(1)
            throw Exception("Something Went Wrong")
            emit(2)
        }.map { value ->
            "Emitted Value $value"
        }.catch { emit("An Exception Occurred") }
            .collect { value -> println(value) }
    }

    @Test
    fun `flow_exception_use_try-catch`() = runBlocking {
        flow_exception_handling_try_catch()
    }

    private suspend fun `flow_exception_handling_try_catch`() {
        return (0..10).asFlow()
            .map {
                try {
                    10 / it // 10을 Flow에서 방출된 해당 값으로 나누기
                } catch (e: ArithmeticException) {
                    0 // 0으로 나누려고 하면 예외 발생. 이 때 0을 넘겨서 예외 복구.
                }
            }.collect { print("$it ") } // 출력 : 0 10 5 3 2 2 1 1 1 1 1
    }

 

 

Flow의 Combining 및 Flattening

지금까지 다룬 것들은 모두 Flow를 생성하고, 변환하고, 이에 반응하는 것들을 보여주었지만, Flow는 어떻게 하나로 모을 수 있을까요? 첫번째 방법은 combine과 같은 결합 연산자를 사용하여 Flow를 결합하는 것입니다. 이를 통해 서로 다른 두 Flow의 방출을 하나의 함수를 사용하여 결합할 수 있습니다. 다른 방법은 flatMapLatest와 같은 연산자를 사용하여 하나의 Flow를 다른 하나의 방출로부터 생성하는 것입니다. 결과적으로 단일 Flow가 되지만 다른 Flow의 결과로 생성된 값으로 구성됩니다.

    @Test
    fun `combine_flatmaplatest_test`() = runBlocking {
        combine_flatmaplatest()
    }

    suspend fun combine_flatmaplatest() {
        val flow1 = flow {
            emit(1)
            emit(2)
        }

        val flow2 = (0..10).asFlow()

        // Combine
        println("Combine 결과")
        combine(flow1, flow2) { val1, val2 -> val1 + val2 }
            .collect { print("$it ") } // 1 3 5 6 7 8 9 10 11 12 13

        // Flat Map
        println("\nFlatMapLatest 결과")
        flow1.flatMapLatest { value -> flow2.onStart { emit(value) } }
            .collect { print("$it ") }
        // 1 0 1 2 3 4 5 6 7 8 9 10 2 0 1 2 3 4 5 6 7 8 9 10
    }

 

flow combine 연산자에 대한 마블

 

flatMapLatest 연산자에 대한 마블

 

결론

본 포스팅에서 설명한 것들은 Flow를 조금 맛본 것에 불과합니다. 다양한 Flow 작업을 볼 수 있는 훌륭한 리소스는 Flow Marbles입니다. 이는 옵션을 보고 Flow가 적용될때 Flow에 어떤 영향을 미치는지 생각해보고 탐구하는데 좋습니다. 이 다음 포스팅에서는 Flow의 하위 항목인 SharedFlow에 대하여 다루어 보려고 합니다. 감사합니다.

 

 

---