How can I get the latest emission upon an error and continue - modify that? I have sth like a `Flow&...

# coroutines

[Paul Woitaschek]

How can I get the latest emission upon an error and continue - modify that?
I have sth like a Flow<ViewState>. This ViewState also has a boolean property if an error should be shown.
Upon error I want to emit the latest view state but call copy(showError = true) and complete the flow so that a consumer can use sth like:
.repeatWhen { repeatClickedFlow() }

[streetsofboston]

How does your `Flow<ViewState>`’s code look like without the error handling? (ie the code that creates the Flow<ViewState> instance)

[Paul Woitaschek]

@streetsofboston https://gist.github.com/PaulWoitaschek/6220f24ea69172350ac03dbaff1b91b0

[streetsofboston]

Thanks.

Can you restart/re-get the contentFlow when the first one produces an error/exception?

[Paul Woitaschek]

Yes

[streetsofboston]

Then wrap the emitAll inside a try-catch clause.
In the catch, get the contentFlow again (i would change the type of contentFlow from Flow<...> into () -> Flow<...>, and use contentFlow() to actual obtain the flow.

[Paul Woitaschek]

But how would I be able to get the latest state and flip the error flag?

[streetsofboston]

And for this snippet, add a state-full variable.

...
    return flow {
      var latestState = baseState
      emit(latestState)
      while (true) {
          val sourceFlow = contentFlow()
          try {
              sourceFlow.collect {
                  latestState = it
                  emit(latesState)
              }
          } catch (t: Throwable) {
              latestState = latestState.copy(....)
              emit(latestState)
          }
      }
    }

Something like this….

[streetsofboston]

With using contentFlow(), I assume you would need to create/get a brand-new one. If not, if just starting to collect it again is enough, just use contentFlow.collect instead of the `sourceFlow.connect`…

[streetsofboston]

Another way, but that is not yet possible, is to contentFlow.share(1) and use this shared-flow to collect and use it in the onErrorResumeNext(...) as well. But Flow<T>.share(n) does not yet exist…..

[Paul Woitaschek]

I tried it in isolation. Does it throw a backend error for you too? @streetsofboston https://gist.github.com/PaulWoitaschek/f70dd8a8dbb2807bc1da84e8d2dd270c

[streetsofboston]

What is a backend error?

[Paul Woitaschek]

https://gist.github.com/PaulWoitaschek/3164e15d091b0fd0704c77530a069ddc

[streetsofboston]

I don’t think I have all your code…

[Paul Woitaschek]

What's missing?

[streetsofboston]

E.g. I have no ‘main()’ function…

[Paul Woitaschek]

Just compile it

[streetsofboston]

Ah.. sorry…. let met try 🙂

[streetsofboston]

yup, at my end as well, i get the compiler crash

[Paul Woitaschek]

It's because of the recursion of the inner function calling itself

[streetsofboston]

Yup, removing that removes the crash

[streetsofboston]

With refactoring, the crash goes away:

...
fun viewState(repeat: Flow<Unit>): Flow<ViewState> {
    val initialState = ViewState(content = null, loading = true, showError = false)
    return flow {
        emit(initialState)
        var lastState = initialState

        suspend fun updateAndEmit(update: ViewState.() -> ViewState) {
            lastState = update(lastState)
            emit(lastState)
        }

        repeat.startContentFlow(::updateAndEmit)
    }
}

suspend fun <T> Flow<T>.startContentFlow(updateAndEmit: suspend (ViewState.() -> ViewState) ->Unit) {
    try {
        contentFlow().collect {
            updateAndEmit { copy(content = it, loading = false) }
        }
    } catch (t: IOException) {
        updateAndEmit { copy(loading = false, showError = true) }
        take(1).collect {
            updateAndEmit { copy(loading = true, showError = false) }
            startContentFlow(updateAndEmit)
        }
    }
}
...

[streetsofboston]

But this is definitely a compiler bug. You may want to report this one to JetBrains.

[Paul Woitaschek]

Ah nice, thanks!
https://gist.github.com/PaulWoitaschek/7e6318614307556fae6770556f83c570
It actually works. However the code isn't the simplest, can you make it simpler?

[streetsofboston]

Not so much… they haven’t yet implemented the share or cache methods on Flows… that is coming though

[streetsofboston]

you can make it more iterative instead of recursive. The call to ‘startContentFlow’ is not tail-recursive… it can be prone to stack-overflow issues.

[Paul Woitaschek]

How would I do that?

[streetsofboston]

I think that take(1) is a Flow that terminates after emitting 1 item. Try to just move the startContentFlow(updateAndEmit) call outside of that collect { .. } block

[streetsofboston]

If that works, you can remove that startContentFlow(updateAndEmit) call and just wrap the body of startContentFlow inside a while(true) { ... } loop

[Paul Woitaschek]

https://youtrack.jetbrains.com/issue/KT-32734