I am a bit new to coroutines/flow and wondering if this is a good design or not. I have a server that can receive long running job to process. Each job must be run sequentially. So I have something like this:

class JobsMgr {
    private val _jobsToProcess = Channel<Job>(Channel.UNLIMITED)
    private val _jobsProcessed: Flow<Job> = _jobsToProcess.consumeAsFlow().map { job -> /* processing */ }.flowOn(Dispatchers.Default)

  init {
       // I find this very ugly.. is there a better way?
        Executors.newSingleThreadExecutor().execute {
            runBlocking {
                _jobsProcessed.collect {
                    delay(10 * 1000)
                    _lock.withLock { _jobs.remove(it.jobId) }
                }
            }
        }
  }

fun enqueueJob(...) { 
//...
               runBlocking {
                _jobsToProcess.send(job)
            }
}

}

It is running in a spring boot application so I do not have control over the main loop. This is why I created a process to collect and remove the completed jobs. Is this how Channel/Flow is supposed to be used? And is there a better way to "start" the collection. That seems like ugly code to me but not sure how to do better.

I would take a look at the

actor

coroutine builder. An actor is guaranteed to process its messages sequentially.

i’d say actor does not address the underlying issue of OP. It seems to be more how to switch from non-coroutine world to coroutine world.

Thank you for the answers. The

delay

/

remove

part is because the job (once completed) is stored in a map so you can still access it and get details of a recent job that was run. After a while (10s is for testing purposes...) the job is deleted from the map to avoid filling up the memory

CoroutineScope(job + dispatcher)

where job is

Job()

or

SupervisorJob()

depending on expected behavior. And dispatcher is most likely one of

<http://Dispatchers.IO|Dispatchers.IO>

and

Dispatchers.Default

Awesome thanks!