One shot cloud db request with callback flow

get_user_data.kt

    @ExperimentalCoroutinesApi
    suspend fun getUserData(id : String?) : Flow<Resource<User>> = withContext(ioDispatcher) {
        callbackFlow {

            if (id == null) {
                offer(Resource.Error(Exception("Id must not be null")))
                return@callbackFlow
            }

            // 1- Create query
            val query: CloudDBZoneQuery<User> = CloudDBZoneQuery.where(User::class.java).equalTo("accountId", id)
            // 2 - Create task
            val queryTask: CloudDBZoneTask<CloudDBZoneSnapshot<User>> = cloudDBZone.executeQuery(
                query,
                CloudDBZoneQuery.CloudDBZoneQueryPolicy.POLICY_QUERY_FROM_CLOUD_PRIOR
            )

            try {
                // 3 - Listen callbacks
                offer(Resource.Loading)
                queryTask
                    .addOnSuccessListener {
                        LogUtils.i("queryTask: success")
                        // Get user data from db
                        if (it.snapshotObjects != null) {

                            // Check item in db exist
                            if (it.snapshotObjects.size() == 0) {
                                offer(Resource.Error(Exception("User not exists in Cloud DB!")))
                                return@addOnSuccessListener
                            }

                            while (it.snapshotObjects.hasNext()) {
                                val user: User = it.snapshotObjects.next()
                                offer(Resource.Success(user))
                            }
                        }
                    }
                    .addOnFailureListener {
                        LogUtils.e(it.localizedMessage)
                        it.printStackTrace()
                        // Offer error
                        offer(Resource.Error(it))
                    }


            } catch (e : Exception) {
                LogUtils.e(e.localizedMessage)
                e.printStackTrace()
                // Offer error
                offer(Resource.Error(e))
            }

            // 4 - Finally if collect is not in use or collecting any data we cancel this channel
            // to prevent any leak and remove the subscription listener to the database
            awaitClose {
                queryTask.addOnSuccessListener(null)
                queryTask.addOnFailureListener(null)
            }
        }
    }