https://kotlinlang.org logo
Join Slack
Powered by
I'm trying to do something like the following: ```...
# coroutines
o
I'm trying to do something like the following:
Copy code
fun messagesFromConversation(conversationId: String): Flow<List<Message>> = channelFlow {
    val messageHistory = getMessageHistory()
    offer(messageHistory)
    val accumulatedMessages = messageHistory.toMutableList()
    val sentMessages: Flow<Message> = sentMessagesFromConversation(conversationId)
    sentMessages.collect { sentMessage ->
        acumulatedMessages.add(0, sentMessage)
        offer(accumulatedMessages)
    }
}
but Im getting a weird issue that on every collect of the sentmessages, the accumulatedmessages are not represeting the full list. I think it's some sort of mutable state sharing concurrency issue, but I have no idea how to better approach this issue. Do I need to some sort of ConcurrentMutableList?
m
offer
may return 
false
if an element wasn’t accepted due to the channel being busy. Did you mean to use 
send
?
Interestingly I’ve written something very similar a few days ago. Also conversations and messages. But I exclusively use immutable list and create copies to ensure that I never have shared mutable state.
If something subscribes to your flow holds a long-lived reference to the list, it may get surprised that the list content has suddenly changed.
send(accumulatedMessages.toList())
+ the other 
offer
-> 
send
o
offer is not returning false, also, looks like the issue isnt with the channelflow emitting, but updating the mutable messages list. For example, given the message history is two messages, when the sent message flow emits another message, I breakpoint and I get the following state from the debugger: accumulatedmessages.size = 3. But when after lets say 10 seconds the next sent message gets emitted, when I breakpoint it, the accumulatedmessage.size is back to 2, as of the previous sentmessage didnt happen. Now the previous send chat message appears after some emits after, but its really unstable.
m
Are you sure it’s still the same flow? I.e. has 
this
changed between breakpoints?
o
@Marc Knaup wow thats really great you also did this, helps alot. I have to tell you, I initially also tried this with immutability, but then you have the following immutability right: 
var accumulatedMessages
m
yes, but that immutability isn’t shared
o
I got the same issue, I was even using ImmutableList and PersistentList from this library:
m
Some 
println
may also help here. Maybe multiple flows are created instead of one. There should be no need for special immutable lists if you only send copies around 🙂
Here’s a random example of my code. Note the 
listOf(message) + snapshot.queue
to create a new immutable list instead of mutating any existing one.
o
Thanks for the help because it drives me crazy. So do you suggest something like this:?
Copy code
var accumulatedMessages: List<Message> = messageHistory
sentMessages.collect { sentMessage ->
    accumulatedMessages = accumulatedMessages + sentMessage
    send(accumulatedMessages)
}
I have to to also tell you, this is not the exact full code. In the same channelFlow, Im also collecting incomingMessages, also a flow. So they are both collecting in their own coroutine. Is it still a good idea then to use send isntead of offer()?
m
If you have two 
collect
in your flow and both are never-ending streams then the second 
collect
should never be called because the first will run forever?
o
Do you mind if I post the whole code here?
m
A Gist may be easier to read
o
I never used Gist, so I put it in a pastebin instead: https://pastebin.com/M16fisXQ
As an answer to your question, you can see all of the collects get launched in the scope of the channelFLow
So they all collect in their own coroutine
m
In any case I guess all these 
offer
should be 
send
. Any reason you’ve used 
offer
?
o
I saw offer being used alot in channelFLow examples, and I thought if send is suspending, it might block the other collecting flows, but I just realised that they are all running in their own coroutine ofcourse, so it wont be a problem
However, I remember also trying send() and it not making a difference. The emitting part is also not the issue, its that the accumulatedlist doesnt seem to get persisted
m
Well 
offer
is okay if you are okay with some elements never being sent in certain situations. But I guess that isn’t okay here 🙂
o
I might as well not use offer and send() at all, and just see how the sent messages dont get persisteted directly to the list
Yes, I will change it to send()
But now posting this pastebin has me realised with the following: all of their collects are being run in their own coroutine, and im not using awaitClose at the end, meaning the channelflow probably hit the end. Do you think that might trouble?
m
Oh, there’s a 
conflate
at the end. That may make it irrelevant whether to use 
offer
or 
send
because with 
conflate
the buffer is never full 🤔
o
Right, I thought conflate() was a good thing here since all that iss important is the latest state of the messages
Anyhow the bug was also there before using conflate()
m
Yeah. I just usually leave the decision up to the consumer. Anyway, not the cause
o
yes, its either conflate() for emitter, or collectLatest for receiver right
Anyway, would this be better you think:?
Copy code
accumulatedMessages = accumulatedMessages + sentMessage
send(accumulatedMessages)
m
Probably not. I guess the issue lies somewhere else
o
Perhaps I need to put awaitClose at the end of the channelflow to keep it running, even tho I have no idea how that can correlate with this persisting issue
m
Since you don’t even use 
awaitClose
you may as well just use 
flow { … }
? It should stay open automatically though.
Can you put a 
println
at the start of the flow? Just to make sure that we’re only observing a single flow.
o
When I was using flow {} I got an illegalstateexception sometihng with flow invariant and emitting from different coroutine, and that I had to use channelFlow instead
Just for your information: incomingmessages gets also generated by a channelflow
m
So you did a workaround instead of understanding the problem. Maybe it tried to tell you where the concurrency issue lies 😄
o
I will try it again using flow {} and share the exception
Also, wait, if I use flow, I can't collect in parallel
}.launchIn(this) wouldn't work since this wouldnt be a CoroutineScope then
m
I think you have to wrap the entire flow content in a 
coroutineScope { … }
o
Allright, will do
Also thought of that, but I remember that exception occuring from there
But lets give it another try
m
yeah that may make sense. not sure if that’s compatible with flow
I’ve just tested it. 
flow {}
and 
coroutineScope {}
work just fine together.
o
I think it depends on what you are doing inside of the coroutineScope
m
Yeah, 
flow {}
and 
launch {}
don’t seem to work together.
o
Copy code
W/System.err: java.lang.IllegalStateException: Flow invariant is violated:
    		Emission from another coroutine is detected.
    		Child of StandaloneCoroutine{Active}@f625a8a, expected child of StandaloneCoroutine{Active}@532c8fb.
    		FlowCollector is not thread-safe and concurrent emissions are prohibited.
    		To mitigate this restriction please use 'channelFlow' builder instead of 'flow'
