:wave: , I’d like to have a `suspending fun` that puts something in a queue and suspends until that...

# coroutines

[frankelot]

👋 , I’d like to have a suspending fun that puts something in a queue and suspends until that “something” gets its turn and is finished being processed

[frankelot]

seems trivial.. but I’m having trouble getting it to work

[frankelot]

the “processor” deals with shared mutable data so I want things to be processed sequentially one by one

[frankelot]

Tried using channels and actors but .send() suspends until the channel has buffer capacity and starts processing… not when its done processing

[Nick Allen]

Sounds like you just want a coroutine Mutex

[frankelot]

I think you are right, taking a look

[Casey Brooks]

The purpose of channels is to make a "pipeline" of async communication. Putting an item into a channel and then waiting for processing to complete isn't how they're intended to be used, which is why it's not working for you.

A better pattern would be to have something put an item into Channel A, and when that item is done have it put an item into Channel B to be processed later. And ultimately when the whole pipeline is finished, data is posted back to the original caller through a callback or something like that. This would loosely be considered an "actor pattern"

Alternatively, a Flow might be a better mechanism than raw channels, since by design the emitter will suspend until the collector has finished processing the item

[frankelot]

somehow the idea of “mutex are no longer needed with coroutines” got into my head

[Nick Allen]

The queue thing is possible with a CompleteableJob in the item to signal when done but it’s easy to mess up error situations and you lose natural structured concurrency

[frankelot]

I’ll explore both approaches mutex and flows and report back, thanks! for the help

[frankelot]

The purpose of channels is to make a “pipeline” of async communication. Putting an item into a channel and then waiting for processing to complete isn’t how they’re intended to be used, which is why it’s not working for you.

this makes sense.. I don’t want async communication

[frankelot]

I was just using an actor to “group” mutliple coroutines into one and avoid concurrent modifications to state (which should be done using a mutex)

[darkmoon_uk]

One approach to this would be to use the suspendCoroutine function e.g. in pseudocode:

fun queuedProcess(someParam: Param) = suspendCoroutine { continuation -> 
    addToQueue(param, continuation)
    ensureQueueIsRunning()
}

...and elsewhere, the queue processing calls queueItem.continuation(result)

[darkmoon_uk]

To add some description to that; suspendCoroutine suspends the execution at that point and gives you a Continuation object that you can invoke with a result when you want that point of execution to continue. You can then store this object as part of your queued operation, and invoke it when the queued process completes.

[K Merle]

You could possibly have a simple for loop of suspend functions and run them sequentially by your need.

[Casey Brooks]

The suspendCoroutine route works, but it might be easier and make the intent clearer to use a CompletableDeferred, which would effectively do the same thing