Scaling Up to Kotlin Coroutines

series_of_items.each { item -> print(item) }

const server = http.createServer((req, res) => {
  res.statusCode = 200;
  res.setHeader('Content-Type', 'text/plain');
  res.end('Hello World\n');
});

server.listen(port, hostname, () => {
  console.log(`Server running at http://${hostname}:${port}/`);
});

console.log('Server starting');

                            Optional.of("something")
                                .map(item -> item + " else")
                                .ifPresent(System.out::println);

            promiseToEventuallyHaveValue
               .then((retrievedValue) => { doSomething(retrievedValue) })
               .catch((err) => { handleError(err) })

/**
 * A simple implementation of {@link Promise}.
 *
 * Does not provide any concurrency support, but shows how the other aspects
 * of how a Promise works.
 */
class ConceptualPromise {

    /**
     * @param {executorCallback} executor - handler for when this receives a value
     */
    constructor(executor) {
        // no concurrency, but support the Promise "flow"
        executor(
            (resolveValue) => {
                this.resolvedValue = resolveValue
            },
            (rejectedValue) => {
                this.rejectedValue = rejectedValue
            }
        );
    }

    /**
     * If this contains a value, pass it to the given handler.
     *
     * @param {thenCallback} callback - handler for the passed value
     * @returns {ConceptualPromise}
     */
    then(callback) {
        if (this.rejectedValue !== undefined) {
            return ConceptualPromise.rejected(this.rejectedValue)
        } else {
            return new ConceptualPromise((resolve, reject) => {
                try {
                    resolve(callback(this.resolvedValue))
                } catch (e) {
                    reject(e)
                }
            })
        }
    }

    /**
     * If this contains a rejection, invokes the provided callback.
     *
     * @param {catchCallback} callback - error handler
     * @returns {ConceptualPromise}
     */
    catch(callback) {
        if (this.rejectedValue !== undefined) {
            const reason = callback(this.rejectedValue) || this.rejectedValue;
            return ConceptualPromise.rejected(reason);
        } else {
            return ConceptualPromise.resolved(this.resolvedValue);
        }
    }

    // *********************************
    //
    // STATIC METHODS
    //
    // *********************************

    /**
     * Create a ConceptualPromise that contains the given value.
     *
     * @param {*} resolvedValue - the value to resolve to
     * @returns {ConceptualPromise}
     */
    static resolved(resolvedValue) {
        return (resolvedValue instanceof ConceptualPromise) ?
            resolvedValue :
            new ConceptualPromise((resolve, reject) => {
                resolve(resolvedValue)
            });
    }

    /**
     * Create a ConceptualPromise that is in a rejected state.
     *
     * @param {*} reason - the reason the for rejection
     * @returns {ConceptualPromise}
     */
    static rejected(reason) {
        return (reason instanceof ConceptualPromise) ?
            reason :
            new ConceptualPromise((resolve, reject) => {
                reject(reason)
            });
    }
}

/**
 * Adapter function to translate the Node callback API for creating a Server into a Promise.
 *
 * @returns {Promise<Server>}
 */
function createServerPromise() {
    return new Promise((resolve, reject) => {
        const server = http.createServer((req, res) => {
            res.statusCode = 200;
            res.setHeader('Content-Type', 'text/plain');
            res.end('Hello World\n');
        });
        resolve(server);
    });
}

/**
 * Adapter function to translate the Node callback API for create a connection listener for the Server into a Promise.
 *
 * @param {Server} server - the Server instance to create the connection listener for
 * @param {number} port - the port to create the connection listener on
 * @param {string} hostname - the hostname for the connection listener
 * @returns {Promise<{server, port, hostname}>}
 */
function createServerListenerPromise(server, port, hostname) {
    return new Promise((resolve, reject) => {
        server.listen(port, hostname, () => {
            resolve({server, port, hostname})
        })
    })
}

/**
 * Create and start an HTTP server on port 8089 using Promises.
 *
 * @returns {Promise<void>}
 */
function startPromiseServer() {
    return createServerPromise()
        .then((server) => {
            return createServerListenerPromise(server, 8089, 'localhost')
        })
        .then(({port, hostname}) => {
            console.log(`PromiseServer running at http://${hostname}:${port}/`);
        });
}

/**
 * Create and start an HTTP server on port 8090 using async/await.
 *
 * @returns {Promise<void>}
 */
async function startAsyncAwaitServer() {
    const server = await createServerPromise();
    const {port, hostname} = await createServerListenerPromise(server, 8090, 'localhost');
    console.log(`AsyncAwaitServer running at http://${hostname}:${port}/`);
}

/**
 * Create and start an HTTP server on port 8090 using async/await.
 *
 * @returns {Promise<void>}
 */
async function startAsyncAwaitServer() {
    const server = await createServerPromise();
    const {port, hostname} = await createServerListenerPromise(server, 8090, 'localhost');
    console.log(`AsyncAwaitServer running at http://${hostname}:${port}/`);
}

