switchMap operator for Kotlin's Flow type.

SwitchMap.kt

import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.Job
import kotlinx.coroutines.cancelAndJoin
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.launch

/**
 * switchMap operator. Like RxJava's switchMap operator it cancels any ongoing flow and starts the new one on every new
 * value that gets emitted from upstream.
 */
@FlowPreview
public fun <T, R> Flow<T>.switchMap(
    mapper: suspend (value: T) -> Flow<R>
): Flow<R> {
    return flow {
        coroutineScope {
            var currentJob: Job? = null
            collect { outerValue ->
                val inner = mapper(outerValue)
                currentJob?.cancelAndJoin()
                currentJob = launch {
                    inner.collect { value ->
                        emit(value)
                    }
                }
            }
        }
    }
}