I've updated to 1.6.0-native-mt and on iOS it has ...

# coroutines

[Trevor Stone]

I've updated to 1.6.0-native-mt and on iOS it has broken a lot of my test cases, I found https://youtrack.jetbrains.com/issue/KTOR-3612 but I am also having issues for tests that cover code that calls stateIn. I am currently doing

.stateIn(
    CoroutineScope(Dispatchers.Default),
    SharingStarted.WhileSubscribed(),
    null
)

and getting Cannot start an undispatched coroutine in another thread DefaultDispatcher from current MainThread as an error from runTest

[Paul Woitaschek]

But does it work on iOS? I’m seeing that issue in the iOS application, not only in tests

[Paul Woitaschek]

@Trevor Stone

[Trevor Stone]

I do get this crash at runtime as well

[Trevor Stone]

I removed stateIn from my app and it seems to work ok (except for the things that breaks)

[Trevor Stone]

but anywhere I do .stateIn on a flow it breaks

[Trevor Stone]

Ok, it actually looks like the issue isn't shareIn it is using SharingStarted.WhileSubscribed()

[Trevor Stone]

if I change stateIn to use SharingStarted.Eagerly it no longer crashes

[Paul Woitaschek]

Hm that one is backing a core function of our shared lib, :/

[Trevor Stone]

Seems like Lazily is broken too. Only Eagerly seems to work for me

[Trevor Stone]

So.....I posted a message to the main channel with some more context, but I did find a workaround to seemingly make it function like it used to

[Trevor Stone]

You could either make sure you only call stateIn/shareIn in Dispatchers.Default, like

runBlocking {
    withContext(Dispatchers.Default) {
        parentFlow().stateIn(GlobalScope, SharingStarted.WhileSubscribed(), null)
    }
}

[Trevor Stone]

or

[Trevor Stone]

you can make a local implementation of SharingStarted with a custom equals function

[Trevor Stone]

private class StartedWhileSubscribed(
    private val stopTimeout: Long = 0,
    private val replayExpiration: Long = Long.MAX_VALUE
) : SharingStarted {
    init {
        require(stopTimeout >= 0) { "stopTimeout($stopTimeout ms) cannot be negative" }
        require(replayExpiration >= 0) {
            "replayExpiration($replayExpiration ms) cannot be negative"
        }
    }

    override fun command(subscriptionCount: StateFlow<Int>): Flow<SharingCommand> = subscriptionCount
        .transformLatest { count ->
            if (count > 0) {
                emit(SharingCommand.START)
            } else {
                delay(stopTimeout)
                if (replayExpiration > 0) {
                    emit(SharingCommand.STOP)
                    delay(replayExpiration)
                }
                emit(SharingCommand.STOP_AND_RESET_REPLAY_CACHE)
            }
        }
        .dropWhile { it != SharingCommand.START } // don't emit any STOP/RESET_BUFFER to start with, only START
        .distinctUntilChanged() // just in case somebody forgets it, don't leak our multiple sending of START


    @OptIn(ExperimentalStdlibApi::class)
    override fun toString(): String {
        val params =
            buildList(2) {
                if (stopTimeout > 0) add("stopTimeout=${stopTimeout}ms")
                if (replayExpiration < Long.MAX_VALUE) add("replayExpiration=${replayExpiration}ms")
            }
        return "SharingStarted.WhileSubscribed(${params.joinToString()})"
    }

    // equals & hashcode to facilitate testing, not documented in public contract
    override fun equals(other: Any?): Boolean =
        (other == SharingStarted.Eagerly) ||
            (other is StartedWhileSubscribed &&
                stopTimeout == other.stopTimeout &&
                replayExpiration == other.replayExpiration)

    override fun hashCode(): Int = stopTimeout.hashCode() * 31 + replayExpiration.hashCode()
}

[Trevor Stone]

so it is equal to Eagerly and then uses the same CoroutineStart.DEFAULT it used to

[Trevor Stone]

this is a hack and might cause other issues, but it seems to work at first glance