복수의 Flow를 합치는 방법으로 zip 연산자와 combine 연산자가 존재한다.
1. zip()
public fun <T1, T2, R> Flow<T1>.zip(other: Flow<T2>, transform: suspend (T1, T2) -> R): Flow<R> = zipImpl(this, other, transform)zip 함수는 두 개의 플로우로부터 쌍을 만드는 함수이다. Flow의 확장함수 형태로 제공되며, 다른 Flow와 transform 블록을 전달하면 된다. 여기서 쌍은 영어로 pair라고 번역할 수 있는데 Kotlin에서 기본적으로 제공되는 Pair data class는 아니므로 헷갈리지 않도록 주의하자.
fun main() = runBlocking<Unit> {
val flow1 = flowOf("A", "B", "C")
.onEach { delay(400) }
val flow2 = flowOf(1, 2, 3, 4)
.onEach { delay(1000) }
val startTime = System.currentTimeMillis()
flow1.zip(flow2) { f1, f2 -> "${f1}_${f2}" }
.collect {
println(it)
println(getElapsedTime(startTime))
}
}
fun getElapsedTime(startTime: Long): String = "지난 시간: ${System.currentTimeMillis() - startTime}ms"
/* 결과 :
A_1
지난 시간: 1039ms
B_2
지난 시간: 2033ms
C_3
지난 시간: 3037ms
*/예시에서는 flow1에서 생산된 값과 flow2에서 생산된 값 하나의 쌍이 되어, "${f1}_${f2}"형태의 String을 생산하는 새로운 Flow를 생성한다. flow1의 값이 생산되는 데 400ms가 소요되고 flow2의 값이 생산되는 데 1000ms(1초)가 소요되는데, 생산된 값은 하나의 쌍의 일부가 되기 때문에 다른 Flow의 쌍이 될 값을 기다려야 한다.
flow1에서 A가 생산될 때는 아직 flow2의 1이 생산되지 않은 시점이다. flow1에서 A가 생산되어도 flow2의 1이 생산될 때까지 기다려야 한다. 그래서 flow2의 1이 생산된 1초가 지나고 나서야 collect 블록의 코드가 실행된다.
지금은 flow1이 데이터를 생산하는 데 더 짧은 시간이 걸리지만, delay(400)과 delay(1000)의 위치를 바꿔도 결과는 동일하다. zip함수가 두 Flow의 값이 생산될 때까지 기다렸다가 결합하는 함수이기 때문이다.
그리고 두 개의 Flow 중 하나가 완료되면 남은 Flow는 cancel이 호출되어 취소된다. 예시에서도 flow2의 데이터 4는 쌍을 이루지 못하고 유실된다.
2. combine()
zip 함수는 두 개의 생산된 값을 합쳐서 하나의 값으로 내보내는 반면에, combine은 값이 생산될 때마다 합쳐서 하나의 값으로 내보낸다. zip은 두 개의 Flow 중 하나의 Flow가 완료되면 나머지 Flow가 cancel되어 zip 함수가 종료되지만, combine은 그런 제한이 없기 때문에 두 Flow가 모두 완료될 때까지 실행된다.
fun main() = runBlocking<Unit> {
val flow1 = flowOf("A", "B", "C")
.onEach { delay(400) }
val flow2 = flowOf(1, 2, 3, 4)
.onEach { delay(500) }
val startTime = System.currentTimeMillis()
flow1.combine(flow2) { f1, f2 -> "${f1}_${f2}" }
.collect {
println(it)
println(getElapsedTime(startTime))
}
}
/*
A_1
지난 시간: 543ms
B_1
지난 시간: 835ms
B_2
지난 시간: 1048ms
C_2
지난 시간: 1239ms
C_3
지난 시간: 1549ms
C_4
지난 시간: 2055ms
*/예시 코드의 과정을 살펴보자.
1. flow1에서 A 생산.
2. flow2에서 1 생산. 이때 flow1의 최신값은 A이기 때문에 combine된 결과는 A_1
3. flow1에서 B 생산. 이때 flow2의 최신값은 1이기 때문에 combine된 결과는 B_1
4. flow2에서 2 생산. 이때 flow1의 최신값은 B이기 때문에 combine된 결과는 B_2
5. flow1에서 C 생산. 이때 flow2의 최신값은 2이기 때문에 combine된 결과는 C_2
이후 과정도 비슷하게 이루어진다.
flow2의 마지막 값인 4의 생산이 끝나고 나서야 combine이 종료된다.
정리하면, 두 개의 Flow에서 새로운 값이 각각 생산될 때마다, 그 시점에서 각 Flow의 최신값을 합치는 함수가 combine 함수이다.
한가지 주의할 점이 있다. 두 개의 Flow에서 생산된 값이 존재해야만 combine이 이루어진다는 것이다.
fun main() = runBlocking<Unit> {
val flow1 = flowOf("A", "B", "C")
.onEach { delay(400) }
val flow2 = flowOf(1, 2, 3, 4)
.onEach { delay(1000) }
val startTime = System.currentTimeMillis()
flow1.combine(flow2) { f1, f2 -> "${f1}_${f2}" }
.collect {
println(it)
println(getElapsedTime(startTime))
}
}
/* 결과 :
B_1
지난 시간: 1043ms
C_1
지난 시간: 1243ms
C_2
지난 시간: 2051ms
C_3
지난 시간: 3055ms
C_4
지난 시간: 4059ms
*/만약 위와 같이 flow1에서 A와 B가 생산될 동안 flow2에서 데이터가 생산되지 않는다면 combine은 이루어지지 않는다. flow2가 최초로 데이터를 생산하는 시점(최초 실행 1초 후)이 되어서야 combine이 이루어지고, 그 시점에서의 flow1 최신 데이터는 B이기 때문에 A_1가 아닌 B_1부터 출력된다.
3. combine 사용사례
zip 함수는 잘 사용되는 것을 못봐서 사용사례를 잘 모르겟지만 combine은 흔히 사용되는 함수라서 내가 진행한 샘플 프로젝트에서 예시를 가지고 왔다. 북마크된 아이템 목록을 방출하는 Flow와 검색된 아이템 목록을 방출하는 Flow(getSavedDocumentsUseCase()와 _searchedItems)를 combine하여 아이템의 isFavorite을 true로 처리할지 false로 처리할지를 결정하는 예시이다.
@HiltViewModel
class SearchViewModel @Inject constructor(
getSavedDocumentsUseCase: GetSavedDocumentsUseCase
) : BaseViewModel() {
private val _searchedItems = MutableStateFlow<List<DocumentListItem>?>(null)
val uiState: StateFlow<SearchUiState> = combine(
getSavedDocumentsUseCase(),
_searchedItems
) { savedDocuments, searchedItems ->
when {
searchedItems == null -> SearchUiState.Idle
searchedItems.isEmpty() -> SearchUiState.Empty
else -> {
val updatedItems: List<DocumentListItem> = searchedItems.map { searchedItem ->
if (searchedItem is DocumentModel) {
if (savedDocuments.any { it.thumbnailUrl == searchedItem.thumbnailUrl }) {
searchedItem.copy(isFavorite = true)
} else {
searchedItem.copy(isFavorite = false)
}
} else {
searchedItem
}
}
SearchUiState.Success(updatedItems)
}
}
}.catch {
emit(SearchUiState.Error(it))
}.stateIn(
scope = this,
started = SharingStarted.WhileSubscribed(),
initialValue = SearchUiState.Idle
)
// ...
}이 코드는 북마크된 아이템 목록이 변경될 때마다, 그리고 검색된 아이템 목록이 변경될 때마다 combine이 동작하여 uiState를 새로운 SearchUiState로 초기화한다.
참고로 zip은 쌍으로 값을 이루기 때문에 두 개의 Flow로만 가능하다. 하지만 combine은 그런 제한이 없기 때문에 여러 개의 Flow를 combine하는 것이 가능하다.
참고자료
