Is there a nice way to trigger a restart of a flow...

# flow

[Stylianos Gakis]

Is there a nice way to trigger a restart of a flow that is turned into a StateFlow using statein? I’ll show you my use case and try to explain what I mean in the thread 🧵

[Stylianos Gakis]

I have some code that looks like this:

val viewState: StateFlow<ViewState> = flow {
    getDataUseCase.invoke().fold(
        ifLeft = { emit(ViewState.Error) },
        ifRight = { emit(ViewState.Content(it)) },
    )
}.stateIn(
    viewModelScope,
    SharingStarted.WhileSubscribed(5000),
    ClaimDetailViewState.Loading
)

And I want to on demand (user interaction), be able to restart the flow {} part.

p.s: I know that this one could be written as something like this

private val _viewState: MutableStateFlow<ViewState> = MutableStateFlow(ViewState.Loading)
val viewState: StateFlow<ViewState> = _viewState.asStateFlow()

init {
    loadContent()
}

fun retry() {
    loadContent()
}

private fun loadContent() {
    viewModelScope.launch {
        val result = getDataUseCase.invoke().fold(
            ifLeft = { ViewState.Error },
            ifRight = { ViewState.Content(it) },
        )
        _viewState.value = result
    }
}

But this is just a simplified example. In other cases I have a mix of suspend functions like this invoke on the useCase and functions returning Flow that I want to use to transform the results of these into a StateFlow

[wasyl]

Just a quick idea, you could have a restartTrigger.flatMap { yourFlow() } and emit from the trigger somehow?

[Stylianos Gakis]

Thanks for the suggestion, playing around with this a bit, I am now looking at this:

private val retryChannel = Channel<Unit>()
val viewState: StateFlow<ClaimDetailViewState> = retryChannel.receiveAsFlow().transformLatest { // same code }

init {
    viewModelScope.launch {
        retryChannel.send(Unit)
    }
}

fun retry() {
    viewModelScope.launch {
        retryChannel.send(Unit)
    }
    // Or maybe 
    retryChannel.trySend(Unit)
    // But I think the suspending `send` is a better idea, even if someone spams the retry() function, the transformLatest should just take in the last one anyway.
}

But it’s a bit weird, since we now have an extra channel in the class, which might not be 100% obvious to a dev seeing it for the first time, plus the need to add an element in the init function 🤔

Do you think we could have this restartTrigger be something a bit nicer? I thought of it being a Channel<Unit> but maybe it shouldn’t? Or did you mean something a bit different?

[Stylianos Gakis]

This is honestly working fine as I’m testing it now, but I really wish there was a way to have this channel start with an initial item in it so that I don’t have to do this init thing?
Or not use a Channel altogether if possible, but I can’t think of any other way to do this atm 🤔

[Stylianos Gakis]

Experimenting with it a bit more I’ve got this:

class RetryChannel private constructor(channel: Channel<Unit>) : Channel<Unit> by channel {
    init {
        channel.trySendBlocking(Unit)
    }
    companion object {
        operator fun invoke() = RetryChannel(Channel(Channel.CONFLATED))
    }
}

Which makes it as simple as this to use on the call site:

private val retryChannel = RetryChannel()
val viewState: StateFlow<ViewState> = retryChannel.receiveAsFlow().transformLatest { // same code as before }

But now I am not sure about the trySendBlocking and if I am digging a hole for myself instead of finding a better all-around approach 😅

[wasyl]

Yep I meant Channel<Unit> or something along that, but I can’t play around with it right now to suggest something with a default value and I don’t remember what could be used instead

[Stylianos Gakis]

Awesome, thank you so much for the help!
It does work at the moment with the limited testing I’ve done, so I’ll leave it like this for now and see if it fits my needs as I wanted it to. If you (or someone else) do at some point come up with something nicer I would really appreciate if you shared it with me though!

[ephemient]

instead of setting it up in init, you can add the initial element at the consumer:

retryChannel.receiveAsFlow()
    .onStart { emit(Unit) }
    .flatMap { // ... }

[Stylianos Gakis]

Nice, that sounds better. This does put the burden on the call-site though, which I could negate by adding it to be done automatically inside my RetryChannel class.

But I wonder, this is the second time I see the flatMap being mentioned on the flow, but as I see it in my IDE, it is deprecated. Is it a simple mis-spell, or am I missing something here?

[ephemient]

ah you probably want .flatMapLatest() for this use case. it's deprecated to make it clearer what behavior you want (compare to .flatMapConcat() and .flatMapMerge())

[Stylianos Gakis]

Made it work like this actually:

class RetryChannel private constructor(private val channel: Channel<Unit>) {
    suspend fun retry() { channel.send(Unit) }

    fun <R> transformLatest(@BuilderInference transform: suspend FlowCollector<R>.(value: Unit) -> Unit): Flow<R> {
        return channel
            .receiveAsFlow()
            .onStart { emit(Unit) }
            .transformLatest(transform)
    }

    companion object {
        operator fun invoke() = RetryChannel(Channel(Channel.CONFLATED))
    }
}

And it’s quite straight-forward to use on the call-site.
I am not quite sure what I am missing and how flatMap would help here 🤔
Call-site code now looks like this and it seems to suit my needs now:

private val retryChannel = RetryChannel()
val viewState: StateFlow<ViewState> = retryChannel.transformLatest {
    someUseCase.invoke(claimId).fold(
        ifLeft = { emit(ViewState.Error) },
        ifRight = { emit(ViewState.Content(it)) },
    )
}.stateIn(
    viewModelScope,
    SharingStarted.WhileSubscribed(5000),
    ViewState.Loading
)

fun retry() { viewModelScope.launch { retryChannel.retry() } }

[Stylianos Gakis]

Side note: I don’t really like how I am naming it with the “Channel” suffix now even though it’s not exposed as a Channel 🤔 But I don’t think it needed to implement that interface anyway, all I need is this distinct functionality of it hmm 🤔

[Stylianos Gakis]

Now I see, I can just add this as a function too, if I want the transform block to return a flow instead of having the FlowCollector as a receiver of the lambda to build a flow manually

fun <R> flatMapLatest(@BuilderInference transform: suspend (value: Unit) -> Flow<R>): Flow<R> {
    return channel
        .receiveAsFlow()
        .onStart { emit(Unit) }
        .flatMapLatest(transform)
}