Merge

두개의 flows를 하나로 결합하는 것에 대하여 3가지 방법이 있습니다. 가장 쉬운것은 두개의 flow를 하나로 만들기 위해 합치는 것입니다. 이미 있는 flows들의 원소들의 데이터들이 만들어진 것을 수정하지 않고 merge 할 수 있습니다.

예제1

suspend fun main() {
    val ints: Flow<Int> = flowOf(1, 2, 3)
    val doubles: Flow<Double> = flowOf(0.1, 0.2, 0.3)

    val together: Flow<Number> = merge(ints, doubles)
    print(together.toList())
}

[결과]

[1, 0.1, 0.2, 0.3, 2, 3]
or [1, 0.1, 0.2, 0.3, 2, 3]
or [0.1, 1, 2, 3, 0.2, 0.3]
or any other combination

merge를 사용할때 주의할 점은 하나의 flow는 다른 flow들을 기다리지 않는 다는 점입니다. 대신에 아래처럼 사용하면 첫 flow가 delay되지만 다만 이것은 두번째 flow가 멈추게 되는데 관여하지 않습니다.

예제2

suspend fun main() {
    val ints: Flow<Int> = flowOf(1, 2, 3)
        .onEach { delay(1000) }
    val doubles: Flow<Double> = flowOf(0.1, 0.2, 0.3)

    val together: Flow<Number> = merge(ints, doubles)
    together.collect { println(it) }
}

[결과]

// 0.1
// 0.2
// 0.3
// (1 sec)
// 1
// (1 sec)
// 2
// (1 sec)
// 3

동일한 작업으로 이어져야 하는 여러 이벤트 소스가 있을 때 merge을 사용합니다.

fun listenForMessages() {
    merge(userSentMessages, messagesNotifications)
        .onEach { displayMessage(it) }
        .launchIn(scope)
}

Zip

zip은 양쪽의 flows들의 짝을 만들어 줄 수 있는 function 입니다. 어떤 elements들이 쌍으로 묶일때 어떻게 할 지 정할 수 있습니다. (이것은 새로운 flow를 생성하여 emit될때 변환하게 됩니다) 각 element들은 하나의 쌍으로 묶이는데 기다리는 시간이 각각 필요하게 됩니다. elements들이 쌍이 없이 남게 된 경우에도 해당 flow들은 완료되었다고 보고 flow의 결과 또한 완료 되었다고 봅니다.

suspend fun main() {
    val flow1 = flowOf("A", "B", "C")
        .onEach { delay(400) }
    val flow2 = flowOf(1, 2, 3, 4)
        .onEach { delay(1000) }
    flow1.zip(flow2) { f1, f2 -> "${f1}_${f2}" }
        .collect { println(it) }
}

[결과]

// (1 sec)
// A_1
// (1 sec)
// B_2
// (1 sec)
// C_3

zip은 춤인 폴로네즈를 떠오르게 합니다. 이 춤의 특징중 하나는 한쌍씩 구성된 줄이 중간에서 분리되었다가 다시 만날때 재구성 됩니다

Combine

마지막으로 combine은 두개의 flows들을 결합합니다. 각 요소들이 쌍으로 결합된 마치 zip 처럼 적용됩니다. 만약 처음 flow가 다른 flow보다 늦게 걸려도 기다려주기까지 합니다. 그러나 이것은 폴로네즈 춤과는 다릅니다. combine을 사용하게 되면 새로운 요소들이 대체될때 이전의 것을 사용하게 될 수 있습니다. 한 쌍이 이미 구성되어 있더라도 새로운쌍을 만들때 다른 flow의 이전 요소의 값에서 새롭게 구성될 수 있습니다

zip이 새로운 쌍을 만들때 첫번째 flow가 끝났을경우 combine은 여기서 제한을 두지 않고 두개의 flow들이 끝날때까지 기다려서 emit됩니다.

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*

suspend fun main() {
    val flow1 = flowOf("A", "B", "C")
        .onEach { delay(400) }
    val flow2 = flowOf(1, 2, 3, 4)
        .onEach { delay(1000) }
    flow1.combine(flow2) { f1, f2 -> "${f1}_${f2}" }
        .collect { println(it) }
}

[결과]

// (1 sec)
// B_1
// (0.2 sec)
// C_1
// (0.8 sec)
// C_2
// (1 sec)
// C_3
// (1 sec)
// C_4

combine은 두개의 소스의 변화를 활동을 관찰하는데 주로 사용됩니다. 만약 변화가 발생할때마다 요소의 emit이 필요할 경우 각 combined flow들의 초기 값을 지정해줄 수 있습니다.

userUpdateFlow.onStart { emit(currentUser) }

전형적인 사용방법으로는 두개의 요소의 변화가 하나의 요소를 접근하는 경우 사용됩니다. 예를 들어서 notification badge가 현재 상태의 사용자 정보와 같은 notifications를 사용하는데 사용될 두개의 같은 변경사항의 값을 한개의 뷰에 변경할때 사용합니다.

userStateFlow
    .combine(notificationsFlow) { userState, notifications ->
        updateNotificationBadge(userState, notifications)
    }
    .collect()

Reference

• https://kt.academy/article/cc-flow-combine