Hi all - I'm trying to make sense of BroadcastChan...

# coroutines

[synhershko]

Hi all - I'm trying to make sense of BroadcastChannel and how to use it. Basically I have a flow of channel -> channel -> broadcast channel -> multiple observers / consumers -> channel (aka sink). I want to be able to get all the processing results in that last channel, and when all messages from it are consumed return. Basically the last line in my code is return@runBlocking tagsChannel.toList().
Problem is, I run my code and while I can see things being emitted from one channel to another, the code runs forever without any of the broadcastchannel subscribers receiving any message.
What am I missing here?

[Dominaezzz]

Code example?

[synhershko]

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.text.BreakIterator
import java.util.*
import kotlin.collections.ArrayList

class Test() {

    @ExperimentalCoroutinesApi
    fun annotate(text: String): List<String> {
        val tagsChannel = Channel<String>()
        val shinglesChannel = BroadcastChannel<String>(1000)
        val sentencesChannel = Channel<String>()

        return runBlocking {

            launch {
                shinglesChannel.consumeEach {
                    println("Hebrew: " + it)
                    tagsChannel.send("Foo")
                }
            }
            launch {
                shinglesChannel.consumeEach {t ->
                    println("English: " + t)
                    tagsChannel.send("Bar")
                }
            }

            launch {
                sentencesChannel.consumeEach { text ->
                    repeat(10) {
                        shinglesChannel.send("Foo bar")
                    }
                }
                shinglesChannel.close()
            }

            launch {
                val breakIterator = BreakIterator.getSentenceInstance(Locale("he"))
                breakIterator.setText(text)

                val start = breakIterator.first()
                var end = breakIterator.next()
                while (end != BreakIterator.DONE) {
                    val sentence = text.substring(start, end).trim { it <= ' ' }
                    sentencesChannel.send(sentence)
                    end = breakIterator.next()
                }
                sentencesChannel.close()
            }.join()

            val ret = ArrayList<String>()
            launch {
                tagsChannel.consumeEach {
                    println("test")
                    ret.add(it)
                }
                tagsChannel.close()
            }.join()
            return@runBlocking ret
        }
    }
}

@ExperimentalCoroutinesApi
fun main() {
    val annotaor = Test()
    val ret = annotaor.annotate("Test one two three")
}

[Dominaezzz]

I think you have to call openSubscription on the broadcast channel.

[synhershko]

consumeEach does it for you , at least that's what the internet says. I do get the messages, but this code hangs and never finishes and I'm not sure why?

[synhershko]

I do close the root channel so it should have propagated the close on all the others, no?

[Dominaezzz]

Oh I see the issue now.

[Dominaezzz]

Your final launch is waiting for the previous lines of code to execute.

[Dominaezzz]

But the previous code is waiting for the tagsChannel to be consumed.

[Dominaezzz]

Which the final launch is supposed to be doing.

[Dominaezzz]

i.e The code should be execute fine once you remove the first .join().

[Dominaezzz]

(Although, using launch {}.join() is rather suspicious code).

[synhershko]

It still hangs. I actually started adding the `.join()`s after trying to figure out why it hangs. It also hangs when all joins are removed.

[Dominaezzz]

Found the second issue.

[Dominaezzz]

tagsChannel.consumeEach {
    println("test")
    ret.add(it)
}
tagsChannel.close()

[Dominaezzz]

consumeEach doesn't return until `close`/`cancel`is called.

[synhershko]

close was called on the root channel sentencesChannel.close()

[Dominaezzz]

Yes but not on tagsChannel.

[synhershko]

ok I removed it

[synhershko]

val ret = ArrayList<String>()
            launch {
                tagsChannel.consumeEach {
                    println("test")
                    ret.add(it)
                }
            }
            return@runBlocking ret

[synhershko]

still hangs

[Dominaezzz]

No I meant, tagsChannel.close() is not being called.

[synhershko]

I had it before...

[synhershko]

val ret = ArrayList<String>()
            launch {
                tagsChannel.consumeEach {
                    println("test")
                    ret.add(it)
                }
                tagsChannel.close()
            }
            return@runBlocking ret

[synhershko]

still hangs

[Dominaezzz]

