1 /*
<lambda>null2 * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
3 */
4
5 // This file was automatically generated from coroutines-guide-reactive.md by Knit tool. Do not edit.
6 package kotlinx.coroutines.rx2.guide.operators04
7
8 import kotlinx.coroutines.*
9 import kotlinx.coroutines.reactive.*
10 import org.reactivestreams.*
11 import kotlin.coroutines.*
12
13 fun <T> Publisher<Publisher<T>>.merge(context: CoroutineContext) = publish<T>(context) {
14 collect { pub -> // for each publisher collected
15 launch { // launch a child coroutine
16 pub.collect { send(it) } // resend all element from this publisher
17 }
18 }
19 }
20
<lambda>null21 fun CoroutineScope.rangeWithInterval(time: Long, start: Int, count: Int) = publish<Int> {
22 for (x in start until start + count) {
23 delay(time) // wait before sending each number
24 send(x)
25 }
26 }
27
<lambda>null28 fun CoroutineScope.testPub() = publish<Publisher<Int>> {
29 send(rangeWithInterval(250, 1, 4)) // number 1 at 250ms, 2 at 500ms, 3 at 750ms, 4 at 1000ms
30 delay(100) // wait for 100 ms
31 send(rangeWithInterval(500, 11, 3)) // number 11 at 600ms, 12 at 1100ms, 13 at 1600ms
32 delay(1100) // wait for 1.1s - done in 1.2 sec after start
33 }
34
<lambda>null35 fun main() = runBlocking<Unit> {
36 testPub().merge(Dispatchers.Unconfined).collect { println(it) } // print the whole stream
37 }
38