Flow 이해하기 2(flow 연산자)

지난시간에는 flow가 어떻게 생겼는지, 그리고 구독자와 발행자가 어떻게 값을 주고 받는지에 대해서 알아보았습니다.

이번 포스팅에서는 Flow의 연산자가 어떤것이 있는지 각 연산자가 수행하는 것이 무엇인지에 대해서 알아보겠습니다.

 

collect 연산자(종단 연산자)

Flow 이해하기 1 에서 아주 간략하게 설명 드렸습니다만 그래고 Flow의 연산자들을 이해하고 공부해보는 시간이니 만큼 좀 더 자세하게 알아보겠습니다.

collect는 emit을 통해 값을 받을 수 있는 연산자 입니다. 

값을 방출 하는것이 emit 이라면, 구독자 측에서 값을 받는 연산자를 종단 연산자라고 합니다. 

그리고 collect는 대표적인 종단 연산자 라고 합니다.

fun flow(): Flow<Int> = flow {
    repeat(5) {
        emit(Random.nextInt(0, 500))
    }
}

fun main() = runBlocking {
    flow().collect {
        println(it)
    }
}

//실행결과 
441
323
155
397
490

flow 함수를 실행하였고 emit으로 방출한 값을 collect가 받고 있습니다.

 

map 연산자 (중간 연산자)

중간 연산자는 상위 플로우가 하위 플로우에게 값을 보내면서 한번 거치게 되는 연산자 입니다.

 

map 연산자는 수신측에서 개발자가 원하는대로 매핑 할 수 있습니다.

fun flow(): Flow<Int> = flow {
    emit(Random.nextInt(0, 500))
}

fun main() = runBlocking {
    flow().map {
        "${it * 2}"
    }.collect { value ->
        println(value)
        println(value::class.java)
    }
}

//실행결과
682
class java.lang.String

발신자 측에서는 0부터 500까지의 랜덤한 값을 흘려 보냈고 수신자 측에서는 collect 하기 전 받은 값의 2배를 곱하였고 이를 String으로 변환 하였습니다. 

그 결과 수신측에서 받은 값의 자료형을 보니 String으로 변환 된 것을 알 수 있습니다.

 

Kotlin의 컬렉션 함수 중에서도 map{} 함수가 있으니 그 함수를 생각해보시면 보다 이해가 빠르실 것 같습니다.

 

filter 연산자(중간 연산자)

fliter 연산자는 filter 블럭안에 조건을 명시 함으로써 그 조건이 true 인 것 들만 collect로 흘려 보냅니다.

fun main() = runBlocking<Unit> {
    (1..20).asFlow().filter {
        (it % 2) == 0
    }.collect {
        println(it)
    }
}

//실행결과
2
4
6
8
10
12
14
16
18
20

 

내부적으로 까보면 

public inline fun <T> kotlinx.coroutines.flow.Flow<T>.filter(
		crossinline predicate: suspend (T) -> kotlin.Boolean
        ): kotlinx.coroutines.flow.Flow<T>

콜백으로 Boolean을 보내는 것을 알 수 있습니다.

 

transform 연산자 (변환 연산자)

우선 코드 실행 전에 transform이 어떻게 생겼는지 부터 알아 보겠습니다.

public inline fun <T, R> kotlinx.coroutines.flow.Flow<T>.transform(
		@kotlin.BuilderInference crossinline 
        transform: suspend kotlinx.coroutines.flow.FlowCollector<R>.(T) -> kotlin.Unit
        ): kotlinx.coroutines.flow.Flow<R> {
}

transform은 내부적으로 FlowCollector를 받고 있습니다.

때문에 emit 연산자를 사용가능하다는 이 점이 있는데요.

예제를 먼저 써보자면 

fun main() = runBlocking<Unit> {
    (1..5).asFlow().transform {
        emit(it)
        emit(it - 1)
    }.collect {
        println(it)
    }
}

//실행결과 
1
0
2
1
3
2
4
3
5
4

transform 연산자는 FlowCollector 인터페이스를 받고 있습니다.

때문에 emit() 함수를 사용가능 하고 이에 따라 값을 방출 가능한 연산자 입니다.

 

take 연산자

public fun <T> kotlinx.coroutines.flow.Flow<T>.take(
		count: kotlin.Int
	): kotlinx.coroutines.flow.Flow<T> { /* compiled code */ }

take 연산자는 구독자 측에서 값을 받을때 최대 몇번 까지 받을 껀지 정수를 인자값으로 넣어 정의 할 수 있는 함수 입니다.

예제를 보겠습니다.

fun main() = runBlocking<Unit> {
    (1..10).asFlow().transform {
        emit(it)
    }.take(5).collect {
       println(it)
    }
}

//실행 결과
1
2
3
4
5

1 부터 10까지 emit 하여도 5까지 찍히는 것을 확인 할 수 있습니다.

좀 더 정확하게 말하자면 인자로 받고 있는 값 만큼 collect 하고 난 뒤 함수를 바로 끝내버리는 특징이 있습니다.

 

drop 연산자

take 연산자는 인자 값 까지만 받는다면, drop 연산자는 인자값 까지 버리고 난 뒤 방출해주는 특징이 있습니다. 

public fun <T> kotlinx.coroutines.flow.Flow<T>.drop(
		count: kotlin.Int
	): kotlinx.coroutines.flow.Flow<T> { /* compiled code */ }

예제를 통해 살펴보겠습니다.

fun main() = runBlocking<Unit> {
    (1..10).asFlow().transform {
        emit(it)
    }.drop(5)
    .collect {
        println(it)
    }
}

// 실행결과 
6
7
8
9
10

drop의 인자값만큼 emit함수를 무시하고 난 다음 emit 하는 연산자 입니다.

 

reduce 연산자

reduce 는 값 두 가지를 리턴합니다.

public suspend fun <S, T : S> kotlinx.coroutines.flow.Flow<T>.reduce(
		operation: suspend (S, T) -> S
	): S { /* compiled code */ }

연산결과 S를 리턴합니다. 예제를 보겠습니다.

fun main() = runBlocking<Unit> {
    val value = (1..10)
        .asFlow()
        .reduce { a, b ->
            a + b
        }
    println(value)
}

// 실행결과 
55

1 부터 10까지 더 하면서 a를 리턴 하고 있습니다.