/**
 * Create and start an HTTP server on port 8090 using async/await.
 *
 * @returns {Promise<void>}
 */
async function startAsyncAwaitServer() {
    const server = await createServerPromise();
    const {port, hostname} = await createServerListenerPromise(server, 8090, 'localhost');
    console.log(`AsyncAwaitServer running at http://${hostname}:${port}/`);
}

                        ["a", "functional", "world"].map { it.length }
                          ==>
                        [1, 10, 5]

                    [[], ["a"], [], ["functional", "world"], []].flatten()
                      ==>
                    ["a", "functional", "world"]

/**
 * Applies the transformer to every element in the collection, then flattens the result.
 */
fun Collection<T>.flatMap(transformer: (T) -> Collection<R>): Collection<R>

Stream.of(2, 3, 4, 7, 9)
    .filter(i -> i % 2 == 0)
    .map(i -> i * 3)
    .findFirst()
    .ifPresent(i -> System.out.println("First filtered and mapped number: " + i));

// First filtered and mapped number: 6

fun something(aParam: String): String {
    return aParam + " is nice"
}

// exactly the same as

fun something(aParam: String) = aParam + " is nice"

fun String.makeNice() = this + " is nice"
"Music".makeNice()

// compiles to the same as

fun makeNice(str: String) = str + " is nice"
makeNice("Music")

fun processLauncher(max: Int, launcher: (Int) -> Unit) {
    val start = Instant.now()
    println("START [${Thread.currentThread().name}] - ${Thread.activeCount()} active")

    launcher(max)

    val end = Instant.now()
    println("END Duration: " + Duration.between(start, end).toMillis())
}

fun thread_scaling(max: Int) {
    val threads = mutableListOf<Thread>()
    for (i in 0..max) {
        val thread = Thread {
            Thread.sleep(10_000L) //simulate waiting 10s for a remote call
            if (i % 1_000 == 0) {
                println("[${Thread.currentThread().name}] - ${Thread.activeCount()} active")
            }
        }
        threads.add(thread)
        thread.start()
    }

    // make sure every thread has finished before returning
    threads.forEach { thread -> thread.join() }
}

fun main() {
    processLauncher(5_000) { thread_scaling(it) }
}

fun thread_scaling(max: Int) {
    val threads = mutableListOf<Thread>()
    for (i in 0..max) {
        val thread = Thread {
            Thread.sleep(10_000L) //simulate waiting 10s for a remote call
            if (i % 1_000 == 0) {
                println("[${Thread.currentThread().name}] - ${Thread.activeCount()} active")
            }
        }
        threads.add(thread)
        thread.start()
    }

    // make sure every thread has finished before returning
    threads.forEach { thread -> thread.join() }
}

fun main() {
    processLauncher(6_000) { thread_scaling(it) }
}

fun coroutine_main_scaling(max: Int) {
    runBlocking {
        for (i in 0..max) {
            launch {
                delay(10_000L) //simulate waiting 10s for a remote call
                if (i % 1_000 == 0) {
                    println("[${Thread.currentThread().name}] - ${Thread.activeCount()} active")
                }
            }
        }
    }
}

fun main() {
    processLauncher(5_000) { coroutine_main_scaling(it) }
}

fun coroutine_main_scaling(max: Int) {
    runBlocking {
        for (i in 0..max) {
            launch {
                delay(10_000L) //simulate waiting 10s for a remote call
                if (i % 1_000 == 0) {
                    println("[${Thread.currentThread().name}] - ${Thread.activeCount()} active")
                }
            }
        }
    }
}

fun main() {
    processLauncher(5_000) { coroutine_main_scaling(it) }
}

fun coroutine_main_scaling(max: Int) {
    runBlocking {
        for (i in 0..max) {
            launch {
                delay(10_000L) //simulate waiting 10s for a remote call
                if (i % 1_000_000 == 0) {
                    println("[${Thread.currentThread().name}] - ${Thread.activeCount()} active")
                }
            }
        }
    }
}

fun main() {
    processLauncher(5_000_000) { coroutine_main_scaling(it) }
}

fun coroutine_IO_scaling(max: Int) {
    runBlocking {
        for (i in 0..max) {
            launch(Dispatchers.IO) {
                delay(10_000L) //simulate waiting 10s for a remote call
                if (i % 1_000 == 0) {
                    println("[${Thread.currentThread().name}] - ${Thread.activeCount()} active")
                }
            }
        }
    }
}

fun main() {
    processLauncher(5_000) { coroutine_IO_scaling(it) }
}

                       runBlocking { /* "this" is a new CoroutineScope */ }

