Marc Knaup
10/11/2020, 9:06 PMList<Flow<T>>
into Flow<List<T>>
where List<T>
contains the latest value of each Flow<T>
at the same index. combine()
is not an option.MiSikora
10/12/2020, 8:49 AMfun <T> Iterable<Flow<T>>.unwrap(): Flow<List<T>> {
val emptyFlow = emptyFlow<List<T>>()
return fold(emptyFlow) { flows, flow ->
if (flows === emptyFlow) {
flow.map(::listOf)
} else {
flows.combine(flow) { group, item -> group + item }
}
}
}
Marc Knaup
10/12/2020, 9:36 AMMiSikora
10/12/2020, 9:36 AMbezrukov
10/12/2020, 10:37 AMyield
each time when element offered with fast path. As far as I understand it's necessary for sequential flows.listOf(flow1,flow2,flow3).combine()
you can expect that it will use order of these flows. With your extension you can't expect this. It may lead to the following behavior:
time: | 1 | 2 | 3 | 4 | 5
flow1: | A | | B | |
flow2: | w | x | y | z |
combine:| Aw| Ax| Ay| Az| Bz| <-your extension
stdlib: | Aw| Ax| Bx| By| Bz|
Marc Knaup
10/12/2020, 11:47 AMbezrukov
10/12/2020, 11:49 AMOr is it elements that happen to be collectible in the same “tick”?yes, it's just shows that at some time called "3" B and y are ready for collect
Marc Knaup
10/12/2020, 11:52 AM