suspend fun someFunctionCreatesAFlow(context: Dispatchers.Default): Flow<String> = withContext(context) {
 flow {
    emit(somethingAsAString)
   } 
}

flow::flowOn()

coroutineScope

if (collection.isNotEmpty()) {
  coroutineScope {
    collection.forEach { item -> launch { doSomeComplexSuspendingWork(item) } }
  }
}

coroutineScope {
  collection.forEach { item -> launch { doSomeComplexSuspendingWork(item) } }
}

rxjava

flow

subscribe

flow

flowable

runBlocking {
            getSubtitleFromProjectDataManager()
        }

private suspend fun getSubtitleFromProjectDataManager() {
        scope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {

            val flow = projectDataManager.getProjectNameFlowable().asFlow()
            flow.collect {
                subtitle = it

            }
            launch(Dispatchers.Main) {
                view?.rebindToolbar()
            }
        }
    }

public Flowable<String> getProjectNameFlowable() {
        return RxJavaInterop.toV2Flowable(
                getSelectedProject()
                        .switchMap(projectEntity -> {
                                    Timber.e("fire projectEntity " + projectEntity.title());
                                    return Observable.just(projectEntity.title());
                                }
                        ));
    }

class Foo(ctx: CoroutineContext) {
    val scope = CoroutineScope(ctx)

    fun launchSome() = scope.launch { ... }
}

class Foo(ctx: CoroutineContext = ... : CoroutineScope { ... }

fun CoroutineScope.launchSome() = launch(ctx) {...}

GlobalScope.launch {
    flow.collect {
    }
}

Exception in thread "main" java.lang.IllegalStateException: Flow invariant is violated:
		Emission from another coroutine is detected.
		Child of StandaloneCoroutine{Active}@2b552920, expected child of BlockingCoroutine{Completing}@2758fe70.
		FlowCollector is not thread-safe and concurrent emissions are prohibited.
		To mitigate this restriction please use 'channelFlow' builder instead of 'flow'

fun updateDetails() = GlobalScope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {
    // loop forever
    updateDetailsAsync()
    // wait until updateDetailsAsync is done
}

fun CoroutineScope.updateDetailsAsync() {
    does an REST API lookup
    writes into a database
}

val scope = SupervisorScope()

val scope = CoroutineScope(CoroutineExceptionHandler { ctx, e -> })

Flowable.never()

object NeverFlow : Flow<Nothing> {
    override suspend fun collect(collector: FlowCollector<Nothing>) {
        delay(Long.MAX_VALUE)
    }
}

fun main() = runBlocking {
    val f = channelFlow {
        launch {
            while (true) {
                offer(1)
                delay(500)
            }
        }

        awaitClose {
            println("f1 closed")
        }
    }

    val b = f.broadcastIn(this).apply {
        /**
         * If uncomment the following code, Exception below will be raised.
         *
         * java.lang.IllegalStateException: Another handler was already registered 
         *
         * You cannot onClose callback twice for same SendChannel.
         *
         * There is already onClose callback in this code: awaitClose callback inside channelFlow builder
         *
         */
        // invokeOnClose { println("b1 closed") }
    }

    launch {
        b.openSubscription().consumeAsFlow().take(5).collect {
            println(it)
        }
        b.cancel()
    }.join()
}

invokeOnClose

java.lang.IllegalStateException Another handler was already registered

broadcastIn

fun cancel(id: Id)

println("Start receiving...

await()

await()

class Handler : MessageHandler {

    private val channel = Channel<Message>(1)

    override fun handle(message : Message) {
        // TODO send event.
    }

    fun send(event : Message) {
        
    }

/*
    fun observe() : Channel<Message>
    fun observe() : Flow<Message>
*/
}

subjects

onNext()

Observable

Coroutines

handle(message : Message)

java.util.Stream

Flow

result.iterator().asFlow().onCompletion { result.close() }

flow {
    result.use {
        for (edge in it) {
            emit(edge)
        }
    }
}

1

2

Flow<T>

Flow<T>

val job = launch {
    val flow = slowBlockingCallThatReturnsIterator().asFlow()
    flow.collect { 
        yield() // allows for cancel() to work
        channel.send(it) 
    }
}

job.cancel()

yield()

GlobalScope

GlobalScope

GlobalScope

supervisorScope

PublishSubject

Channel(RENDEZVOUS)

consumeAsFlow

If the flow consumer fails with an exception, channel is cancelled.

callbackFlow

@ExperimentalCoroutinesApi
fun Query.getQuerySnapshotFlow(): Flow<QuerySnapshot?> = callbackFlow {
    val listenerRegistration = addSnapshotListener { querySnapshot, e ->
        if (e != null) {
            cancel(message = "", cause = e)
            return@addSnapshotListener
        }
        offer(querySnapshot)
    }
    awaitClose {
        listenerRegistration.remove()
    }
}

@ExperimentalCoroutinesApi
fun <T> Query.getDataFlow(mapper: (QuerySnapshot?) -> T): Flow<T> {
    return getQuerySnapshotFlow().map { mapper(it) }
}

Flow

else

switchMap

val myAppointments = _appointmentDate.switchMap { date ->
	//HERE I WANT TO CANCEL PREVIOUS FLOW STARTED IN ELSE CASE BELOW
	liveData {
		if (date == null)
			emit(AppointmentState.NoAppointment)
		else {
			myCollection.whereEqualTo(Appointment.DATE, date)
					.getDataFlow { it!!.toObjects<List<Appointment>>() }
					.collect { emit(AppointmentState.Loaded(it)) }
		}
	}
}