StateFlow

Flow

runInterruptible

consumeAsFlow

receiveAsFlow

delay(1000L)

MutableStateFlow

A mutable [StateFlow] that provides a setter for [value] and a method to [close] the flow.

close

fun main() = runBlocking {
    val fl1 = flow {
        for (i in 1..3)
            emit(i)
    }
    val time = measureTimeMillis {
        fl1.map { slowComp(it) }.buffer().collect { println(it) }
    }
    println("TIME: $time")
}

suspend fun slowComp(i: Int): Int {
    delay(1000)
    return i * i
}

StateFlow

StateFlow

stateFlow.map { } as StateFlow<T>

fun asyncOp(nums: List<Int>, mult: Int, id: String) = GlobalScope.async {
    println("1 - $id: $mult times list computing asynchronously")
    Thread.sleep(2000L * mult)
    println("2 - $id: $mult times list computing asynchronously")
    Thread.sleep(1000L * mult)
    nums.map { it * mult }
}

suspend fun main() {

    val nums = listOf(1,2,3,4,5,6)
    val double = asyncOp(nums, 2,"first")
    val triple = asyncOp(nums, 3,"second")

    println("Double requested: $double")
    println("Triple requested: $triple")

    println("Waiting...")

    println(awaitAll(triple, double))
}

suspend fun asyncOp(nums: List<Int>, mult: Int, id: String) =
    withContext(Dispatchers.Default) {
        println("1 - $id: $mult times list computing asynchronously")
        Thread.sleep(2000L * mult)
        println("2 - $id: $mult times list computing asynchronously")
        Thread.sleep(1000L * mult)
        nums.map { it * mult }
    }

More than one file was found with OS independent path 'win32-x86-64/attach_hotspot_windows.dll'

BroadcastChannel.openSubscriber().consumeAsFlow()

launchIn(Scope)

Flow.first

class JobProcessor(val coroutineContext: CoroutineContext): CoroutineScope {
  val queue1 = Channel<Job>(Channel.UNLIMITED)
  val queue2 = Channel<Job>(Channel.UNLIMITED)

  init {
    while (isActive) {
      selectJob()
    }
  }

  private suspend fun selectJob() {
    select {
      queue1.onReceive { job -> process(job) }
      queue2.onReceive { job -> launch { process(job) } }
    }
  }

  fun sendJobToQueue1() { ... }
  fun sendJobToQueue2() { ... }

}

select

select

while (isActive && (!queue1.isEmpty || !queue2.isEmpty)) {
  selectJob()
}

suspend fun loadData(id: Long) = flow {
    fetchData(id) <-- now this is blocking function, how can I launch it in non blocking mode?
    roomDao.select(id)
        .collect {
            emit(result)
        }
}.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

suspend fun fetchData(id) {
    // make http request here and put data into database
}

fetchData

flow {}

Mutex.withLock { ... }

To run the coroutine only on the main UI thread, we should specify Dispatchers.Main as an argument:
launch(Dispatchers.Main) {
    updateResults()
}

Dispatchers.Swing

Dispatchers.Main

Dispatchers.Swing

Dispatchers.Main

Dispatchers.Swing

Dispatchers.Default

<http://Dispatchers.IO|Dispatchers.IO>

Executors.newSingleThreadExecutor().asCoroutineDispatcher()

Executors

CoroutineDispatcher

doSomethingElse()

suspend fun signOut() {
     coroutineScope {
         launch {
             makeNetworkRequest() // suspending function
         }
     }
     doSomethingElse()
}

CoroutineScope

ViewModel.viewModelScope

suspend

suspend

withSomething

try { block() } finally { … }

withSomethingSuspend

withContext

class UserApiClient(private val rxJavaBasedApi) {
    suspend fun getUser(): User = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
        rxJavaBasedApi.getUser().blockingGet()
    }
}

class org.test.reactive.kitchen.SuspendedKitchen {
  suspend fun getPotatoes(): Potatoes = ...
  fun chop(potatoes: Potatoes): ChoppedPotatoes = ...
  fun fry(fish: Fish): FriedFish = ...
  suspend fun fry(choppedPotatoes: ChoppedPotatoes): Chips = ...
}

launch {
  val kitchen = SuspendedKitchen()
   val chips = kitchen.getPotatoes()
        .let(kitchen::chop)
        .let(kitchen::fry) // <-- does not work
}

.let(kitchen:fry)

Callable reference resolution ambiguity: 
public final suspend fun fry(choppedPotatoes: ChoppedPotatoes): Chips defined in org.test.reactive.kitchen.SuspendedKitchen public final fun fry(fish: Fish): FriedFish defined in org.test.reactive.kitchen.SuspendedKitchen

launch {
  val kitchen = SuspendedKitchen()
   val chips = kitchen.getPotatoes()
     .let(kitchen::chop)
     .let { kitchen.fry(it) } // <-- works
}

class Foo {
   suspending fun foo() { /* some code */ }

coroutineScope {}

class Foo : CoroutineScope

/user/{userId}

userId

userId

userId

suspend fun doSomething() = withContext(coroutineContext) { 
   launch { 
       while(isActive) { 
           println("I'm Still Here")
           delay(1000)
       }
   } 
}

private val repository = DataRepository()
private val viewModelJob = SupervisorJob()
private val viewModelScope = CoroutineScope(Main+viewModelJob)
val mNowPlayingDataLiveData = MutableLiveData<NowPlayingMovies>()

fun DataFromrepository(page :Int){
    viewModelScope.launch {
                 val response = repository.getDataFromServer(page)

        if(response.isSuccessful){
                     mNowPlayingDataLiveData .value = response.body()
}}

COUNTER=0
while :
do
	COUNTER=$((COUNTER + 1))
	echo "$COUNTER"
	sleep 1
done

fun InputStream.linesToFlow() = channelFlow<String> {
    reader().forEachLine {
        offer(it)
    }
}

fun main() {
    runBlocking {
        val p = ProcessBuilder("./cmd.sh").start()
        p.inputStream.linesToFlow().map {
            println(it)
            it
        }.filter { "3" in it }.first()
        println("Done")
        p.waitFor()
    }
}

1
2
3
Done