W/System.err: java.lang.IllegalStateException: Flow invariant is violated:
    		Emission from another coroutine is detected.
    		Child of StandaloneCoroutine{Active}@f625a8a, expected child of StandaloneCoroutine{Active}@532c8fb.
    		FlowCollector is not thread-safe and concurrent emissions are prohibited.
    		To mitigate this restriction please use 'channelFlow' builder instead of 'flow'
Concurrent emissions are prohibited it says
m
Well there is a way around. But I’m not sure if it’s the proper way 😄
You can save the context outside: 
val flowContext = coroutineContext
and then use it inside:
Copy code
withContext(flowContext) {
   emit(…)
}
o
I dont know if thats recommended: since at the end of the exception it also says to use:
Copy code
FlowCollector is not thread-safe and concurrent emissions are prohibited.
    		To mitigate this restriction please use 'channelFlow' builder instead of 'flow'
m
It says that the approach you’re using isn’t thread-safe. If you use 
channelFlow
you’ve silenced the warning but you code must also be thread-safe then. But it isn’t due to mutable state.
o
Maybe I have some useful information to share
the 
sentChatMessages
flow is coming from here, outside of the class as a private field
one moment
Copy code
private val _sentOneOnOneChatMessages: Channel<OneOnOneChatMessage<ChatMessageContent>> = Channel()
    private val sentChatMessages: Flow<OneOnOneChatMessage<ChatMessageContent>> =
        _sentOneOnOneChatMessages.receiveAsFlow()