Yes because tagsChannel.consumeEach waits until tagsChannel.close() is called. And here you only call tagsChannel.close() after tagsChannel.consumeEach.

[Dominaezzz]

You've created a circular dependency.

[synhershko]

uhm because I don't have a way to know the emittance has stopped

[synhershko]

but if I call sentencesChannel.close() up stream , doesn't it propagate the close message downstream?

[Dominaezzz]

Yes but only to coroutines directly listening to sentencesChannel.

[Dominaezzz]

A quick and dirty workaround is to do another launch that will call join on the two `launch`es and call tagsChannel.close() after.

[Dominaezzz]

val hack1 = launch {
    shinglesChannel.consumeEach {
        println("Hebrew: $it")
        tagsChannel.send("Foo")
    }
}
val hack2 = launch {
    shinglesChannel.consumeEach { t ->
        println("English: " + t)
        tagsChannel.send("Bar")
    }
}
launch {
    tagsChannel.consume {
        hack1.join()
        hack2.join()
    }
}

[Dominaezzz]

The "correct"/nice solution would be to use Flow or some CoroutineScope passing, depending on what your actual code is doing.

[synhershko]

I'll try what you suggested

[synhershko]

The idea behind the code is mostly arranging what could have been a very messy Java code into a nicely ordered "stream processing" way of thinking. The code you see here is being run on a single method call, so the motivation is not performance by running lengthy code in coroutines, but mostly motivation is to keep code clean and extensible

[Dominaezzz]

Yeah, then Flow is probably the answer.

[synhershko]

The code you wrote also hangs

[Dominaezzz]

Oh. Doesn't hang when I run it.

[Dominaezzz]

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consume
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import java.text.BreakIterator
import java.util.*
import kotlin.collections.ArrayList

class Test() {

    @ExperimentalCoroutinesApi
    fun annotate(text: String): List<String> {
        val tagsChannel = Channel<String>()
        val shinglesChannel = BroadcastChannel<String>(1000)
        val sentencesChannel = Channel<String>()

        return runBlocking {
            val hack1 = launch {
                shinglesChannel.consumeEach {
                    println("Hebrew: $it")
                    tagsChannel.send("Foo")
                }
            }
            val hack2 = launch {
                shinglesChannel.consumeEach { t ->
                    println("English: " + t)
                    tagsChannel.send("Bar")
                }
            }
            launch {
                tagsChannel.consume {
                    hack1.join()
                    hack2.join()
                }
            }

            launch {
                sentencesChannel.consumeEach { text ->
                    repeat(10) {
                        shinglesChannel.send("Foo bar")
                    }
                }
                shinglesChannel.close()
            }

            launch {
                val breakIterator = BreakIterator.getSentenceInstance(Locale("he"))
                breakIterator.setText(text)

                val start = breakIterator.first()
                var end = breakIterator.next()
                while (end != BreakIterator.DONE) {
                    val sentence = text.substring(start, end).trim { it <= ' ' }
                    sentencesChannel.send(sentence)
                    end = breakIterator.next()
                }
                sentencesChannel.close()
            }.join()

            val ret = ArrayList<String>()
            launch {
                tagsChannel.consumeEach {
                    println("test")
                    ret.add(it)
                }
                tagsChannel.close()
            }.join()
            return@runBlocking ret
        }
    }
}

@ExperimentalCoroutinesApi
fun main() {
    val annotaor = Test()
    val ret = annotaor.annotate("Test one two three")
}

[synhershko]

Yeah your code works. One sec, looking.

[synhershko]

Oh it's because I removed the last launch block... thanks so much for the help!

[synhershko]

can you shortly explain what Flow is and how it's different from what we just wrote together?

[Dominaezzz]

Flow are basically lazy channels, that handle resource management for you. i.e `close`/`cancel`. They also have nice stream operators

[synhershko]

cool, thanks again!

[Dominaezzz]

Here are a few useful links.
https://proandroiddev.com/an-early-look-at-kotlin-coroutines-flow-62e46baa6eb0
https://medium.com/@elizarov/simple-design-of-kotlin-flow-4725e7398c4c
https://medium.com/@elizarov/kotlin-flows-and-coroutines-256260fb3bdb

[Dominaezzz]

No problem!