Coroutines with Flow example

coroutines_flow.kt

class UploadFilesUsecase() {
    private val MAX_RETRIES = 3L

    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)

    private val atomic = atomic<Flow<File>?>(null)

    suspend fun uploadFiles(): Flow<File> {
        atomic.value?.let { return it }
        val channel = ConflatedBroadcastChannel<File>()
        atomic.value = channel.asFlow()

        val fileA = flow { emit(processAndMergeFilesOfTypeA()) }
                .retry(MAX_RETRIES)
                .cacheConflate(scope)

        val fileB = flow { emit(processAndMergeFilesOfTypeB()) }
                .retry(MAX_RETRIES)
                .cacheConflate(scope)

     
        
        val result = fileA.zip(fileB) { a, b -> listOf(a, b) }
                .map { compressMergedFiles(it) }.retry(MAX_RETRIES).cacheConflate(scope)
                .onEach {
                    uploadFileToServer(it)
                    channel.send(it)
                }.retry(MAX_RETRIES)
                .onCompletion {
                    atomic.value = null
                    deleteTempDir()
                }
       
        return result
    }


    private suspend fun uploadFileToServer(archive: File) = suspendCoroutine<Int> { cont ->
        HttpManager.uploadFiles(archive,
                onDone = { code: Int, body: ByteArray ->
                    if (code / 100 == 2) {
                        cont.resume(code)
                    } else {
                        cont.resumeWithException(Throwable())
                    }

                },
                onFailure = {
                    cont.resumeWithException(Throwable())
                }
        )
    }

    private suspend fun processAndMergeFilesOfTypeA(): File = TODO()

    private suspend fun processAndMergeFilesOfTypeB(): File = TODO()

    private suspend fun compressMergedFiles(files: List<File>): File = TODO()

    fun deleteTempDir(): Unit = TODO()
}