I have this function `suspend fun foo(progressLambda: (Long) -> Unit)` and I want to convert this...

# coroutines

[Mark]

I have this function suspend fun foo(progressLambda: (Long) -> Unit) and I want to convert this to a function returning a StateFlow<State> where State is a sealed class providing Progress , Success etc. Initially I thought of using callbackFlow but that seems to be more for functions that run asynchronously. Is there a more appropriate way?

[gildor]

Looks that you need standard flow { } builder

[Mark]

But I cannot emit from within the lambda because it is not a suspend lambda.

[Mark]

Another option (if not using callbackFlow ) would be something like:

val progressFlow = MutableStateFlow(State.Progress(0L))
return flow {
    emit(null)
    foo { progressValue ->
        progressFlow.value = State.Progress(progressValue)
    }
    emit(State.Success)
}.combine(progressFlow) { invokeState, progressState ->
    invokeState ?: progressState
}

[Mark]

Alternatively, without using `combine`:

flow {
    val progressFlow = MutableStateFlow(State.Progress(0L)
    coroutineScope { 
        val progressJob = launch { 
            progressFlow.collect { progress -> 
                emit(progress)
            }
        }
        foo { progressValue ->
            progressFlow.value = State.Progress(progressValue)
        }
        progressJob.cancel()
        emit(State.Success)
    }
}

[Mark]

I think I prefer the combine approach, because I am more confident there will never be a progress state emitted after the success state.

[gildor]

if you wrap some callback, indeed you need callbackFlow

[Mark]

Are you saying the other two solutions above, don’t work?

[gildor]

no they will work

[gildor]

but looks overcomplicated for me

[gildor]

callbackFlow is probably way to go

[Mark]

callbackFlow {
    send(State.Progress(0L))
    try {
        foo { progressValue ->
            offer(State.Progress(progressValue))
        }
        send(State.Success)
        close()
    } catch (e: Exception) {
        close(e)
    }
}.conflate() // we only care about the latest value

[Mark]

Or, if there is a Failure state:

callbackFlow {
    send(State.Progress(0L))
    try {
        foo { progressValue ->
            offer(State.Progress(progressValue))
        }
        send(State.Success)
    } catch (e: Exception) {
        send(State.Failure(e.message))
    }
    // we need to close() or awaitClose(), but nothing to wait for
    close()
}.conflate() // we only care about the latest value

[Mark]

But is there any guarantee that the progress state cannot be emitted after the Success/Failure state?

[gildor]

there is no such guarantee, after all it purely related on foo implementation. if it stop send progress before return. then you are fine,. but if you not sure, implementation is not under your control, you always can just use something like AtomicBoolean as flag and prevent send progress before sending Success

[gildor]

I’m a bit worry about this implementation though, it looks that foo is blocking function, so you probably want also add .flowOn(IO/Computation)

[Mark]

foo does not launch another coroutine so I think it’s impossible for the lambda to be called/completed after foo returns. So offer can only be called before foo returns, and so the question comes down to: if offer is called before send (potentially using a different dispatcher) is the order of emitted values guaranteed to be the same?

[gildor]

yes, it guaranteed

[gildor]

also you always can use sendBlocking

[gildor]

this function anyway blocking

[gildor]

also allows you to avoid missing progress (thought offer looks fine for progress)

[Mark]

That’s what I like about the combine solution. You get a clear priority of the ‘invoke’ flow over the ‘progress’ flow. invokeState ?: progressState

val progressFlow = MutableStateFlow(State.Progress(0L))
val invokeFlow = flow {
    emit(null)
    try {
        foo { progressValue ->
            progressFlow.value = State.Progress(progressValue)
        }
        emit(State.Success)
    } catch (e: Exception) {
        emit(State.Failure(e.message))
    }
}
return invokeFlow.combine(progressFlow) { invokeState, progressState ->
    invokeState ?: progressState
}

[gildor]

I don’t know callbackFlow looks fine too, also has priority

[Mark]

Another benefit would be that StateFlow will swallow duplicates. Also using separate progress flow, allows us to do things like debounce the progress.

[gildor]

also true