[Flow] 연산자 내부코드를 열어보자!

안녕하세요. 🤗

안드로이드 개발을 위해 Flow를 공부하던 중, 여러 Flow 연산자에 대해

내부 코드 내용도 함께 정리해두면 좋을거 같아서!

이번 포스트를 작성하게 되었습니다.

 

코드의 출저는 😙

안드로이드 스튜디오에서 직접 열어보았습니다.ㅎㅎ

 

flow의 연산자를 공부해보면 알 수 있듯이,

collection 연산자와 매우! 비슷합니다.👍🏻

 

1. map

- 데이터를 가공하는 작업을 수행합니다.

public inline fun <T, R> Flow<T>.map(
	crossinline transform: suspend (value: T) -> R
): Flow<R> = transform { value ->
    return@transform emit(transform(value))
}

 

2. transform

- map의 코드 내에서 호출되는 transform 연산자 메서드입니다. flow를 원하는 방식으로 수정하려고 할 때 사용할 수 있습니다.

public inline fun <T, R> Flow<T>.transform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow {
    collect { value ->
        return@collect transform(value)
    }
}

 

코드를 보면 map과 비슷하게 값을 변경하는 연산자인 것 같아 보이지만

FlowCollector가 수신객체로 지정된 메서드라는 점에 유의하셔야 합니다 🤔

 

그리고 FlowCollection는 아래와 같이 구현되어 있습니다.

public fun interface FlowCollector<in T> {
    public suspend fun emit(value: T)
}

 

transform을 더 확실히 알기 위해 다른 예제도 확인해보겠습니다. 

flowOf(1,2,3,4).transform{
	emit(it)
    emit(it*2)
}.collect {
	print("$it / ")
}

위와 같은 코드가 있다면 어떻게 출력될까요?

flow 내에 있는 emit이 순서대로 방출(?)되어

📌 1 / 2 / 2 / 4 / 3 / 6 / 4 / 8 / 

이렇게 출력될 것입니다.

 

3. filter

- filter도 map과 마찬가지로 내부에서 transform을 호출하는 것을 알 수 있습니다.

public inline fun <T> Flow<T>.filter(
	crossinline predicate: suspend (T) -> Boolean
): Flow<T> = transform { value ->
    if (predicate(value)) return@transform emit(value)
}

하지만 map과는 다르게 값을 변경하는 것이 아닌 predicate(value)가 true인 경우에만 emit을 하고 있습니다.

 

3.1 filterNot

- 이름에서 유추할 수 있듯이, filter와 반대로 전달된 술어가 false를 반환하는 경우에만 emit을 하는 연산자 메서드입니다.

public inline fun <T> Flow<T>.filterNot(
	crossinline predicate: suspend (T) -> Boolean
): Flow<T> = transform { value ->
    if (!predicate(value)) return@transform emit(value)
}

 

4. take

fun <T> Flow<T>.take(count: Int): Flow<T>

- 전달된 count만큼의 결과가 사용하는 연산자 메서드입니다.

 

예를 들어, count = 3이라면 flow의 collect가 시작되고 첫 3개의 결과값만 받아와 사용하고, 그 이후에는 Exception을 발생시켜 flow를 종료합니다.

public fun <T> Flow<T>.take(count: Int): Flow<T> {
    require(count > 0) { "Requested element count $count should be positive" }
    return flow {
        var consumed = 0
        try {
            collect { value ->
                if (++consumed < count) {
                    return@collect emit(value)
                } else {
                    return@collect emitAbort(value)
                }
            }
        } catch (e: AbortFlowException) {
            e.checkOwnership(owner = this)
        }
    }
}

 

코드에서 볼 수 있듯, count만큼만 실행시키기 위해 consumed 변수를 두어 조건문에 검사하고 있습니다.

또한,

private suspend fun <T> FlowCollector<T>.emitAbort(value: T) {
    emit(value)
    throw AbortFlowException(this)
}

count만큼 실행되었을 때 호출되는 emitAbort에서 Exception을 발생시키고 있습니다.

 

4.1 takeWhile

- take와 유사하게 collect가 시작되고 전달된 술어가 false가 될때까지의 결과값만을 사용하는 연산자 메서드입니다.🤗

 

예를 들어, emit된 값이 5보다 작을 때까지만 사용하고자 한다면 아래와 같이 구현할 수 있습니다.

flowOf(1,2,3,4,5,6)
    .takeWhile { it < 5 }

public fun <T> Flow<T>.takeWhile(predicate: suspend (T) -> Boolean): Flow<T> = flow {
    // This return is needed to work around a bug in JS BE: KT-39227
    return@flow collectWhile { value ->
        if (predicate(value)) {
            emit(value)
            true
        } else {
            false
        }
    }
}

 