public fun <T> runBlocking(context: CoroutineContext = EmptyCoroutineContext,
                           block: suspend CoroutineScope.() -> T): T {
    val currentThread = Thread.currentThread()
    // ...

                       launch { /* "this" is a new CoroutineScope */ }

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job {

                                     delay(5_000)

suspend fun delay(timeMillis: Long) {

// classic blocking
fun sendEmail(emailArgs: EmailArgs): EmailResult

// Node-ish callback; quickly turns into callback hell
fun sendEmail(emailArgs: EmailArgs, callback: (Throwable?, EmailResult?) -> Unit)

// Future/Promise; good convention is to use the *Async to remind caller to do the `await`
fun sendEmailAsync(emailArgs: EmailArgs): Deferred<EmailResult>

// Coroutine
suspend fun sendEmail(emailArgs: EmailArgs): EmailResult

fun largerBusinessProcessAsync(): Deferred<Unit> = GlobalScope.async {
    // a lot of code here, then somewhere inside
    val emailResult = sendEmailAsync(emailArgs).await()
    // something else goes on after that
}

suspend fun largerBusinessProcess() {
    // a lot of code here, then somewhere inside
    val emailResult = sendEmail(emailArgs)
    // something else goes on after that
}

// simulate making 100 remote calls that take 1 sec each; total time is just over 1 sec
suspend fun spreadSum(): Int = coroutineScope {
    val deferredCalls: List<Deferred<Int>> = (1..100).map {
        async {
            delay(1000) // remote service call
            2
        }
    }
    deferredCalls
        .map { deferred -> deferred.await() }
        .sum()
}

interface SendChannel<T> {
    suspend fun send(value: T)
    fun close(cause: Throwable? = null): Boolean
    // ...
}

interface ReceiveChannel<T> {
    suspend fun receive(): T
    operator fun iterator(): ChannelIterator<E>
    fun cancel(cause: CancellationException? = null)
    // ...
}

interface Channel<E> : SendChannel<E>, ReceiveChannel<E>

fun CoroutineScope.fibonacci(): ReceiveChannel<Int> = produce {
    var x = 0
    var y = 1
    while (true) {
        send(x)
        val next = x + y
        x = y
        y = next
    }
}

fun main() = runBlocking {
    for (i in fibonacci().take(10)) {
        println("fib: $i")
    }
}

suspend fun fibonacci(n: Int, channel: SendChannel<Int>) {
    var x = 0
    var y = 1
    for (i in 0 until n) {
        channel.send(x)
        val next = x + y
        x = y
        y = next
    }
    channel.close()
}

fun main() = runBlocking {
    val fsc = Channel<Int>() // akin to a Go Channel
    launch { fibonacci(10, fsc) } // akin to a Go Routine
    for (i in fsc) {
        println("fib: $i")
    }
}

suspend fun fibonacci(c: SendChannel<Int>, quit: ReceiveChannel<Int>) {
    var x = 0; var y = 1
    whileSelect {
        c.onSend(x) {
            val next = x + y
            x = y; y = next
            true // continue while loop
        }
        quit.onReceive {
            println("quit")
            false // break while loop
        }
    }
}
fun main() = runBlocking {
    val sfsc = Channel<Int>()
    val quitChannel = Channel<Int>()
    launch {
        for (i in 0 until 10) {
            println("fib: ${sfsc.receive()}")
        }
        quitChannel.send(0)
    }
    fibonacci(sfsc, quitChannel)
}

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    for (i in 1..9)
        launch(Dispatchers.Default + CoroutineName("launch$i")) {
            for (str in channel) {
                println("${Thread.currentThread().name} - $str")
            }
        }

    for (letter in 'a'..'z') {
        channel.send(letter.toString())
    }
    channel.close()
}

sealed class CounterMsg
object IncCounter : CounterMsg() // one-way message to increment counter
class GetCounter(val response: CompletableDeferred<Int>) : CounterMsg() // a request with reply

fun main() = runBlocking<Unit> {
    val counterChannel: SendChannel<CounterMsg> = actor {
        var counter = 0 // private actor state
        for (msg in this.channel) { // iterate over incoming messages
            when (msg) {
                is IncCounter -> counter++
                is GetCounter -> msg.response.complete(counter)
            }
        }
    }

    for (i in 1..100) {
        counterChannel.send(IncCounter)
    }

    val response = CompletableDeferred<Int>()
    counterChannel.send(GetCounter(response))
    println("Counter = ${response.await()}")
    counterChannel.close() // shutdown the actor
}

fun main() = runBlocking {
    val mainContext = coroutineContext + CoroutineName("main-fun")

    val baseFlow = flow {
        for (i in 1..9) {
            emit(i)
            delay(200)
        }
    }

    baseFlow
        .sample(500) // every 500ms open the sluice for another item
        .map { item -> "$item - [${Thread.currentThread().name}]" }
        .flowOn(Dispatchers.Default + CoroutineName("proc-int-flow"))
        .collect {
            withContext(mainContext) {
                println("printing on [${Thread.currentThread().name}] -> $it")
            }
        }
}