override fun encode(
    inputStream: Publisher<out Any>,
    bufferFactory: DataBufferFactory,
    elementType: ResolvableType,
    mimeType: MimeType?,
    hints: MutableMap<String, Any>?
): Flux<DataBuffer> {
   1) return Flux.from(inputStream).map { bufferFactory.wrap(protobufSerializer.encodeToByteArray(elementType)) }
   vs 
   2) return inputStream
        .asFlow()
        .map { bufferFactory.wrap(protobufSerializer.encodeToByteArray(elementType)) }
        .asFlux()
}

channel

flow

flow.buffer(3).collect { action() }

val search = MutableStateFlow("")

suspend fun track(query: String) {
    delay(1000)
    println(query)
}

runBlocking {
    search
        .onEach { track(it) }
        .map {
            "The search is $it"
        }
        .collectLatest {
            println("Collected $it")
        }
}

runTest

@Test
fun `test shared flow with deferred`() = runBlockingTest {
    val sharedFlow = MutableSharedFlow<Int>(replay = 0)
    val deferred = async { sharedFlow.first() }
    sharedFlow.emit(1)
    assertEquals(1, deferred.await())
}

@Test
fun `test shared flow with deferred - runTest`() = runTest {
    val sharedFlow = MutableSharedFlow<Int>(replay = 0)
    val deferred = async { sharedFlow.first() }
    sharedFlow.emit(1)
    assertEquals(1, deferred.await())
}

Exception in thread "DefaultDispatcher-worker-6 @track-session/cu#80" java.lang.NullPointerException: Cannot invoke "kotlinx.coroutines.flow.Flow.collect(kotlinx.coroutines.flow.FlowCollector, kotlin.coroutines.Continuation)" because "this.$this_unsafeTransform$inlined" is null
	at kotlinx.coroutines.flow.FlowKt__TransformKt$onEach$$inlined$unsafeTransform$1.collect(SafeCollector.common.kt:113)
	at kotlinx.coroutines.flow.FlowKt__CollectKt.collect(Collect.kt:30)
	at kotlinx.coroutines.flow.FlowKt.collect(Unknown Source)
	at gragas.play.TrackSession.playRegisteredTracks(TrackSession.kt:117)
	at gragas.play.TrackSession.access$playRegisteredTracks(TrackSession.kt:38)
	at gragas.play.TrackSession$3.invokeSuspend(TrackSession.kt:58)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:106)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:570)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:749)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:677)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:664)
	Suppressed: kotlinx.coroutines.DiagnosticCoroutineContextException: [CoroutineName(track-session/cu), CoroutineId(80), "track-session/cu#80":StandaloneCoroutine{Cancelling}@6465a67d, Dispatchers.Default]

private suspend fun playRegisteredTracks() {
  queue
    .onEach { song ->
      player.playTrack(song)
    }
    .collect()
}

withContext(<http://Dispatchers.IO|Dispatchers.IO> + NonCancellable)

fun <V : Any> SharedPreferences.asFlow(
    key: String,
    defaultValue: V?,
    retrieve: SharedPreferences.(String, V?) -> V?,
): Flow<Optional<V>> {
    return callbackFlow {
        val prefsListener = SharedPreferences
            .OnSharedPreferenceChangeListener { sharedPreferences, k ->
                if (k == key) {
                    val value = sharedPreferences.retrieve(key, defaultValue)
                    trySendBlocking(value.asOptional())
                }
            }
        registerOnSharedPreferenceChangeListener(prefsListener)
        awaitClose {
            unregisterOnSharedPreferenceChangeListener(prefsListener)
        }
    }
        .onStart {
            emit(Optional.ofNullable(retrieve(key, defaultValue)))
        }
        .distinctUntilChanged()

onStart

runBLocking

fun test() = runTest(UnconfinedTestDispatcher(), dispatchTimeoutMs = 200) {
            val scope = this
            val counter = MutableStateFlow(0)
            counter.test {
                awaitItem() shouldBe 0
                scope.launch {
                    counter.value = 1
                    counter.value = 0
                }
                awaitItem() shouldBe 1
                awaitItem() shouldBe 0
            }
        }

StandardTestDispatcher

UnconfinedTestDispatcher

scope.launch {
  fooSuspendedFun()
  barSuspendedFun()
}

class AppIOCoroutineScope : CoroutineScope by CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>)

appIOCoroutineScope.launch {
  // do the work
}

context(Toggled) @Base
suspend fun simultaneously(
    context: CoroutineContext = EmptyCoroutineContext,
    block: suspend CoroutineScope.() -> (Unit)
) {
    val current = currentCoroutineContext()[ContinuationInterceptor]!! + context
    val scope = object : CoroutineScope { override val coroutineContext = current }
    var task: Job? = null
    onEnabled { task = scope.launch(context) { block() } }
    onDisabled {
        println("Children: ${task?.children?.count()}")
        println("Active: ${currentCoroutineContext()[Job]?.isActive}")
        try {
            task?.cancelAndJoin(); task = null
        } catch (reason: Throwable) {
            reason.printStackTrace()
        }
        println("Active: ${currentCoroutineContext()[Job]?.isActive}")
    }
}

block

null

fun <T : Any> takeFirstNotNull(flow1: Flow<T?>, flow2: Flow<T?>): Flow<T?> {
    return flow1.transform {
            if (it != null) emit(it) else emitAll(flow2)
    },
}

runTest { ... }

myScope (supervisor + Main + interceptor)

myScope.launch {
  text = mySusFunc()
}

suspend fun mySusFunc(): String {
return withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
          delay(1000)
         "myValue"
  }
}

text =

val callA = async { the call }
val callB = async { the call }

if (callA.await().isSuccess) {
     callB.cancel()
     return callA.await()
}else {
     callA.cancel()
     return callB.await()
}

@OptIn(ExperimentalCoroutinesApi::class)
inline fun <T, S> Flow<T>.mapDisposable(
    crossinline transform: suspend (value: T) -> S,
    crossinline onDispose: (S) -> Unit
) : Flow<S> {
    return flatMapLatest { value ->
        callbackFlow {
            val transformed: S = transform(value)
            trySend(transformed)
            awaitClose {
                onDispose(transformed)
            }
        }
    }
}

class Repository {
	private val idsFlow = MutableStateFlow<List<String>>(emptyList())

	fun getMyFlow(): Flow<List<String>> = idsFlow.onSubscription { 
		if (idsFlow.value.isEmpty() idsFlow.value = getRemoteIds()) 
	}

	private suspend fun getRemoteIds(): List<String> {
		delay(2_000)
		return listOf("1", "2")
	}

	.. 
	// other functions that can modify myFlow (ie. add / delete / update an item from it)
}