mirror of
https://github.com/InsanusMokrassar/TelegramBotAPI.git
synced 2024-06-29 04:47:48 +00:00
40 lines
972 B
Kotlin
40 lines
972 B
Kotlin
package com.github.insanusmokrassar.TelegramBotAPI.extensions.utils
|
|
|
|
import kotlinx.coroutines.CoroutineScope
|
|
import kotlinx.coroutines.channels.BroadcastChannel
|
|
import kotlinx.coroutines.channels.Channel
|
|
import kotlinx.coroutines.flow.*
|
|
|
|
/**
|
|
* Analog of [merge] function for [Flow]s. The difference is in the usage of [BroadcastChannel] in this case
|
|
*/
|
|
fun <T> aggregateFlows(
|
|
withScope: CoroutineScope,
|
|
vararg flows: Flow<T>,
|
|
internalBufferSize: Int = Channel.BUFFERED
|
|
): Flow<T> {
|
|
val bc = BroadcastChannel<T>(internalBufferSize)
|
|
flows.forEach {
|
|
it.onEach {
|
|
safely { bc.send(it) }
|
|
}.launchIn(withScope)
|
|
}
|
|
return bc.asFlow()
|
|
}
|
|
|
|
fun <T> Flow<Iterable<T>>.flatMap(): Flow<T> = flow {
|
|
collect {
|
|
it.forEach {
|
|
emit(it)
|
|
}
|
|
}
|
|
}
|
|
|
|
fun <T, R> Flow<T>.flatMap(mapper: (T) -> Iterable<R>): Flow<R> = flow {
|
|
collect {
|
|
mapper(it).forEach {
|
|
emit(it)
|
|
}
|
|
}
|
|
}
|