class SampleViewModel : ViewModel() {
    private val interactions = Channel<Any>()
    private val state = MutableSharedFlow<Any>(
        onBufferOverflow = BufferOverflow.DROP_LATEST,
        extraBufferCapacity = 1
    )

    init {
        viewModelScope.launch {
            interactions.receiveAsFlow().collect { action ->
                processAction(action)
            }
        }
    }

    fun flow() = state.asSharedFlow()

    fun handleActions(action: Any) {
        viewModelScope.launch {
            interactions.send(action)
        }
    }

    suspend fun processAction(action: Any) {
        state.emit("Show loading state")

       // process something...

        state.emit("Show success state")
    }
}

Controller.kt

main()

fun <T> awsrun(f: (Ec2Client) -> T): List<T> {
        return clients.values.map { f(it) }
    }

Flow<T>

collect { }

fun doStuff(flow: Flow<String>) {
   scope.launch {
      flow.collect { 
        if(!it in myList) { 
           dataSource.add(it)
        }
      }
   }
}

dataSource.add(it)

myList

coroutineContext[ReactorContext.Key]

chain.filter

coroutineContext[ReactorContext.Key]?.context

yield()

[1, 2, 3] \n [1, 2, 3, 4] \n End

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val mutableStateFlow = MutableStateFlow(emptyList<Int>())

    val collectJob = launch {
        combine(
            flowOf(listOf(1, 2, 3)),
            mutableStateFlow
        ) { list1, list2 ->
            list1 + list2
        }.collect { list ->
            println(list)
        }
    }

    yield()
    mutableStateFlow.value = listOf(4)
    yield()
    collectJob.cancel()
    
    println("End")
}

delay(100)

yield()

class ChannelWorker() {
    private val requestChannel = Channel<FooRequest>()
    private val responseChannel = Channel<FooResult>()

    init {
        GlobalScope.launch {
            while(true) {
                val receive = requestChannel.receive()
                val resp = fetchStuff(receive.id)
                responseChannel.send(FooResult(resp))
            }   
        }
    }

    private suspend fun fetchStuff(id: String): Int {
        delay(1000)
        return id.toInt()
    }
}

class ChannelWorker {
    private val requestChannel = Channel<FooRequest>()
    private val responseChannel = Channel<FooResult>()

    init {
        CoroutineScope(EmptyCoroutineContext).launch {
            for (message in requestChannel) {
                val resp = fetchStuff(message.id)
                responseChannel.send(FooResult(resp))
            }
        }
    }

    private suspend fun fetchStuff(id: String): Int {
        delay(1000)
        return id.toInt()
    }
}

stateIn

// presenter/viewmodel:

val myState: StateFlow<MyState> = api.subscribeToFlow()
  .map {..}
  .stateIn(scope = mainScope + Dispatcher.Default, started = SharingStarted.Lazily, initalValue = MyState.Initial)

// somewhere else:
fun subscribeToFlow(): Flow<Int> {
  println("thread: ${Thread.currentThread()}.") // prints main thread
  ...
  return someFlow()
}

api.subscribeToFlow()

withContext()

subscribeToFlow()

suspend stateIn(scope: CoroutineScope)

stateIn

import kotlinx.coroutines.launch
import kotlinx.coroutines.delay
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.async
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.flowOf
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.runBlocking

suspend fun flowsWithCombine() {
    val numbersFlow = flowOf(1, 2, 3)
    val lettersFlow = flowOf("A")
    combine(numbersFlow, lettersFlow) { number, letter ->
        "$number$letter"
    }.collect {
       // delay(100)
        println(it)   
    }
}
fun main() = runBlocking {
    val deferred: Deferred<Unit> = async {
         println("waiting...")
        flowsWithCombine()
    }
    println(deferred.await())
}

waiting...
1A
2A
3A
kotlin.Unit

delay

waiting...
1A
3A
kotlin.Unit

mutableStateFlow.value = newValue
mutableStateFlow.emit(newValue)

runBlocking(<http://Dispatchers.IO|Dispatchers.IO>)

runBlocking { }

1.6.20

suspendCoroutine

var topupCountries: Deferred<Set<String>> = buildCountriesAsync()
    private set

private fun buildCountriesAsync() = CoroutineScope(backgroundDispatcher).async(start = CoroutineStart.LAZY) {
    getCountries()
}



private suspend fun getCountries(): Set<String> {
    return try {
        topupCountries.getCompleted()
    } catch (exception: IllegalStateException) {
    
        // Crash here
        fetchRestrictedTopupCountries()
    } }

private suspend fun fetchRestrictedTopupCountries(): Set<String> {
    return suspendCoroutine { continuation ->
        val request = ApiTopupRequestFactory.getTopupCountries(
            listener = {
                val setResult = it.countries.toSet()
                continuation.resume(setResult)
            },
            errorListener = {
                continuation.resume(setOf())
            }
        )
        execute(request)
    }
}

java.lang.ClassCastException: kotlin.coroutines.intrinsics.CoroutineSingletons cannot be cast to java.util.Set

StateFlow<NodeState>

NodeManager

MutableList<Node>

List<Node>

List<NodeState>

StateFlow<List<Node>>

NodeManager

override suspend fun requestBooks(): Flow<List<Book>> {
  return callbackFlow {
    val listener =
        FirebaseFirestore.getInstance().collection("books").addSnapshotListener { value, error
          ->
          trySend(value!!.toObjects())
        }
    awaitClose { listener.remove() }
  }
}

trySend(value!!.toObjects())

value!!.toObjects()

trySend

launch(<http://Dispatchers.IO|Dispatchers.IO>) { 
  trySend(value!!.toObjects())
}

viewModel

init{}

viewModel

sharedStateFlow

viewModel

init

viewModel

init

sharedFlow

@Test 
fun test = runTest { 
  val viewModel = createViewModel() <-- already call the init
  viewModel.stateFlow.test { 
    //What should I call here? If it's in init...
  }
}

StateFlow

StateFlow

CoroutineScope

mapSync

import kotlinx.coroutines.flow.*

fun <T, R> StateFlow<T>.mapSync(transform: (T) -> R): StateFlow<R> = object : StateFlow<R> {

    override val replayCache: List<R> get() = listOf(value)

    override suspend fun collect(collector: FlowCollector<R>): Nothing {
        this@mapSync.collect {
            collector.emit(transform(it))
        }
    }

    private var lastUpstreamValue = this@mapSync.value

    override var value: R = transform(lastUpstreamValue)
        private set
        get() {
            val currentUpstreamValue: T = this@mapSync.value
            if (currentUpstreamValue == lastUpstreamValue) return field
            field = transform(currentUpstreamValue)
            lastUpstreamValue = currentUpstreamValue
            return field
        }
}

runTest

runCurrent()

suspend fun getData(): Data = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    val a = async(<http://Dispatchers.IO|Dispatchers.IO>) {
        delay(1000)
        Data("some data...")
    }
    a.await()
}

