A

B

C

Flow<B>

Flow<C>

Flow<A>

fun <T> loopFlow(initial: T, f: suspend (T) -> Flow<T>): Flow<T>

initial

f

private suspend fun sendPacketWithAcknowledgement(packet: RequestPacket) {
        try {
            withTimeout(PACKET_CONFIRMATION_TIMEOUT) {
                sendPacket(packet.bytes)
            }
            // Wait here until ack packet received
            // mutex.lock() // I thought this will suspend the code here and on mutex.unlock() it will resume here
        } catch (e: TimeoutCancellationException) {
            // do sth.
        }
    }

sendPacketWithAcknowledgement(StartCommunication()) // should be sequential
isCommunicationEstablished = true

isCommunicationEstablished = true

MutableSharedFlow<T>(replay = 0, extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)

1

test

doSomeOtherWork

launchIn

launchIn

suspend fun

suspend fun getChatUserFlow(uid: String): Flow<ChatUser?> {
        val results = userCollection.whereEqualTo(FieldPath.documentId(), uid).get().await()
        return if (!results.isEmpty) {
            results.documents[0].reference.asFlow<ChatUser>()
        } else {
            flowOf(null)
        }
    }

suspend fun <T> PreparedStatement.withCancellation(callback: suspend () -> T): T {
    val stmt = this
    suspendCancellableCoroutine { cont ->
        cont.invokeOnCancellation { 
            stmt.cancel()
        }
    }

    return callback()
}

callbackFlow

catch

suspendCancellableCoroutine

catch

Query()

suspend fun getUser():User

suspend fun getUserFromDb():User {

return withcontext (<http://dispatcher.IO|dispatcher.IO>) {return data.getUser()}

}

suspend

.scan(..)

SharedFlow

.scan(loadInitialValueAsynchronously()) { .. }

actor

SharedFlow

fun <T> StateFlow<T>.withPrevious(): Flow<Pair<T?, T>> {
    return flow {
        var previous: T? = null
        this@withPrevious.onEach {
            emit(Pair(previous, it))
            previous = it
        }.collect()
    }
}

fun main() = runBlocking {

    val httpClient = HttpClient {
        install(Logging) {
            level = LogLevel.ALL
        }

        @OptIn(KtorExperimentalAPI::class)
        install(WebSockets)

        expectSuccess = false

        install(JsonFeature) {
            @OptIn(KtorExperimentalAPI::class)
            acceptContentTypes = listOf(ContentType("application", "json-rpc"))
            serializer = KotlinxSerializer()
        }
    }
    val wss =httpClient.webSocketSession(method = HttpMethod.Get, host = "127.0.0.1", port = 6800, path = "/jsonrpc")
    wss.close()
}

Flow

throttleFirst

async

null

class ApiClient {

    private var deferredLoad: Deferred<Result>? = null

    suspend fun load(throttleMillis: Long = 300): Result? {
        return coroutineScope {
            if (deferredLoad?.isCompleted != false) {
                async {
                    val result = /** make GET API call **/
                    delay(throttleMillis)
                    result
                }
                .apply { deferredLoad = this }
                .await()
            } else {
                null
            }
        }
    }
}

var

Deferred

Mutex

.asSingle()

Deferred<T>

kotlinx-coroutines-rx2

coroutineContext

.asSingle()

suspendCancellableCoroutine

suspendCoroutine

CancellableContinuation::invokeOnCancellation

override suspend fun saveConfession(text: String): SaveConfessionResponse = suspendCoroutine {cont ->
        val confessionToSave = ConfessionDataModel(text)
        val call = newConfessionService.saveConfession(confessionToSave)
        call.enqueue(object : Callback<SaveConfessionResponse>{
            override fun onResponse(call: Call<SaveConfessionResponse>, response: Response<SaveConfessionResponse>) {
               cont.resume(response.body()!!)
            }

            override fun onFailure(call: Call<SaveConfessionResponse>, t: Throwable) {
                cont.resumeWithException(t)
            }
        })
    }

And I am trying to call it from my useCase object

   suspend fun execute(text: String): Result = suspendCoroutine{ cont->
        try {
            val response = confessionRepository.saveConfession(text)
            when (response.status) {
                200 -> Result.Success(response.id, 200)
                400 -> Result.Error(null)
                500 -> Result.Error(null)
                else -> Result.Error(null)
            }
        } catch (e: Exception) {
            return Result.Error(e)
        }
    }

ExperimentalCoroutinesApi

FlowPreview