- take와 달리 횟수를 세는 것이 아닌, 술어(predicate)를 실행하고 있는 것을 알 수 있습니다.

 

5. drop

fun <T> Flow<T>.drop(count: Int): Flow<T>

- take와는 반대로 collect가 시작되고 첫 count개의 결과값은 사용하지 않고 버리는 연산자 메서드입니다.

 

public fun <T> Flow<T>.drop(count: Int): Flow<T> {
    require(count >= 0) { "Drop count should be non-negative, but had $count" }
    return flow {
        var skipped = 0
        collect { value ->
            if (skipped >= count) emit(value) else ++skipped
        }
    }
}

 

5.1 dropWhile

-take와 마찬가지로 count가 아닌, 조건문으로 결과값을 사용하지 않고 버릴지 말지 정의할 수 있는 dropWhile이 있습니다.

- 술어의 결과값이 false이기 전까지 flow에서 emit된 값을 사용하지 않고 넘어갑니다.

 

public fun <T> Flow<T>.dropWhile(predicate: suspend (T) -> Boolean): Flow<T> = flow {
    var matched = false
    collect { value ->
        if (matched) {
            emit(value)
        } else if (!predicate(value)) {
            matched = true
            emit(value)
        }
    }
}

 

코드에서도 알 수 있듯이 matched값이 false인 경우에는 아무런 동작을 하지 않습니다.

 

6. reduce

- emit된 값들을 누진적으로 계산합니다.

🌹 reduce는 종단 연산자이기 때문에 flow가 아닌, 특정 값을 반환합니다.

 

예를 들어서!

val result = flowOf(1,2,3,4,5,6)
	.reduce { ecc, value -> acc + value }

위와 같을 경우, result의 값은 얼마일까요?

 

처음 emit된 값이 1이고, 이후에 emit되는 2,3,4,5,6 이 순차적으로 누적되어

최종적으로 반환된 결과값은 이 모두를 더한 21일 것입니다.

 

내부 코드는 아래와 같습니다.

public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S, value: T) -> S): S {
    var accumulator: Any? = NULL

    collect { value ->
        accumulator = if (accumulator !== NULL) {
            @Suppress("UNCHECKED_CAST")
            // accumulator와 현재값을 이용해 operation 연산 수행
            // 그 결과를 다시 accumulator에 저장
            operation(accumulator as S, value)
        } else {
        	// 첫 emit된 값인 경우에는 그대로 사용
            value
        }
    }

    if (accumulator === NULL) throw NoSuchElementException("Empty flow can't be reduced")
    @Suppress("UNCHECKED_CAST")
    return accumulator as S
}

 

6.1 fold

- fold는 reduce와 동일한 작업을 수행하는 연산자 메서드입니다.

- 유일한 차이점은 reduce는 가장 처음 emit된 값을 초기값으로 사용한다면, fold는 개발자가 직접 초기값을 매개변수로 전달한다는 점입니다.

public suspend inline fun <T, R> Flow<T>.fold(
    initial: R,
    crossinline operation: suspend (acc: R, value: T) -> R
): R {
    var accumulator = initial
    collect { value ->
        accumulator = operation(accumulator, value)
    }
    return accumulator
}

 

때문에 코드도 reduced의 코드에서 초기값이 설정되어 있는지 아닌지 검사하는 if문만 없어졌을 뿐이지 동일하게 수행되는 것을 알 수 있습니다.👍🏻

 

7. count

suspend fun <T> Flow<T>.count(predicate: suspend (T) -> Boolean): Int

- flow에서 emit된 값들 중 전달된 술어(predicate)가 true인 값의 개수를 반환하는 종단 연산자 메서드입니다.

- 때문에 flow가 아닌 Int 타입의 값을 반환합니다.

 

코드는 매우 간단합니다. 🤗

public suspend fun <T> Flow<T>.count(predicate: suspend (T) -> Boolean): Int {
    var i = 0
    collect { value ->
        if (predicate(value)) {
            ++i
        }
    }

    return i
}

 

물론! 조건문 없이, flow에서 emit된 값들의 전체 개수를 반환하는(마치 Collection의 size와 같이) 메서드 또한 존재합니다.👏🏻

👇🏻 요렇게!

public suspend fun <T> Flow<T>.count(): Int  {
    var i = 0
    collect {
        ++i
    }

    return i
}

 

---

안드로이드를 독학해보면서 수강하고 있는 코루틴 강의에서 학습한 flow 연산자들을

복습하는 겸 정리해보았습니다.

 

감사합니다.😌