Resettable lazy version of a shared Flow, which allows for multiple observers to consume a single emitter, with an automatic reset after the last consumer stops and an automatic resume when another consumer starts.

SharedFlow.kt

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock

fun <T> Flow<T>.shareIn(
        scope: CoroutineScope
): Flow<T> = SharedFlow(
        BroadcastManager(
                this,
                scope
        )
)

internal class SharedFlow<T>(
        private val broadcastManager: BroadcastManager<T>
) : Flow<T> {

    internal var count = 0

    override suspend fun collect(
            collector: FlowCollector<T>
    ) = collector.emitAll(createFlow())

    internal suspend fun createFlow(): Flow<T> = broadcastManager.getChannel()
            .asFlow()
            .onStart { broadcastManager.onNewConsumer() }
            .onCompletion { broadcastManager.onConsumeEnd() }

}

internal class BroadcastManager<T>(
        private val sourceFlow: Flow<T>,
        private val scope: CoroutineScope
) {

    private var count = 0

    private val mutex = Mutex(false)

    private var channelRef: BroadcastChannel<T> = createChannel()

    private fun createChannel(): BroadcastChannel<T> = sourceFlow
            .broadcastIn(scope)

    private fun reset() {
        channelRef = createChannel()
    }

    suspend fun onNewConsumer() = mutex.withLock { count++ }
    suspend fun onConsumeEnd() = mutex.withLock { if (--count == 0) reset() }

    suspend fun getChannel(): BroadcastChannel<T> = mutex.withLock { channelRef }

}