플로우 직접 구현하기

💡 flow 인터페이스와 flow 빌더가 실제로 어떻게 동작하는지 이해하기

Flow 직접 구현

• 플로우의 원리에 대해 이해하기 위해서 직접 코드로 구현하였습니다.

FlowCollector

• 람다식을 통해서 플로우 동작을 구현할 수 있습니다.
• 람다식은 또한 함수를 나타내는 파라미터를 가질 수 있습니다.
  • flow 동작을 이해하기 위해서 해당 파라미터를 emit으로 정의하였습니다.

// 이전의 코드는 함수를 파라미터로 전달합니다.
// 복잡한 함수형을 간결하게 만들기 위하여 함수형 인터페이스로 추상화합니다.
private suspend fun <T> before(){
    val f: suspend ((T) -> Unit) -> Unit = { emit ->
        emit("flow 방출" as T)
    }
}

• 이 때 emit은 중단 함수로 구현되어야 합니다.
  • flow는 비동기 스트림으로, 데이터를 순차적으로 방출하면서도 일시 중단이 가능해야 합니다.
  • 흐름을 중단하고 재개하는 방식을 활용할 수 있어야 합니다.
• 함수형 파라미터를 활용하면 코드가 복잡하고 가독성이 떨어집니다.
  • 이를 해결하기 위해서 FlowCollector 함수형 인터페이스를 정의하였습니다.

// 함수형 인터페이스를 화용하여 람다식으로 정의합니다.
fun interface FlowCollector<T> {
    suspend fun emit(value: T)
}

// 람다식 내부에 FlowColletor 타입의 리시버를 생성했습니다.
// 이를 활용하면 it.emit() 형식이 아닌 emit() 형식으로 호출이 가능합니다.
private suspend fun <T> after(){
    val f: suspend FlowCollector<T>.() -> Unit = {
        emit("flow 방출" as T)
    }
}

Flow 인터페이스

• 람다식을 전달하는 대신에, 인터페이스를 구현하는 객체를 만드는 것이 가독성과 유지보수에 좋습니다.

// 기존의 FlowCollector를 다음과 같이 사용할 수 있습니다.
// 람다식 전달은 가독성이 떨어지므로, 인터페이스를 구현한 객체를 생성합니다.
private suspend fun <T> before() {
    val f: suspend FlowCollector<T>.() -> Unit = {
        emit("flow 방출" as T)
    }
    f {
        println(it) // flow 방출
    }
}

• 인터페이스 이름을 Flow라 하고, 해당 인터페이스의 정의를 객체 표현식으로 래핑합니다.

interface Flow<T> {
    suspend fun collect(collector: FlowCollector<T>)
}

private suspend fun <T> after() {
    val flow: Flow<T> = object : Flow<T> {
        override suspend fun collect(collector: FlowCollector<T>) {
            collector.emit("flow 방출" as T)
        }
    }

    flow.collect {
        println(it)
    }
}

FlowBuilder

• 플로우 생성 로직을 매번 구현하기에는 복잡합니다.
• 이를 해결하기 위해서 플로우를 생성하는 flow 빌더를 제작하였습니다.

// 복잡한 플로우 생성 로직을 내부로 캡슐화합니다.
fun <T> flowBuilder(
    builder: suspend FlowCollector<T>.() -> Unit
) = object : Flow<T> {
    override suspend fun collect(collector: FlowCollector<T>) {
        collector.builder()
    }
}

Custom Flow

• Flow, FlowCollector, flowBuilder를 통해서 실제 Flow와 동일한 방식의 코드를 작성하였습니다.

// 플로우와 똑같이 동작합니다.
suspend fun main() {
    val f: Flow<String> = flowBuilder {
        emit("flow 방출 1")
        emit("flow 방출 2")
    }

    f.collect {
        println("수집1 : $it")
    }
    f.collect {
        println("수집2 : $it")
    }
}

• collect를 호출하면, flow 빌더를 호출할 때 넣은 람다식이 호출됩니다.
  • 이는 플로우 작동 원리와 같습니다.

Flow 처리 방식

• 플로우의 각 원소를 반환하는 map 함수도 쉽게 제작할 수 있습니다.
• flow를 새로 생성하는 로직이기 때문에 플로우 빌더를 활용합니다.
• 래핑하고 있는 플로우를 시작하게 되므로, 빌더 내부에서 collect 메서드를 호출합니다.
• 이 외에 onEach, onStart, filter도 쉽게 제작 가능합니다.

flowBuilder - map

// 원소를 변환하여 새로운 플로우 방출
fun <T, R> Flow<T>.map(
    trasformation: suspend (T) -> R
): Flow<R> = flowBuilder {
    collect {
        emit(trasformation(it))
    }
}

flowBuilder - filter

fun <T> Flow<T>.filter(
    predicate: suspend (T) -> Boolean
): Flow<T> = flowBuilder {
    collect {
        if (predicate(it)) {
            emit(it)
        }
    }
}

flowBuilder - onEach

fun <T> Flow<T>.onEach(
    action: suspend (T) -> Unit
): Flow<T> = flowBuilder {
    collect {
        action(it) // 추가 동작 수행
        emit(it)
    }
}

flowBuilder - onStart

fun <T> Flow<T>.onStart(
    action: suspend () -> Unit
): Flow<T> = flowBuilder {
    action() // 추가 동작을 한 번만 수행
    collect{
        emit(it)
    }
}

flowBuilder - flowOf

// 플로우 생성 함수로 listOf와 비슷하게 동작
fun <T> flowOf(vararg elements: T): Flow<T> = flowBuilder {
    elements.forEach {
        emit(it)
    }
}

Ex

suspend fun main() {
    flowOf("A", "B", "C")
        .filter {
            it != "A"
        }
        .onEach {
            println("B or C")
        }
        .onStart {
            println("시작되었습니다")
        }
        .map {
            delay(1000)
            it.plus(it)
        }
        .collect {
            println(it)
        }
}

동기로 작동하는 Flow

• 플로우 또한 중단 함수처럼 동기로 작동하기 때문에, 플로우가 완료될 때까지 collect 호출이 중단됩니다.

플로우는 새로운 코루틴을 시작하지 않습니다. 각각의 처리 단계는 동기로 실행되기 때문에, onEach 내부에 delay가 있으면 모든 원소가 처리되기 전이 아닌 각 원소 사이에 지연이 생깁니다.

suspend fun main() {
    flowOf("A","B","C")
        .onEach { delay(1000) }
        .collect{ println(it) }
}