What would be the proper way to have a continuous ...

# coroutines

[Tolriq]

What would be the proper way to have a continuous source of elements that one could take elements from without closing? Like a pool of 10 items that require heavy computation, that is only refilled by the correct amount each time one is removed.

[gildor]

chunked operator for Flow (like existing one, but for List) would be perfect for this, but it doesn't exist yet, you can check issue tracker of Coroutines, there is some ad-hoc implementations as I remember

[gildor]

There is even a PR:
https://github.com/Kotlin/kotlinx.coroutines/pull/1558

[gildor]

Or maybe you want to get up to 10, all the time, not chunks, so you could just limit it using flatMap concurrency param

[Tolriq]

Yes up to X and usually will be 1 unless user want to see what will come next and then the X are consumed. Concept would be I pass a function to a consumer and as the consumer consume the items, new items are generated up to a limit to avoid pre generating all possible items at once.

[Tolriq]

So i need to be able to collect from different places even if never at the same time

[gildor]

Yeah, you can use flatMap* operator wth concurrency param

[gildor]

this concurrency limits amount of simultaneously running flows

[gildor]

so you processing could be represented as flow (even with 1 element), so flatMap will help limit amount of simultaneous processings

[Tolriq]

Ok but how do I collect without closing then? I'm missing some conceptual stuff about Flow it seems, do you know any advanced tutorials? (And thanks for the help)

[gildor]

Ok but how do I collect without closing then?

What do you mean? flatMap concurrency limits how many Flow which it produced are consuming at the same time. It doesn’t close, just doesn’t subscribe on new

[Tolriq]

Everything seems terminal in Flow, what would be the equivalent of channels Poll ?

[gildor]

Flow is cold stream

[gildor]

it doesn’t use poll strategy, it uses push

[gildor]

It’s hard to discuss for me particular use case without some example of code or just some scheme, what kind behaviour you expect

[Tolriq]

The one I tried to explain at first, you proposed Flow so I tried to see how I can fit it 🙂 I'll try to explain better The need is that I need to pass "something" to a client that client needs to be able to take 1 up to X items from that something that will be removed. And that something needs to always have X items pre generated as the generation is heavy.

[Tolriq]

There's a few ways to achieve that but I'm pretty sure there's a "native" way to do that with channels or flow but just can't grasp it.

[gildor]

something that will be removed

What does it mean?

[Tolriq]

That's it's no more present in the source. Imagine the something as a diamond builder, creating a diamond takes times so you want to have some stock before selling them. But not too much in case no one purchase them. My first idea what to have a channel that the client pool and a loop that check the count of the pool and emit to it when under the X. But that sounds hacky and was wondering what is the proper way.

[gildor]

still not sure, about semantics, you dont’t want emit values before you have some amount of them already in buffer?

[Tolriq]

I can deal with both cases.

[gildor]

With flow you can just write custom operator using transform

[gildor]

so it will cache required amount and emit when it’s ready to emit

[Tolriq]

Ok but then previous question is back how do I collect only X without closing?

[Tolriq]

I really need to find a good tutorial I'm sure I'm just missing a piece of understanding

[gildor]

you just collect it one by one

[gildor]

val buffer = mutableListOf<Something>()

flow.collect {
buffer += it
if (buffer > 10) {
emit(buffer.removeFirst())
}
}

something like this

[Tolriq]

This does not support 2 collect ? Like for the diamonds, the one the guy is purchasing and another is looking at what will be available for purchase.

[gildor]

you can just merge any amount of lfows

[gildor]

flow = merge(flow1, flow2)
flow.collect {
buffer += it
   if (buffer > 10) {
     emit(buffer.removeFirst())
   }
}

[gildor]

another is looking at what will be available for purchase

One, who is looking is collect, isn’t it?

[Tolriq]

Yes it can collect (so remove) and I'll store at the shop level, or it can only look but not collect I can deal with both cases.

[gildor]

or not collect, you can use transform who does this, sorry my example above was not exactly correct, it should be:

val diamondMiner = flow { 
   while(true) {
     emit(mineDimond())
   }
}
val diamondStore = diamondMiner.transform { // your collector
    buffer += it
   if (buffer > 10) {
     //sends value to buyer
     emit(buffer.removeFirst())
   }
}

// purchase idamngs
diamondStore.collect {
  purchase(it)
}

[Tolriq]

Ok and so the one who want to look just look at the buffer. Sounds simple. Just last question, can I cancel the collect scope without cancelleling the store?

[gildor]

no

[gildor]

okay, it’s more complicated

[gildor]

see, Flow itself is single consumer abstraction, so if you start collecting it you have to close it or collect until it will end

[Tolriq]

That's what I had understood before having doubts 😛

[gildor]

now there are no operators to share flow

[gildor]

but work on them in progress

[Tolriq]

The close reuse is not that important it would avoid regenerating the X items on migration to another client but this is not a frequent thing.

[gildor]

share means that you can have 1 flow and just create some sort of shared flow, which can be closed without closing your original flow

[gildor]

so now it possible to use BrodactastChannel and convert it to flow

[gildor]

so consumers will use flow, but under the hood it will be channel, which will not be cancelled when flow is canceleed

[Tolriq]

From recent post from elizarov it's all deprecated to something that will come 🙂

[gildor]

It’s correct

[gildor]

but broadcast channel is still not deprecated

[Tolriq]

I suppose I'll wait a little for the shared flow stuff 🙂

[gildor]

it’s experimental and probably most of it usecases will be replaced with Flow and share operators

[gildor]

you can wait

[Tolriq]

There's no urgency in implementing this for me I think I'll wait now that I understand better thanks a lot for the details.

[gildor]

or you can implement everything what we discussed with RxJava or other feature-full Reactive library

[gildor]

it has all required operators

[gildor]

but transform will be not so easy to write

[Tolriq]

I'm full coroutines with actors and channels those I understand 😛 Won't add RX just for that

[gildor]

but if you have channels, what is probelm to use BroadcastChannel?

[gildor]

it will be quite easy later convert to Flow with sharing, if it would require

[Tolriq]

I need to use BroadcastChannel + asFlow ? As I don't know how to refill the channel.

[Tolriq]

Anyway don't want to take too much time from you, you already helped a lot.

[gildor]

You do not refill the channel

[gildor]

channel is your mine with miner

[gildor]

it just produces diamonds

[gildor]

and continue do this even if client is cancelled

[gildor]

Anyway don’t want to take too much time from you, you already helped a lot.

Hope it helped
Reactive approach requires a bit different mindset, so it needs time to adapt tasks to it, but it has many advantages
Like in your case consumer is simple, all caching logic is hidden above, no need to request particular amount of items

[Tolriq]

I have the basic mindset for simple cases 🙂 But still far from handing all complex cases with multiple endpoints 😞