Flow

Observable.create(
            { emitter ->
                val process: Process = ProcessBuilder(commandAndArgs)
                    .redirectErrorStream(true)
                    .start()

                emitter.setCancellation {
                    process.destroy()
                }
        )

suspend fun send(someId) {
   val currentState = db.getState(someId)
   if (currentState == null) error("Unknown id")
   if (currentState == Sent) return // idempotent
   if (currentState == Sending) .. mutex?

   try {
      db.setState(Sending, someId)
      api.send(..)
      db.setState(Sent, someId)
   } catch (ex) {
      db.setState(Idle, someId)
   }
}

1.5.0-RC-native-mt

viewModelScope.launch { 
    flow1.collect {}
    flow2.collect {}
}

fun <T> LifecycleOwner.stateFlow(stateFlow: StateFlow<T>, funCollect: (T) -> Unit) {
        lifecycleScope.launchWhenStarted {
            stateFlow.collect() {
                funCollect(it)
            }
        }
    }

Timber.d(//This log is hit)
            model.data.emit(result)

@Test
fun `test network call`() {
    var response: Response<Projects>
    GlobalScope.async {
        response = service.getMostPopularProjects()
    }

    assertThat(response.body().popularProjectSize().toString()).isGreaterThan("1")
}

trySend

tryReceive

receiveCatching

poll

offer

callbackFlow

channelFlow

CoroutinesTimeout

java.lang.Process

waitFor

ClassCastException: kotlin.coroutines.intrinsics.CoroutineSingletons cannot be cast to ...

suspend fun

class Syncer(private val somePartialSyncComponentNeverRelatedToUi) {
   private val scope = CoroutineScope(<http://Dispatcher.IO|Dispatcher.IO>)

   fun sync() {
      scope.launch {
         somePartialSyncComponent.doWhatever()
      }
   }
}

class SomePartialSyncComponentNeverRelatedToUi {
   suspend func doWhatever {
      withContext here ??? <--------
   }
}

suspend fun

val message = objectMapper.readTree(inputString)

viewmodelScope.launch {
      val deferredFetchAndPersist = async {
          some suspend fun()
      }
      val deferredTimer = async {
          delay(3000)
      }
      awaitAll(deferredFetchAndPersist, deferredTimer)

      hideImageView()
   }

fun <T> debounce(
    delayMillis: Long = 300L,
    scope: CoroutineScope,
    action: (T) -> Unit
): (T) -> Unit {
    var debounceJob: Job? = null
    return { param: T ->
        if (debounceJob == null) {
            debounceJob = scope.launch {
                action(param)
                delay(delayMillis)
                debounceJob = null
            }
        }
    }
}

fun setDebounceListener(view: Button, onClickListener: View.OnClickListener) {
    val clickWithDebounce: (view: View) -> Unit =
        debounce(scope = MainScope()) {
            onClickListener.onClick(it)
        }
    view.setOnClickListener(clickWithDebounce)
}

GlobalScope

CoroutineScope

GlobalScope

GlobalScope

.value

Flow

.value

.value

emit

Flow

@InternalCoroutinesApi
    override fun isDispatchNeeded(context: CoroutineContext): Boolean {
        if (Bukkit.isPrimaryThread()) return false

        // Check if the primary server thread is blocked during the execution of this coroutine.
        // If it is, using the Bukkit scheduler would cause a deadlock, as scheduled tasks
        // are executed at the end of the tick, which requires the server thread to be running.

        // To detect whether the server thread is blocked, we traverse the hierarchy of Jobs upwards
        // to check if there's a coroutine running in a [runBlocking] context.
        // If so, we don't dispatch, causing the action to be executed on the current thread.
        // As the primary thread is locked, this has no thread-safety implications with regards to
        // the Bukkit API.
        // However, Bukkit methods that check whether they're performed on the server thread,
        // such as spawning an entity, will fail exceptionally - if this becomes an issue,
        // we can move the [isDispatchNeeded] logic into [dispatch], toggling a global flag while
        // directly executing the runnable, and patch the server software to respect this flag
        // during those checks.

        var job: Job? = context[Job]
        while (job != null) {
            if (job.javaClass == blockingCoroutineClass) {
                // we found a coroutine spawned via runBlocking - check whether it blocks the
                // primary server thread using the same logic as [CraftServer.isPrimaryThread]
                val blockedThread = Reflect.on(job).field("blockedThread").get<Thread>()
                if (blockedThread == MinecraftServer.getServer().serverThread ||
                    blockedThread == MinecraftServer.getServer().shutdownThread ||
                    blockedThread is TickThread) {
                    return false
                }
            }

            if (AbstractCoroutine::class.java.isAssignableFrom(job.javaClass)) {
                // traverse context hierarchy upwards
                job = Reflect.on(job).field("parentContext").get<CoroutineContext>()[Job]
            } else {
                throw UnsupportedOperationException("Can't handle jobs of type ${job.javaClass}")
            }
        }

        return true
    }

AbstractCoroutine.parentContext

Job

CoroutineScope

init { // }

.dispose()

onDestroy()

lateinit var foo : Foo
onCreate() {
  foo = Foo() // resources get allocated (init openGL context, open a file, etc)
}

onDestroy() {
  foo.dispose() // must not forget!
}

private val _program = MutableStateFlow(NwProgram())
override val program: StateFlow<NwProgram> = _program

private val stateFlowScope: CoroutineScope by lazy { CoroutineScope(Dispatchers.Default) }

override val entities: StateFlow<List<NwEntity>> = program.transform {
    emit(it.entities ?: emptyList())
}.stateIn(
    scope = stateFlowScope,
    started = WhileSubscribed(5000),
    initialValue = emptyList()
)

e: java.lang.AssertionError: Unbound symbols not allowed

Execution failed for task ':coresdk:compileDebugKotlin'.
> Internal compiler error. See log for more details

Kotlin v1.5.0

coroutines v1.5.0

livedata

stateflow

@JvmInline
value class ValueClassId(private val value: UUID)

class SomeService {

    suspend fun outerFails(id: UUID): ValueClassId {
        return inner(id)
    }

    suspend fun outerWorks(id: UUID) = try {
        somethingTrowing()
        justValueId(id)
    } catch (e: RuntimeException) {
        fallbackValueId(id)
    }

    private suspend fun inner(id: UUID) = try {
        somethingTrowing()
        justValueId(id)
    } catch (e: RuntimeException) {
        fallbackValueId(id)
    }

    private suspend fun justValueId(id: UUID): ValueClassId {
        delay(1)
        return ValueClassId(id)
    }

    private suspend fun fallbackValueId(id: UUID): ValueClassId {
        delay(1)
        return ValueClassId(id)
    }

    private suspend fun somethingTrowing(): Unit {
        delay(1)
        throw IllegalArgumentException()
    }
}

fun main(): Unit = runBlocking {
    val s = SomeService()
    val id = UUID.randomUUID()

    // works fine
    s.outerWorks(id)

    // Fails on: ContinuationImpl.kt:33
    // val outcome = invokeSuspend(param)
    // if (outcome === COROUTINE_SUSPENDED) return
    s.outerFails(id)
}

Flow.flatMap

Flow.flatMapMerge

flatMap