[kotlin] Composing multiple flows

드디어 플로우를 합쳐본다.

Composing multiple flows

Zip

Sequence.zip 확장함수처럼 flows의 zip도 두 flow의 각 상응하는 값을 합칠 수 있다.

fun main() = runBlocking<Unit> {
    val nums = (1..3).asFlow() // numbers 1..3
    val strs = flowOf("one", "two", "three", "four") // strings
    nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string
        .collect { println(it) } // collect and print
}

• 결과

1 -> one
2 -> two
3 -> three

four에 상응하는 nums의 value가 없어서 출력되지 않는다. 반대의 경우도 마찬가지

Combine

flow가 가장 최신의 값만 보여주는 경우, upstream이 새로운 값을 방출할 때마다 대응하는 값으로 다시 연산해야 한다. 이 연산자들을 combine이라 부른다.

숫자는 300ms마다, string 업데이트는 400ms마다 이뤄지는 경우

    val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
    val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
    val startTime = System.currentTimeMillis() // remember the start time
    nums.zip(strs) { a, b -> "$a -> $b" } // compose a single string with "zip"
        .collect { value -> // collect and print
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }

• 결과

1 -> one at 442 ms from start
2 -> two at 845 ms from start
3 -> three at 1253 ms from start

combine을 사용하면?

    val nums = (1..3).asFlow().onEach { delay(300) } // numbers 1..3 every 300 ms
    val strs = flowOf("one", "two", "three").onEach { delay(400) } // strings every 400 ms
    val startTime = System.currentTimeMillis() // remember the start time
    nums.combine(strs) { a, b -> "$a -> $b" } // compose a single string with "combine"
        .collect { value -> // collect and print
            println("$value at ${System.currentTimeMillis() - startTime} ms from start")
        }

• 결과

1 -> one at 446 ms from start
2 -> one at 657 ms from start
2 -> two at 852 ms from start
3 -> two at 972 ms from start
3 -> three at 1258 ms from start

300ms 때는 아직 strs에서 대응하는 값이 방출되지 않아서 출력이 되지 않았다.
400ms 때는 1과 one이 대응되었고
600ms 때는 2가 방출되면서 one이랑 대응됐다.
800ms 때는 two가 방출되면서 2와 대응되었고
900ms은 다시 3이랑 two가
1200ms 때 three랑 3가 대응되어 출력되었다.

아무래도 zip보다는 combine을 자주 쓰지 않을까? 하지만 꼭 대응되어야 하는 상황에서는 zip이 안정적으로 보인다.
자세한 구현은 Channel class랑 연관된 부분이 많아보여서 Channel 다룰때 다시 봐야겠다.