I need bit of guidance, I'm trying to use coroutines to offload DB writes to another thread. I have this DB table that is read on app start and is all writes after that thanks to caching. I wanted to avoid blocking the calls for clients so I'm offloading writes using a channel (writes should be executed in-order so I used a channel). Is this ok? (GlobalScope will be replaced later)

internal class BlockedAppsStoreDatabaseImpl(private val blockedAppDao: BlockedAppDao) : BlockedAppsStore {

    private val channel = Channel<DbOp>(Channel.UNLIMITED)

    init {
        GlobalScope.launch {
            select {
                channel.onReceive { op ->
                    when (op) {
                        is Get -> op.cb.complete(blockedAppDao.getAll().toSet())
                        is Put -> blockedAppDao.putOrReplace(BlockedAppEntity(op.uid))
                        is Remove -> blockedAppDao.remove(op.uid)
                        is PutSet -> blockedAppDao.putOrReplace(op.uids)
                        is Fill -> blockedAppDao.fill(op.uids)
                        is RemoveAll -> blockedAppDao.removeAll()
                    }
                }
            }
        }
    }

    override fun getAll(): Set<Uid> {
        val deferred = CompletableDeferred<Set<Uid>>()
        channel.sendBlocking(Get(deferred))
        return runBlocking { deferred.await() }
    }

    override fun put(uid: Uid) {
        channel.sendBlocking(Put(uid))
    }
    ...

This still blocks the calling thread anyway. You need to make those functions suspend.

Get is intentionally blocking, but put shouldn't block, right?

That depends on what you want. Do you need the operations to conclude before the function returns?

I don't need DB write to complete before returning, I'm trying to skip that wait

Then you need to do a

launch

in the set function.

sendBlocking

doesn't block unless channel is full

You could use offer() directly if you know your channel will never block

nice one, thanks

You could simplify your code to avoid using a

select

and just use

channel.consumeEach { op ->

directly as well.

Thanks, that's all usefull

`actor`s are "obsolete" because they are being refactored to support complex cases with provided classes, but I doubt they will force us to implement a class just to use an actor with the newer complex stuff.

consumeEach

is tentatively graduating in 1.4.0. even if it doesn't, it's implementation is fairly simple

suspend fun <T> ReceiveChannel<T>.consumeEach(block: (T) -> Unit) {
    try {
        for (item in this) block(item)
    } catch (e: Throwable) {
        cancel(e as? CancellationException ?: CancellationException("Channel cancelled, consumer failed", e))
    } finally {
        cancel()
    }
}