[Flow Basics 1] Flow 기초(Flow Builder, Cold/Hot Stream)

1. 기본 개념

Flow는 코틀린에서 쓸 수 있는 비동기 '스트림'이다. stream으로 지속적으로 값을 받아들이는 flow는 suspend functions에서 단일 값을 리턴 받는 형식과 반대되는 개념이라고 할 수 있다.

Steam에는 3가지 객체가 존재한다.

1. Producer : stream에 data를 흘려보내는 객체이다. 코루틴을 활용해 비동기적으로 데이터를 방출 가능하다.
2. Intermediaries : 있을수도 없을 수도 있다. 필요한 경우에 방출되는 데이터를 가공하는 역할이다
3. Consumer : Steam으로 부터 data를 소비(consume)하는 객체이다.

Android에서 일반적으로 repository가 producer의 역할이고, 데이터를 활용해 화면을 구성하는 UI가 Consumer의 역할이다. 때때로 User의 input을 받아들이는 UI가 producer가 되기도 한다. 그리고 Producer와 Consumer 사이에 있는 domain layer가 intermediary의 역할을 수행하기에 적합하다.

• Entities involved in streams of data: consumer, optional intermediaries, and producer (from android developer)

2. Cold Stream vs. Hot Stream

flow에는 cold와 hot 두가지 개념이 있다.

• Cold Stream
  • collected나 subscribed되었을때 1:1로 값을 방출하기 시작
  • 예를 들어 HTTP 요청을 만들어 서버에서 데이터를 가져올 때 활용할 수 있다.
• Hot Stream
  • 0개 이상의 상대를 향해 지속적으로 값을 전달
  • collected나 subscribed되지 않았어도 값을 방출한다.
  • 1:1로 데이터를 전달하는 것이 아니라 여러 개의 collector 혹은 subscriber가 방출된 값을 같이 공유할 수 있다.
  • 예를 들어, 센서 데이터를 다룰 때 여러 구독자가 실시간 데이터 스트림을 공유할 수 있다
  • 실시간으로 변경되는 주식 호가창 데이터를 구성할 때 활용가능하다

3. Flow 만들기

flow : 플로우 빌더 함수. Cold Stream 이다.

emit : 스트림에 데이터 방출

collect : 요청 측에서 collect 호출하면 값 발생 시작

fun flowSomething(): Flow<Int> = flow {
    repeat(10) {
        emit(Random.nextInt(0, 500))
        delay(10L)
    }
}
​
fun main() = runBlocking {
    flowSomething().collect { value ->
        println(value)
    }
}

• 제약 1) flow는 순차적으로 처리된다. suspend 함수를 호출할 경우, 해당 함수로부터 값이 리턴될 때까지 기다리게 된다.
• 제약 2) flow 빌더 함수를 쓸 때, producer는 서로 다른 CoroutineContext에서 값들을 emit 할 수 없다. 따라서 withContextwithContext와 같은 블록을 열어서 새로운 코루틴을 만들지 말라고 한다. 대신 callbackFlow를 활용할 수 있다. (해당 내용은 다음 포스팅에서 자세히 다뤄보도록 하자.)

4. Flow Builder Function

1. flow : 내부에서 사용자가 원하는 대로 cold flow 로직을 구성할 수 있음
2. flowOf : 지정된 값 emit 하기. (Ex. flowOf("A", "B", "C") )
3. asFlow : kotlin collection(list, set, array)
   • 직접 테스트해본 결과 list, set, array는 바로 asFlow 호출 가능
   • map은 `map.asIterable().asFlow()`와 같이 iterable 변환 후 호출 가능

fun simpleFlowBuilder(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(1000)
        emit(i)
    }
}
​
fun simpleFlowOfBuilder(): Flow<String> = flowOf("A", "B", "C")
​
fun simpleAsFlowBuilder(): Flow<Double> {
    return listOf(1.1, 2.2, 3.3).asFlow()
}
​
fun main() = runBlocking {
    // Using flow{}
    println("Using flow{} builder function:")
    simpleFlowBuilder().collect { value ->
        println("Collected value: $value")
    }
​
    // Using flowOf{}
    println("\nUsing flowOf{} builder function:")
    simpleFlowOfBuilder().collect { value ->
        println("Collected value: $value")
    }
​
    // Using asFlow{}
    println("\nUsing asFlow{} builder function:")
    simpleAsFlowBuilder().collect { value ->
        println("Collected value: $value")
    }
}
​