And here is where that channel gets its messages from:
message has been deleted
Only relevant line of the last picture is the last one, the 
also {}
so the sentChatMessagesFlow is a hot flow
Perhaps using BroadcastChannel is a better fit there than Channel
m
I’d try this in all `onEach`:
Copy code
withContext(flowContext) {
   // on each code (including mutable state access)
}
to have all code within the Flow run in the same context
The launch causes a thread switch
otoh that should’t be an issue
ugh, it’s complicated 😄
(Have you checked that the flow isn’t collected multiple times?)
o
Yes, the breakpoint only gets hit once at the start
m
good
o
I don't get why the launch would cause a thread switch from collecting the sent messages
launching a coroutine simply inherits the context right? So that means that the coroutine is already running in the flow context
m
It gets its own context because launching creates a new 
Job
.
I’m still puzzled because of your error description.
o
Copy code
Creates an instance of the cold Flow with elements that are sent to a SendChannel provided to the builder's block of code via ProducerScope. It allows elements to be produced by code that is running in a different context or concurrently. The resulting flow is cold, which means that block is called every time a terminal operator is applied to the resulting flow.
m
You say the message list becomes shorter again, which is impossible if it’s still the same flow. There is no way the list can get shorter except for starting a new flow?!
o
I'm not trusting the debugger
I think im just gonna bust out printlns everywhere
m
hehe, good old prints
still, put one at the start please :)
o
yep, going to do
Will come back with the result in ~10 mins
Thank you for helping, its hard to find some real word usecases with Flow
m
Takes a while to think in Flows, yes. And I keep hitting limits 😄
o
PENTAGON: channelFlow start PENTAGON: channelFlow start PENTAGON: channelFlow end PENTAGON: channelFlow end PENTAGON: sentChatMessages.onEach start PENTAGON: sentChatMessages.onEach currentChatMessages.size=21 PENTAGON: adding sentChatMessage, content=Text(value=2) PENTAGON: sending: currentChatMessages.size=22 PENTAGON: sentChatMessages.onEach end PENTAGON: sentChatMessages.onEach start PENTAGON: sentChatMessages.onEach currentChatMessages.size=21 PENTAGON: adding sentChatMessage, content=Text(value=3) PENTAGON: sending: currentChatMessages.size=22 PENTAGON: sentChatMessages.onEach end PENTAGON: sentChatMessages.onEach start PENTAGON: sentChatMessages.onEach currentChatMessages.size=22 PENTAGON: adding sentChatMessage, content=Text(value=testtest) PENTAGON: sending: currentChatMessages.size=23 PENTAGON: sentChatMessages.onEach end
m
How can it end when it’s still emitting? Where exactly did you print start & end?
o
You see, when sentmessage with content 2 is being added (2 is the actual message body), the message list size was together 22. now when the next sent message came in, the list became 21 again
m
But you have two flows running. How do you know what log message is what flow instance?
o
I updated the output log again, check how on the 3rd message, the message size is back to 22
you mean the two channelFlows at the start?
m
yes
o
Yes I dont know where that one comes from, im only using this flow once
Maybe thats the issue
m
yeah, likely 🙂
o
But have no idea how
Im going to put out some more printlns from the caller, will report back
Allright, the code you saw till know was from the repo. The viewmodel calls an usecase, which calls the repo. Just found out the usecases flow also gets called twice
PENTAGON: usecase channelflow start PENTAGON: usecase channelflow start PENTAGON: channelFlow start PENTAGON: channelFlow start PENTAGON: channelFlow end PENTAGON: channelFlow end
Looks like something is going wrong with the viewmodel
Yep omg
Fixed it
m
😄
o
Looks like the issue was a small innocent but righteous evil wrong use of 
asLiveData()
on Flow
I forgot that asLiveData() starts a collect
It was being collected twice from the viewmodel
m
I try to avoid 
LiveData
if possible. Not sure though if Kotlin covers all use cases yet 🤔
o
I will show you what I was trying to do
I commented the only relevant lines with 
// RELEVANT
You can see the line on 115 was the evil one
I guess in order to do what I want to do, is using the share() operator thats still in draft
m
I have merged the entire conversation state into a 
ConversationContentSnapshot
. It includes messages, send queue, loading states, errors, etc. But my UI is still imperative so I just collect all snapshots and save the latest one in a property in the presenter.
o
Yes, I guess I need to use that too.
m
Status quo, in case it helps with thinking :)
o
message has been deleted
Indeed, I have something like this
6 Views
Open in Slack
PreviousNext
Powered by