What do you guys think about this function? It allow to define concurrently running jobs:

class ParallelBuilder<T> {
    private val coroutines = ArrayList<Deferred<T>>()

    fun prun(context: CoroutineContext = CommonPool,
             start: CoroutineStart = CoroutineStart.DEFAULT,
             block: suspend CoroutineScope.() -> T) {
        coroutines.add(async(context, start, block))
    }

    suspend fun resume() = coroutines.map { it.await() }
}

inline suspend fun <R> parallel(crossinline builder: ParallelBuilder<R>.() -> Unit): List<R> {
    val scope = ParallelBuilder<R>()
    scope.builder()
    return scope.resume()
}

fun main(args: Array<String>) = runBlocking {
    val res = parallel<Int> {
        prun {
            println("1")
            delay(1000)
            println("1 done")
            1
        }

        prun {
            println("2")
            delay(1000)
            println("2 done")
            2
        }

        prun {
            println("3")
            delay(1000)
            println("3 done")
            3
        }
    }
}

output

1
2
3
2 done
3 done
1 done
Result: [1, 2, 3]

I suspect there are a number of such functions already written 🙂 However, I didn't find one in the standard lib.

@oleksandr.samsonov has left the channel

Hi, someone try to do some experiment with WebSocket + Coroutine? That’s my first experiment: https://gist.github.com/jacklt/2ada5872fd13f4ef7ae28f7fc19b4019

fun longJob(jobName: String) = launch(CommonPool) {
    val easyWS = client.easyWebSocket("<ws://echo.websocket.org>")
    println("[$jobName] Open: ${easyWS.response}")
    launch(context) {
        for (i in 1..3) {
            delay(1000) // every 1s
            val msg = "Hi, I am $jobName @ $i!"
            println("[$jobName] --> $msg")
            easyWS.webSocket.send(msg)
        }
        easyWS.webSocket.close(1000, "Bye! $jobName")
    }

    for (msg in easyWS.textChannel) {
        println("[$jobName] <-- $msg")
    }

    println("[$jobName] Finish!")
}

I use: OkHttp 3.8 for WebSocket, wrap in a

suspendCoroutine

to obtain

webSocket

, and send all received message in a

Channel<String>

(so I can loop all received message)

After that I wrap all the code for handle a single WebSocket in a

launch(CommonPool)

, so I can run multiple jobs at the same time. Is it right?

@Omar Miatello The client code conceptually looks right.

… but?

xD

It would be interesting to look at your

easyWebSocket

implementation (a link to github repo or gist would be the best way to share)

☝️

before the code 🙂

the seconds

launch

could use

context

instead of

CommonPool

Thanks @elizarov and @vaskir 🙂

override fun onMessage(webSocket: WebSocket, text: String) {
            launch(CommonPool) { easyWs!!.textChannel.send(text) }
    }

It can reorder messages. I would replace

launch

with

runBlocking

here.

@Omar Miatello An important observation here is that blocking a thread is a form of back-pressure in this case. Ideally, you should use something else for back-pressure, some explicit mechanism. However, if

OkHttp

library has no other way of explicit back-pressure signalling, then you should not be shy to block its thread to signal back-pressure.

Ok, I understand 🙂 And that (

suspendCoroutine

+

runBlocking

+

Channel

) is a good practice when you try to wrap a complex callback?

I read: https://github.com/Kotlin/kotlin-coroutines/blob/master/kotlin-coroutines-informal.md#wrapping-callbacks

@Omar Miatello It is not a simple callback. It is a multi-fire callback. There are no written guidelines on how to do it, but given the absence of any way to signal backpressure explicitly, this seems to a good solution.

is it planned to support passing suspend functions into HOFs? I mean:

suspend fun f() = 1
suspend fun h(f: suspend () -> Int) = ...
h(::f) // compilation error

I have to write

h { f() }

@vaskir Yes. It is planned. Here is the issue for your reference: https://youtrack.jetbrains.com/issue/KT-16908

@elizarov thanks

oh man, found some very concurrent heavy code I wrote about 3 years ago, entirely in java, entirely with Blocking Queues, and a dash of recursion, I just hit ctrl + alt + K, and now I'm wondering how far down this rabbit hole I should go (at 1:35AM pacific time)

@groostav I would not recommend going down this hole at 1:30am

Unless you want to stay up to watch the sun rise 🌇

Could I instantiate the coroutine directly, or should I rather instantiate the underlying implementation?

(from thread in #general about serialising and restarting coroutines from where they left off)

@restioson I’m not getting the difference. What do you mean by “instantiate”?

Create, call the constructor