Or a more general question on Channels: if I `clos...
# coroutinesr
reactormonk
04/23/2024, 5:25 PMOr a more general question on Channels: if I
close()Channelsend()👌 1
reactormonk
04/23/2024, 5:44 PMClosing a channel after this function has suspended does not cause this suspended send invocation to abort, because closing a channel is conceptually like sending a special "close token" over this channel. All elements sent over the channel are delivered in first-in first-out order. The sent element will be delivered to receivers before the close token.
c
Casey Brooks
04/23/2024, 5:44 PMYes. Per the docs, calling
channel.close()isClosedForSendisClosedForReceiver
reactormonk
04/23/2024, 5:46 PMTrying to debug this clusterfuck:
Copy code
object CardControlUtils {
suspend fun waitForReconnect(flow: Flow<CardWithCommands?>, timeout: Duration): MutableStateFlow<CardWithCommands?> {
val outgoing = MutableStateFlow<CardWithCommands?>(null)
val delayedReaderDisconnected = SupervisorJob()
CoroutineScope(coroutineContext).launch {
var oldChannel: Channel<CardCommand>? = null
var newChannel: Channel<CardCommand>? = null
fun reset() {
outgoing.value = null
oldChannel?.close()
oldChannel = null
newChannel?.close()
newChannel = null
}
flow.collect { card ->
if (card != null) {
fun setupChannels() {
oldChannel = card.commandChannel
newChannel = Channel<CardCommand>(5)
outgoing.value = card.copy(
commandChannel = newChannel!!
)
}
if (outgoing.value == null) {
setupChannels()
} else {
delayedReaderDisconnected.cancelChildren()
if (card.message == outgoing.value?.message) {
Timber.d { "Reader here again, all fine." }
} else {
newChannel?.send(CardCommand.TagLeftField)
reset()
setupChannels()
}
}
CoroutineScope(coroutineContext).launch {
println("Starting to consume card command channel: $oldChannel")
if (oldChannel != null) {
oldChannel?.consumeAsFlow()?.collect { msg ->
println("Message: $msg")
when (msg) {
CardCommand.ReaderDisconnected -> {
Timber.d { "Reader disconnected, ignoring for now..." }
launch(delayedReaderDisconnected) {
delay(timeout)
Timber.d { "Sending ReaderDisconnected" }
newChannel?.send(msg)
reset()
}
}
CardCommand.RotatedCard -> newChannel?.send(msg)
CardCommand.SkippedCard -> newChannel?.send(msg)
CardCommand.TagLeftField -> {
newChannel?.send(msg)
reset()
}
}
}
}
println("Done consuming.")
}
} else {
reset()
}
}
}
return outgoing
}
} Copy code
class CardControlUtilsTest {
@Test
fun testStandardForward() {
val scope = CoroutineScope(Dispatchers.Default)
val flow = Channel<CardWithCommands?>(5)
scope.launch {
flow.send(null)
val channel = Channel<CardCommand>(5)
flow.send(
CardWithCommands(
<http://CardMessage.AI|CardMessage.AI>,
channel,
MutableStateFlow("uid"),
scope,
)
)
println("launched channel $channel")
delay(500.milliseconds.toJavaDuration())
channel.send(CardCommand.TagLeftField)
println("sent")
channel.close()
flow.send(null)
}
runBlocking {
val gotFlow = waitForReconnect(flow.consumeAsFlow(), 1.seconds)
gotFlow.test {
assertEquals(null, awaitItem())
val card = awaitItem()
println("got item: $card")
assertEquals(<http://CardMessage.AI|CardMessage.AI>, card?.message)
val commandFlow = card?.commandChannel?.receiveAsFlow()
scope.launch {
commandFlow?.collect {
println(it.toString())
}
}
assertNotNull(commandFlow)
commandFlow!!.test {
assertEquals(CardCommand.TagLeftField, awaitItem())
println("got left field")
awaitComplete()
}
println("got commandflow completed")
assertEquals(null, awaitItem())
}
}
}
} Copy code
launched channel capacity=5,data=[kotlinx.coroutines.channels.BufferedChannel$BufferedChannelIterator@3997e019]
got item: CardWithCommands(message=AI(),cardJob=SupervisorJobImpl{Active},uid=uid)
Starting to consume card command channel: capacity=5,data=[kotlinx.coroutines.channels.BufferedChannel$BufferedChannelIterator@3997e019]
sent
Done consuming.
Expected item but found Complete Copy code
Message: ...reactormonk
04/23/2024, 7:54 PMTested some more, the first message doesn't arrive, but if I send a second one, it goes through... WTF?
c
Casey Brooks
04/23/2024, 8:04 PMThe example is pretty long and convoluted, it’s hard to know what exactly you’re trying to do. What’s the actual use-case here?
Casey Brooks
04/23/2024, 8:07 PMBut one thing I notice: you’re launching a job within
flow.collect { card ->r
reactormonk
04/23/2024, 8:17 PMI'm not really sure how to do nested flows exactly, the current setup is
Copy code
┌─────────────────────────────────────────────────────────────────────────┐
│ │
│ ┌────────────────────────────────┐ ┌───────────────────────────────┐ │
│ │Card │ │Card │ │
│ │ │ │ │ │
│ │ ┌───────────┐ ┌───────────┐ │ │ ┌───────────┐ ┌─────────────┐ │ │
│ │ │Command │ │Command │ │ │ │Command │ │Command │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │►│ │ │ │ │ │►│ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │ │ │ ├─►│ │ │ │ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ └───────────┘ └───────────┘ │ │ └───────────┘ └─────────────┘ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ └────────────────────────────────┘ └───────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘reactormonk
04/23/2024, 8:18 PMSo I have a nested channel structure, where the
Cardreactormonk
04/23/2024, 8:18 PM... now I'm writing a function which merges some of the
Cardc
Casey Brooks
04/23/2024, 8:22 PMSo you have a
Cardcommandsr
reactormonk
04/23/2024, 8:24 PMNo, I want to merge two cards, like so:
Copy code
│
│
┌────────────────────────────────────┴────────────────────────────────────┐
│ ▼ │
│ ┌────────────────────────────────┬──┬───────────────────────────────┐ │
│ │Card │ │Card │ │
│ │ │ │ │ │
│ │ ┌───────────┐ ┌───────────┐ │ │ ┌───────────┐ ┌─────────────┐ │ │
│ │ │Command │ │Command │ │ │ │Command │ │Command │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │►│ │ │ │ │ │►│ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ │ │ │ │ ├─►│ │ │ │ │ │ │
│ │ │ │ │ │ │ │ │ │ │ │ │ │
│ │ └───────────┘ └───────────┘ │ │ └───────────┘ └─────────────┘ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ │ │ │ │
│ └────────────────────────────────┴──┴───────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘reactormonk
04/23/2024, 8:25 PMSo I'm sticking together the two command channels one after another
reactormonk
04/23/2024, 8:27 PMNope, removed the fork/launch, same result still, first message is getting swallowed 🤔
reactormonk
04/23/2024, 8:27 PM... conceptually, the forking doesn't make sense either, the channel of the first card will be closed before the second card arrives
c
Casey Brooks
04/23/2024, 8:30 PMIgnoring the channels for a moment, because there might be standard Flow operators that do what you want. And I also don’t really understand the diagrams, so can you please explain the use-case just in terms of high-level data flow? Is it that you have a card reader which is issuing commands from multiple cards in parallel, but you want to process all commands from one card before starting to process the next?
Casey Brooks
04/23/2024, 8:32 PMIn general, Flows process data sequentially, so doing all the work of launching other coroutines inside
flow.collect { }r
reactormonk
04/23/2024, 8:34 PMThe card reader is issuing commands in sequence, like card on field. Then it sends a "Card" message along. Each card also contains a channel for commands, e.g. "tag left field", aka the card left the card reader, or "reader disconnected". Now I wanna add a delay on the "reader disconnected" logic, so it doesn't get send forward immediately, instead it should wait a few seconds to forward it, and if the reader reappears, act as if nothing happened.
reactormonk
04/23/2024, 8:35 PMThe channels are the same, at least according to the object ids
c
Casey Brooks
04/23/2024, 9:02 PMI think you need to change the logic so it’s managing the cancellation of Jobs, not Channels. Perhaps something like this will better handle the kind of logic you need:
Copy code
var cardDisconnectedJob: Job? = null
card
.commandChannel
.filterNotNull()
.onEach { msg: CardCommand ->
// cancel the job that's waiting for the timeout, if it exists
cardDisconnectedJob?.cancel()
when (msg) {
CardCommand.ReaderDisconnected -> {
cardDisconnectedJob = launch(delayedReaderDisconnected) {
delay(timeout)
// we exceeded the timeout, continue cancelling the whole card processing
}
}
else -> {
outgoing.send(msg)
}
}
}
.launchIn(scope)r
reactormonk
04/23/2024, 9:05 PMSplit it up a bit, now it should be easier to understand:
Copy code
suspend fun waitForReconnect(flow: Flow<CardWithCommands?>, delay: Duration): MutableStateFlow<CardWithCommands?> {
val outgoing = MutableStateFlow<CardWithCommands?>(null)
val delayedReaderDisconnected = SupervisorJob()
val topScope = CoroutineScope(coroutineContext)
topScope.launch {
var newChannel: Channel<CardCommand>? = null
fun reset() {
outgoing.value = null
newChannel?.close()
newChannel = null
}
suspend fun setupConsume(channel: Channel<CardCommand>, to: Channel<CardCommand>) {
channel.consumeEach { msg ->
println("Message: $msg")
when (msg) {
CardCommand.ReaderDisconnected -> {
Timber.d { "Reader disconnected, ignoring for now..." }
launch(delayedReaderDisconnected) {
delay(delay)
Timber.d { "Sending ReaderDisconnected" }
to.send(msg)
reset()
}
}
CardCommand.RotatedCard -> to.send(msg)
CardCommand.SkippedCard -> to.send(msg)
CardCommand.TagLeftField -> {
to.send(msg)
reset()
}
}
}
}
flow.collect { card ->
delayedReaderDisconnected.cancelChildren()
if (card != null) {
suspend fun setupChannels() {
newChannel = Channel<CardCommand>(5)
outgoing.value = card.copy(
commandChannel = newChannel!!
)
setupConsume(card.commandChannel, newChannel!!)
}
if (outgoing.value == null) {
setupChannels()
} else {
if (card.message == outgoing.value?.message) {
Timber.d { "Reader here again, all fine." }
setupConsume(card.commandChannel, newChannel!!)
} else {
newChannel?.send(CardCommand.TagLeftField)
reset()
setupChannels()
}
}
} else {
reset()
}
}
}
return outgoing
}3 Views