I have a flow producer that will emit elements with random period, I don't want to handle these elements once it emit, I'd rather gather them into a list, then handle them, and keep this process running.
For example, if I want to have 2s as a period, want I want to get for below flow is
val flow = flow{
for(i in 1..5){
emit(i)
delay(1100)
}
}
//#1 handle [1,2]
//#2 handle [3,4]
//#3 handle [5]
Below is what I tried but failed:
take(2), this will make the flow end after I get two elements, and there's no time to controlcollectToList().sample(2s), this is my customized function, adding elements into a list. But the issue is I need to clear the list in the collectToList() after sample(), otherwise the list will be appended forever, but if I do the clear, this may cause concurrency issues like "add a new element into the list just before I clear the list"Any better ideas?
The short answer is that there is no built-in function for this at the moment AFAIK.
I think what you're looking for is a time-based chunked operator, which has been discussed quite a bit, but still hasn't made it to the library. Not sure if it's in their plans for the near future.
Maybe you can take a look at the proposed implementation in this merge request and adapt it to your needs (you probably don't need as much flexibility, nor the extra chunking strategies). I'm sorry I'm afk so I can't do the adaptation myself.