groupBy

Flow<T>

Flow<Flow<T>>

Flow<*>

transform

fun <A, K> Observable<A>.groupBy(keySelector: (A) -> K) : Observable<GroupedObservable<K,A>>

GroupedObservable

Observable

fun getKey(): K

groupBy

Flow

GroupedFlow

groupBy

Flow

groupBy

List

fun <A, K> Flow<A>.groupBy(keySelector: (A) -> K): Flow<Pair<K, Flow<A>>> {
    val upstream = this

    return flow {
        val downstream = this

        val groupMap: HashMap<K, SendChannel<A>> = HashMap()

        val error = try {
            upstream.collect { value: A ->
                val key = keySelector(value)

                groupMap.getOrPut(key) {
                    Channel<A>().also { consumer ->
                        val flow = flow { consumer.consumeEach { emit(it) } }
                        downstream.emit(key to flow)
                    }
                }.send(value)
            }

            null
        } catch (t: Throwable) {
            t
        }

        groupMap.values.forEach {
            it.close(error)
        }
    }
}

fun test() {
    val numFlow = flow {
        for (i in 1..10) {
            emit(i)
            delay(100)
        }
    }

    runBlocking {
        numFlow.groupBy { it % 2 == 0 }
            .flatMapMerge { (key, groupedFlow) ->
                groupedFlow.map { "$key and $it" }
            }
            .collect {
                println(it)
            }
    }
}