Calling a coroutine on Dispatchers.IO (for saving Api results) inside a coroutine running (for Api call) on Dispatchers.IO to simulate a threadpool

Question

I have a usecase where I want to save to a database after making an API call. I want to do all of this in a single operator/invokee function.

One option is to create a custom coroutine context and then use that as a threadpool. I know Dispathers.IO is already a threadpool. Can I launch the coroutine for the database inside the invoke function after the API call is complete?

Would be something like

suspend operator fun invoke(): Resource<List<JobPW>> {
        try {
            val response = jobsRetroInterface.getJobsList(
                

           
       CoroutineScope(Dispatchers.IO).launch {
                jobsList?.forEach {
                    try {
                        //ToDo make sure that the owners have been set first in code
                        saveJobPWToDatabase.invoke(it)
                    } catch (ex: Exception) {
                        Log.e(TAG, "exception saving job to database ${it.job?.job_id}", ex)
                    }
                }
            }

I don't want the coroutine to wait for the thread to save the items to the database. I would like to save the items to the database on a different thread, so the invoke function can return with the data from the API immediately.

I found what I thought could be another approach; but it's not working. I tried to create a coroutineContext from an executor threadpool like this.

  val threadPoolContext = Executors.newFixedThreadPool(3).asCoroutineDispatcher()

I'm trying this

CoroutineScope(NetworkUtils.threadPoolContext).launch {
                jobsList?.forEach {
                        try {
                            //ToDo make sure that the owners have been set first in code
                            saveJobPWToDatabase.invoke(it)
                        } catch (ex: Exception) {
                            Log.e(TAG, "exception saving job to database ${it.job?.job_id}", ex)
                        }
                    }
            }
            return  convertResponseToJobsResource(jobsList)

Here is the whole invoke function for your reference.

    suspend operator fun invoke(): Resource<List<JobPW>> {
        try {
            val response = jobsRetroInterface.getJobsList(
                TokenUtils.getFirebaseUid(PosterPalApplication.appContext)!!,
                TokenUtils.getFirebaseAuthToken(PosterPalApplication.appContext)!!)

            if (response == null) {
                Log.e(TAG, "jobs response is response is null")
                return Resource.Error("Failed to connect to server")
            }

            val jobsList = response?.entrySet()?.map {
                val id = it.key
                val jobJson = it?.value?.asJsonObject
                return@map NetworkUtils.gson.fromJson(jobJson, JobPW::class.java).apply {
                    this.job?.job_id = id!!
                }!!
            }

            runBlocking {
//
                launch(NetworkUtils.threadPoolContext) {
                    jobsList?.forEach {
                        try {
                            //ToDo make sure that the owners have been set first in code
                            saveJobPWToDatabase.invoke(it)
                        } catch (ex: Exception) {
                            Log.e(TAG, "exception saving job to database ${it.job?.job_id}", ex)
                        }
                    }
                }

            }
            return  convertResponseToJobsResource(jobsList)
        } catch (ex: IOException) {
            Log.e(MainViewModel.TAG, "failed to get from remoted database, maybe no internet  ${ex.message}")
            return Resource.Error("No internet connection")
        } catch (ex: HttpException) {
            Log.e(MainViewModel.TAG, "failed to get from remoted database, maybe bad response  ${ex.message}")
            return Resource.Error("No internet connection")
        } catch (ex: Exception) {
            Log.e(MainViewModel.TAG, "failed to get from remoted database, unknown issue ${ex.message}")
            return Resource.Error("Failed to connect to server")
        }
    }

Answer 1

1. You don't need separate ThreadPool, you need separate CoroutineScope that is tied to lifetime of Database. Creating new CoroutineScope as shown in your code for every call is bad practice, you're not closing them.
2. Inject that CoroutineScope like how you inject the Database object
3. Call coroutineScope.launch{}

Another option:

Having some DatabaseHelper which will contain Database, CoroutineScope etc.. Then your use-case will call DatabaseHelper.saveAsync(). Encapsulating like this is better than passing Database's CoroutineScope around (someone might close/cancel CoroutineScope which is not desired). You can also fake/mock DatabaseHelper for testing.

DatabaseHelper

// Probably Singleton class
class DatabaseHelper{
    private val database: MyDatabaseOrORM
    private val scope: CoroutineScope

    init{
        database = ...
        scope = CoroutineScope(SupervisorJob()+Dispatchers.IO)
    }

    fun saveJobPWAsync(jobList:List<JobPW>){
        scope.launch{
            jobList.forEach{ job->
                // database.insert(job) ...
            }
        }
    }  

    // not necessary if DatabaseHelper is Singleton
    fun close(){
        // database.close()
        // scope.cancel()
    }  
}

Use-Case

suspend operator fun invoke(): Resource<List<JobPW>> {
        try {
            val response = jobsRetroInterface.getJobsList(
                TokenUtils.getFirebaseUid(PosterPalApplication.appContext)!!,
                TokenUtils.getFirebaseAuthToken(PosterPalApplication.appContext)!!)

            if (response == null) {
                Log.e(TAG, "jobs response is response is null")
                return Resource.Error("Failed to connect to server")
            }

            val jobsList = response?.entrySet()?.map {
                val id = it.key
                val jobJson = it?.value?.asJsonObject
                return@map NetworkUtils.gson.fromJson(jobJson, JobPW::class.java).apply {
                    this.job?.job_id = id!!
                }!!
            }

            databaseHelper.saveJobPWAsync(jobsList)

            return  convertResponseToJobsResource(jobsList)
        } catch (ex: IOException) {
            Log.e(MainViewModel.TAG, "failed to get from remoted database, maybe no internet  ${ex.message}")
            return Resource.Error("No internet connection")
        } catch (ex: HttpException) {
            Log.e(MainViewModel.TAG, "failed to get from remoted database, maybe bad response  ${ex.message}")
            return Resource.Error("No internet connection")
        } catch (ex: Exception) {
            Log.e(MainViewModel.TAG, "failed to get from remoted database, unknown issue ${ex.message}")
            return Resource.Error("Failed to connect to server")
        }
    }