coroutine-part-4

valentine-starter/src/test/kotlin/com/unicorn/learning/SimpleSelect.kt

@@ -0,0 +1,189 @@
+/*
+ *    Copyright (c) 2023.  lWoHvYe(Hongyan Wang)
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.unicorn.learning
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.selects.*
+import kotlin.random.Random
+
+@OptIn(ExperimentalCoroutinesApi::class)
+fun CoroutineScope.fizz() = produce {
+    while (true) { // sends "Fizz" every 500 ms
+        delay(500)
+        send("Fizz")
+    }
+}
+
+@OptIn(ExperimentalCoroutinesApi::class)
+fun CoroutineScope.buzz() = produce {
+    while (true) { // sends "Buzz!" every 1000 ms
+        delay(1000)
+        send("Buzz!")
+    }
+}
+
+suspend fun selectFizzBuzz(fizz: ReceiveChannel<String>, buzz: ReceiveChannel<String>) {
+    select { // <Unit> means that this select expression does not produce any result
+        fizz.onReceive { value ->  // this is the first select clause
+            println("fizz -> '$value'")
+        }
+        buzz.onReceive { value ->  // this is the second select clause
+            println("buzz -> '$value'")
+        }
+    }
+}
+
+fun mainS1() = runBlocking {
+    val fizz = fizz()
+    val buzz = buzz()
+    repeat(7) {
+        selectFizzBuzz(fizz, buzz)
+    }
+    coroutineContext.cancelChildren() // cancel fizz & buzz coroutines
+}
+
+suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
+    select {
+        a.onReceiveCatching {
+            val value = it.getOrNull()
+            if (value != null) {
+                "a -> '$value'"
+            } else {
+                "Channel 'a' is closed"
+            }
+        }
+        b.onReceiveCatching {
+            val value = it.getOrNull()
+            if (value != null) {
+                "b -> '$value'"
+            } else {
+                "Channel 'b' is closed"
+            }
+        }
+    }
+
+@OptIn(ExperimentalCoroutinesApi::class)
+fun mainS2() = runBlocking {
+    val a = produce {
+        repeat(4) { send("Hello $it") }
+    }
+    val b = produce {
+        repeat(4) { send("World $it") }
+    }
+    repeat(8) { // print first eight results
+        println(selectAorB(a, b))
+    }
+    coroutineContext.cancelChildren()
+}
+
+@OptIn(ExperimentalCoroutinesApi::class)
+fun CoroutineScope.produceNumbers(side: SendChannel<Int>) = produce<Int> {
+    for (num in 1..10) { // produce 10 numbers from 1 to 10
+        delay(100) // every 100 ms
+        select {
+            onSend(num) {
+                println("onSent $it")
+            } // Send to the primary channel
+            side.onSend(num) {
+                println("side.onSend $it")
+            } // or to the side channel
+        }
+    }
+}
+
+fun mainS3() = runBlocking {
+    val side = Channel<Int>() // allocate side channel
+    launch { // this is a very fast consumer for the side channel
+        side.consumeEach { println("Side channel has $it") }
+    }
+    produceNumbers(side).consumeEach {
+        println("Consuming $it")
+        delay(250) // let us digest the consumed number properly, do not hurry
+    }
+    println("Done consuming")
+    coroutineContext.cancelChildren()
+}
+
+fun CoroutineScope.asyncString(time: Int) = async {
+    delay(time.toLong())
+    "Waited for $time ms"
+}
+
+fun CoroutineScope.asyncStringsList(): List<Deferred<String>> {
+    val random = Random(11)
+    return List(12) { asyncString(random.nextInt(1000)) }
+}
+
+fun mainS4() = runBlocking {
+    val list = asyncStringsList()
+    val result = select {
+        list.withIndex().forEach { (index, deferred) ->
+            deferred.onAwait { answer ->
+                "Deferred $index produced answer '$answer'"
+            }
+        }
+    }
+    println(result)
+    val countActive = list.count { it.isActive }
+    println("$countActive coroutines are still active")
+}
+
+@OptIn(ExperimentalCoroutinesApi::class)
+fun CoroutineScope.switchMapDeferred(input: ReceiveChannel<Deferred<String>>) = produce<String> {
+    var current = input.receive() // start with first received deferred value
+    while (isActive) { // loop while not cancelled/closed
+        val next = select { // return next deferred value from this select or null
+            input.onReceiveCatching { update ->
+                update.getOrNull()
+            }
+            current.onAwait { value ->
+                send(value) // send value that current deferred has produced
+                input.receiveCatching().getOrNull() // and use the next deferred from the input channel
+            }
+        }
+        if (next == null) {
+            println("Channel was closed")
+            break // out of loop
+        } else {
+            current = next
+        }
+    }
+}
+
+fun CoroutineScope.asyncString(str: String, time: Long) = async {
+    delay(time)
+    str
+}
+
+fun main() = runBlocking {
+    val chan = Channel<Deferred<String>>() // the channel for test
+    launch { // launch printing coroutine
+        for (s in switchMapDeferred(chan))
+            println(s) // print each received string
+    }
+    chan.send(asyncString("BEGIN", 100))
+    delay(200) // enough time for "BEGIN" to be produced
+    chan.send(asyncString("Slow", 500))
+    delay(100) // not enough time to produce slow
+    chan.send(asyncString("Replace", 100))
+    delay(500) // give it time before the last one
+    chan.send(asyncString("END", 500))
+    delay(1000) // give it time to process
+    chan.close() // close the channel ...
+    delay(500) // and wait some time to let it finish
+}