Repository: Kotlin/kotlinx.coroutines
File: kotlinx-coroutines-core/common/src/flow/operators/Share.kt (Kotlin)

Code before:
 * @throws NoSuchElementException if the upstream flow does not emit any value.
 */
public suspend fun <T> Flow<T>.stateIn(scope: CoroutineScope): StateFlow<T> {
    val config = configureSharing(1)
    val result = CompletableDeferred<StateFlow<T>>(scope.coroutineContext[Job])
    scope.launchSharingDeferred(config.context, config.upstream, result)
    return result.await()
}

private fun <T> CoroutineScope.launchSharingDeferred(
    context: CoroutineContext,
    upstream: Flow<T>,
    result: CompletableDeferred<StateFlow<T>>
) {
    launch(context) {
        try {
            var state: MutableStateFlow<T>? = null
            upstream.collect { value ->
                state?.let { it.value = value } ?: run {
                    state = MutableStateFlow(value).also {
                        result.complete(ReadonlyStateFlow(it, coroutineContext.job))
                    }
                }
            }
            if (state == null) {
                result.completeExceptionally(NoSuchElementException("Flow is empty"))
            }
        } catch (e: Throwable) {
            // Notify the waiter that the flow has failed
            result.completeExceptionally(e)
            // But still cancel the scope where state was (not) produced
            throw e
        }
    }
}

// -------------------------------- asSharedFlow/asStateFlow --------------------------------

/**
 * Represents this mutable shared flow as a read-only shared flow.
 */
public fun <T> MutableSharedFlow<T>.asSharedFlow(): SharedFlow<T> =
    ReadonlySharedFlow(this, null)

/**
 * Represents this mutable state flow as a read-only state flow.
 */
public fun <T> MutableStateFlow<T>.asStateFlow(): StateFlow<T> =
    ReadonlyStateFlow(this, null)

@OptIn(ExperimentalForInheritanceCoroutinesApi::class)

Review comment (bug):
Notice: I think we shouldn't cancel the collecting scope here, as there is nothing wrong with the flow collection itself, the exception should be limited to `stateIn` IMHO.

Code after:
 * @throws NoSuchElementException if the upstream flow does not emit any value.
 */
public suspend fun <T> Flow<T>.stateIn(scope: CoroutineScope): StateFlow<T> {
    val config = configureSharing(1)
    val result = CompletableDeferred<Result<StateFlow<T>>>(scope.coroutineContext[Job])
    scope.launchSharingDeferred(config.context, config.upstream, result)
    return result.await().getOrThrow()
}

private fun <T> CoroutineScope.launchSharingDeferred(
    context: CoroutineContext,
    upstream: Flow<T>,
    result: CompletableDeferred<Result<StateFlow<T>>>,
) {
    launch(context) {
        try {
            var state: MutableStateFlow<T>? = null
            upstream.collect { value ->
                state?.let { it.value = value } ?: run {
                    state = MutableStateFlow(value).also {
                        result.complete(Result.success(ReadonlyStateFlow(it, coroutineContext.job)))
                    }
                }
            }
            if (state == null) {
                result.complete(Result.failure(NoSuchElementException("Flow is empty")))
            }
        } catch (e: Throwable) {
            // Notify the waiter that the flow has failed
            result.completeExceptionally(e)
            // But still cancel the scope where state was (not) produced
            throw e
        }
    }
}

// -------------------------------- asSharedFlow/asStateFlow --------------------------------

/**
 * Represents this mutable shared flow as a read-only shared flow.
 */
public fun <T> MutableSharedFlow<T>.asSharedFlow(): SharedFlow<T> =
    ReadonlySharedFlow(this, null)

/**
 * Represents this mutable state flow as a read-only state flow.
 */
public fun <T> MutableStateFlow<T>.asStateFlow(): StateFlow<T> =
    ReadonlyStateFlow(this, null)

@OptIn(ExperimentalForInheritanceCoroutinesApi::class)