Multiple, parallel Channel consumers

[grodzickir]

Hello,
I have a problem and an enhancement proposition.
I'm trying to use a Channel to fetch data from web and simultaneously insert it into db in parallel.
Fetching data is much faster than inserting, as I have to check if the data changed etc.

Here I have a problem how to properly implement a single producer and multiple parallel consumers of a single Channel.
Is there any better way than writing:

    val jobList: ArrayList<Job> = ArrayList()
    (1..20).forEach {
        jobList += launch(coroutineContext) {
            val x = myChannel.receiveOrNull()
            if(x != null) {
                //process
            }
        }
    }
    jobList.forEach {
        it.join()
    }

?

Assuming that I launched it in another coroutine with newFixedThreadPoolContext(20, "name") I would get a nice, parallel processing of the Channel.

Using this approach I can't use functions like map or filter without some weird boxing of one value into a collection. Thus it would be really appreciated to be able to do sth like :

    channel.parallel(coroutineContext/newFixedThreadPoolContext())
                .map{ ... }
                .filter{....}
                .consumeEach{...}

[elizarov]

This seems to be a duplicate of #172. Let's continue discussion there.