Loading

first

flow {
    val data = query().first()

    val flow = if (shouldFetch(data)) {
        emit(Resource.Loading(data))

        try {
            saveFetchResult(fetch())
            onFetchSuccess()
            query().map { Resource.Success(it) }
        } catch (t: Throwable) {
            onFetchFailed(t)
            query().map { Resource.Error(t, it) }
        }
    } else {
        query().map { Resource.Success(it) }
    }
    emitAll(flow)
}

emitAll

emitAll

@ExperimentalCoroutinesApi
internal fun <T> Flow<T?>.coalesce(other: Flow<T>): Flow<T> {
    return distinctUntilChanged { old, new -> old == null && new == null }
        .transformLatest {
            if (it != null) {
                // TODO: Will "Latest" wrongly cancel this branch?
                emit(it)
            } else {
                emitAll(other)
            }
        }
}

class Worker(val id: Int) {
    fun work(): Worker {
        Thread.sleep(1000)
        return Worker(id)
    }
}


println(measureTimeMillis { runBlocking {
    listOf(Worker(1), Worker(2), Worker(3))
        .map { worker ->
            async(Dispatchers.Default) {
                val worker2 = worker.work()
                worker2.work()
            }
        }.awaitAll()
}})

listOf(Worker(1), Worker(2), Worker(3)).map {it.work()}.map{it.work()}

private val _isCacheReady = MutableStateFlow(false)
val isCacheReady = _isCacheReady.asStateFlow()

_isCacheReady.value = true

init {
  lobstersRepository.isCacheReady.onEach {
    _savedPosts.value = lobstersRepository.getAllPostsFromCache()
  }.launchIn(viewModelScope)
}

channelFlow

try

awaitClose

inline fun <ResultType, RequestType> networkBoundResource(
    crossinline query: () -> Flow<ResultType>,
    crossinline fetch: suspend () -> RequestType,
    crossinline saveFetchResult: suspend (RequestType) -> Unit,
    crossinline onFetchSuccess: () -> Unit = { },
    crossinline onFetchFailed: (Throwable) -> Unit = { },
    crossinline shouldFetch: (ResultType) -> Boolean = { true }
) = channelFlow {
    val data = query().first()

    if (shouldFetch(data)) {
        val loading = launch {
            query().collect { send(Resource.Loading(it)) }
        }

        try {
            saveFetchResult(fetch())
            onFetchSuccess()
            loading.cancel()
            query().collect { send(Resource.Success(it)) }
        } catch (t: Throwable) {
            onFetchFailed(t)
            loading.cancel()
            query().collect { send(Resource.Error(t, it)) }
        }
    } else {
        query().collect { send(Resource.Success(it)) }
    }
}

suspend fun

suspend

// stream is _really_ someDao.loadAll(): Flow<Something>
val stream = MutableStateFlow("cachedValue")

// refresh is _really_ fetchSomethingFromApi.also { someDao.updateData(it) }
val refresh = suspend {
    delay(1000)
    stream.value = "freshValueFromAPI"
}

suspend fun whyDoesThisNotWork(): Flow<String> = stream
    .onStart {
        coroutineScope {
            launch {
                refresh()
            }
        }
    }

suspend fun thisWorks(): Flow<String> = flow {
    coroutineScope {
        launch {
            refresh()
        }

        stream.collect {
            emit(it)
        }
    }
}

@Test
    fun testFoo() = runBlocking {
        val testCoroutineScope = TestCoroutineScope().apply {
            pauseDispatcher() // This needs to be here because the actual test, handles time.
        }
        val sharedFlow = MutableSharedFlow<Int>()
        val values = mutableListOf<Int>()
        println("before launch")
        val job = testCoroutineScope.launch {
            println("before collect")
            sharedFlow.collect {
                println("before adding $it")
                values.add(it)
            }
        }

        println("before emits")
        sharedFlow.emit(1)
        sharedFlow.emit(2)
        testCoroutineScope.runCurrent()

        assertEquals(mutableListOf(1, 2), values)

        job.cancel()
    }

return suspendCoroutine{
   continuation -> usecase.invoke{result -> continuation.resumeWith(Result.success(result)}
}

Flow.onEmpty

Flow.onNotEmpty

fun getData(page: Int, pageSize: Int): Flow<Item> = ...
fun getAllData() = getAllData(0, 5)

private fun getAllData(page: Int, pageSize: Int): Flow<Item> = flow {
  val data = getData(page, pageSize)
  emitAll(data)
  data.onNotEmpty { emitAll(getAllData(page + 1, pageSize) }
}

private fun getAllData(page: Int, pageSize: Int): Flow<Item> = flow {
  val data = getData(page, pageSize)
  emitAll(data)
  if (data.count() == pageSize) {         // Since `count` does a `collect`, another flow is created
    emitAll(getAllData(page + 1, pageSize) 
  }
}

SharedFlow

@Test
    fun `test event is emitted`() = coroutineRule.runBlockingTest {

       val viewModel = MyViewModel()
       launch {
            val elements = viewModel.mySharedFlow.take(1).toList()
            elements[0] shouldBeEqualTo MyEvent
        }
    }

val mySharedFlow = MutableSharedFlow<MyEvent>(extraBufferCapacity = 1)

 init {
    mySharedFlow.tryEmit(MyEvent)
 }

withContext(context)

async(context) {}

Deferred

Result<T>

Deferred<T>

await

async {}

withContext

withContext

first

SendChannel::isClosedForSend

SharedFlow

BufferOverflow.DROP_OLDEST

val parentScope = CoroutineScope(Dispatchers.Default + Job() + exceptionHandler)

val parentScope = MyScope()

MyScope

class MyScope: CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = Dispatchers.Default + Job() + exceptionHandler
}

inCoroutine {
    var foo: Foo? = Foo()
    val fooRef = WeakReference(foo)

    triggerAndWaitForGc()
    assert(fooRef.get() != null)

    println(foo) // Test passes if this is removed

    inCoroutine {
        foo = null
        triggerAndWaitForGc()
        assert(fooRef.get() == null)
    }
}

Foo

WeakRef

foo = null

Foo

println(foo)

Foo var9 = (Foo)foo.element;
var5 = false;
System.out.println(var9);

var9

Foo

class CoroutineDownloadWorker(
    context: Context,
    params: WorkerParameters
) : CoroutineWorker(context, params) {

    override suspend fun doWork(): Result {
        withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
            val data = downloadSynchronously("<https://www.google.com>")
            saveData(data)
            return Result.success()
        }
    }
}

return@withContext