ReceiveChannel<String>

successionEnds(duration)

flow

delay

source.stateData.testObserve

@Test
        @ExperimentalCoroutinesApi
        fun `should update state after load initial`() {
            val initialParams = mockk<PageKeyedDataSource.LoadInitialParams<Int>>(relaxed = true)
            val initalCallback = mockk<PageKeyedDataSource.LoadInitialCallback<Int, String>>(relaxed = true)

            runBlocking {

                // Fake loading
                source.loadInitial(initialParams, initalCallback)

                source.stateData.testObserve {
                    it shouldBeInstanceOf Result.State::class
                }
            }
        }

flow

flow<String> {
    emit("Foo")
}

channelFlow<String> {
    send("Foo")
}

FlowCollector

ProducerScope

CoroutineScope.cancel(cause: CancellationException? = null)

Unit

Nothing

cancel

throw CancellationException

job.cancel

throw

private val _state = ConflatedBroadcastChannel<SyncStatus>()
   val state: ReceiveChannel<SyncStatus> get() = _state.openSubscription()

  fun forceSync() {
    _state.offer(SyncStatus.SYNC)
    GlobalScope.launch {
      val result = if (load()) SyncStatus.IDLE else SyncStatus.FAILED
      _state.offer(result)
    }
  }

mockito

suspend fun

rxJava

Rule

delay()

collect {}

val hotFlow = baseFlow.conflate().broadcastIn(this).asFlow()

@FlowPreview
@ExperimentalCoroutinesApi
fun main() = runBlocking<Unit> {
    val baseFlow = (1..10).asFlow().delayEach(1_000)
    val hotFlow = baseFlow.conflate().broadcastIn(this).asFlow()
    launch {
        hotFlow.collect {
            println("A: $it")
        }
    }

    delay(5_000)
    launch {
        hotFlow.collect {
            println("B: $it")
        }
    }
}

GlobalScope.launch {
            flowOf(1,2,3,4,5)
                .delayEach(500)
                .map { it * it }
                .collect { println(it) } // Type mismatch: inferred type is () -> ??? but FlowCollector<Int> was expected
        }

val channel = Channel<Int>(1)

fun main() = runBlocking<Unit> {
    launcher().start()
    test().start()
}

fun CoroutineScope.launcher() = launch(start = CoroutineStart.LAZY) {
    while(isActive) {
        channel.send(i)
		delay(Random.nextLong(100, 6000))
    }
    
}

fun CoroutineScope.test() = launch(start = CoroutineStart.LAZY) {
    while (true) {
        doSomething(channel.receive())
    }
}

fun doSomething(value: Int) {
    println(value)
}

Flow

val flow = flow {
    for (i in 1..100) {
        emit(1)
        delay(200)
    }
}.flowOn(Dispatchers.Main) // try replacing it with Dispatchers.IO... 

MainScope().launch {
    flow.collect { 
        println(it)
        delay(1000)
    }
}

buffer()

.flowOn(Dispatchers.Main)

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

flow

suspend operator fun get

fun foo(): (suspend () -> Int) = { 5 }