var toReturn: PixelArt? = null

    CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>).launch {
        AppData.pixelArtDB.pixelArtCreationsDao().getAllPixelArtCreations().observe(this@extendedGetCurrentPixelArtObj.findViewTreeLifecycleOwner()!!) {
            toReturn = it[currentIndex]
        }
    }

    return toReturn!!

.await()

SharedFlow

class MyClass(coroutineScope: CoroutineScope, flowToCollect: Flow<Int>) {
    
    var lastObservedResult: Int? = null

    init {
        coroutineScope.launch {
            flowToCollect.collect { lastObservedResult = it }
        }
    }
}

runTest

TestScope

@Test
fun testMyClass() = runTest {
    MyClass(this, flow)
    // do something here, make some assertions etc.
    // at the end, the test is going to fail because of the running coroutine
}

After waiting for 60000 ms, the test coroutine is not completing, there were active child jobs (...)

val dispatcher = UnconfinedTestDispatcher()

@Test
fun testMyClass() = runTest(dispatcher) {
    val additionalScope = CoroutineScope(dispatcher)
    MyClass(additionalScope, flow)
    // do something here, make some assertions etc.

    additionalScope.close() // Is this necessary, btw? Is the launched coroutine going to leak after the test is finished or something?
    // now the test won't fail, but I must remember to close the additional scope manually
}

fun asyncTest(
    context: CoroutineContext = EmptyCoroutineContext, // in practice, it was never changed
    timeoutInMillis: Long = 15000,
    block: suspend CoroutineScope.() -> Unit
) {
    runBlocking(context) {
        withTimeout(timeoutInMillis, block)
    }
}

runTest

fun asyncTest(
    context: CoroutineContext = UnconfinedTestDispatcher(),
    timeoutInMillis: Long = 15000,
    block: suspend TestScope.() -> Unit
) = runBlocking {
    withTimeout(timeoutInMillis) {
        runTest(context, timeoutInMillis, block)
    }
}

runTest

private val testCoroutineScheduler = TestCoroutineScheduler()
private val standardTestDispatcher = StandardTestDispatcher(testCoroutineScheduler)

@Test
fun `should execute the coroutine with delays`() = runTest(standardTestDispatcher) {
    var iterations = 0

    val job = launch {
        while(isActive) {
            println("iteration")
            iterations++
            delay(10)
        }
    }

    iterations.shouldEqual(0) // this is an assertion

    runCurrent()
    iterations.shouldEqual(1)

    advanceTimeBy(5)
    iterations.shouldEqual(1)
    advanceTimeBy(5)
    iterations.shouldEqual(1)

    runCurrent()
    iterations.shouldEqual(2)

    advanceTimeBy(10)
    runCurrent()
    iterations.shouldEqual(3)

    job.cancel()
}

job.cancel()

println

private val testCoroutineScheduler = TestCoroutineScheduler()
private val standardTestDispatcher = StandardTestDispatcher(testCoroutineScheduler)

class Refresher(scope: CoroutineScope, refreshingDispatcher: CoroutineDispatcher) {

    var refreshCount = 0

    init {
        scope.launch(refreshingDispatcher) {
            while (isActive) {
                refreshCount++
                delay(10)
            }
        }
    }
}

@Test
fun `refreshing should work`() = runTest(standardTestDispatcher) {
    val refresher = Refresher(this, standardTestDispatcher)

    refresher.refreshCount.shouldEqual(0)
    runCurrent()
    refresher.refreshCount.shouldEqual(1)
    advanceTimeBy(15)
    refresher.refreshCount.shouldEqual(2)

    // this test never ends and I can't cancel the launched coroutine
}

runTest

@Test
fun `refreshing should work`() {
    val refresher = Refresher(CoroutineScope(standardTestDispatcher), standardTestDispatcher)

    refresher.refreshCount.shouldEqual(0)
    testCoroutineScheduler.runCurrent()
    refresher.refreshCount.shouldEqual(1)
    testCoroutineScheduler.advanceTimeBy(15)
    refresher.refreshCount.shouldEqual(2)
}

.test

val flow = MutableStateFlow(1)
    flow.test {
        assertEquals(awaitItem(), 1)
        assertEquals(awaitItem(), 2) // <-- waits forever
}
flow.emit(2)

callbackFlow

awaitClose()