Thiyagu
12/30/2019, 7:05 PMchannelFlow in generateData method inherit the coroutinescope passed in the launchIn method? https://gist.github.com/thiyagu06/2afd4437ff9d0eccacafff0bd6216874
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.channels.awaitClose
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.util.concurrent.Executors
import kotlin.coroutines.CoroutineContext
import kotlin.random.Random
class TryChannelFlow : CoroutineScope {
private val supervisorJob = SupervisorJob()
override val coroutineContext: CoroutineContext
get() = Executors.newFixedThreadPool(3).asCoroutineDispatcher()
private val scope = CoroutineScope(supervisorJob + coroutineContext)
fun start() {
val data: Flow<Int> = DataProvider().generateData()
data.onEach{
println("received -->$it")
}.launchIn(scope)
}
fun stop() {
supervisorJob.cancel()
}
}
class DataProvider {
fun generateData(): Flow<Int> {
return channelFlow {
val job = launch {
while (isActive) {
println("sending on ${Thread.currentThread().name}")
channel.send(Random.nextInt(1000)) // pretend receiving data from non-blocking io
delay(100)
}
}
awaitClose {
job.cancel()
}
}
}
}
fun main() = runBlocking {
val tryChannelFlow = TryChannelFlow()
tryChannelFlow.start()
delay(3000) // pretend app crashes/terminated
tryChannelFlow.stop()
println("completed")
}