I'm having an issue where a flow isn't being collected and I can't for the life of me figure out why basic sealed class hierarchy:

sealed class Update {
    object Success : Update()
    data class Progress(val message: String) : Update()
    data class Failure(val throwable: Throwable) : Update()
}

Success

and

Failure

emissions are collected, but

Progress

ones are not applying only two operators:

.filterNotNull()
.transformWhile { update ->
   emit(update)
   update !is Update.Failure && update !is Update.Success
}

I've debugged and logged everywhere I could, and the

Progress

objects are definitely emitted by the flow, but they never make it to the

.collect

. Am I missing something really obvious here?

Not sure what you intended to do with the last line in your transform block, as it doesn't do anything. The

transform

operator only returns

Unit

.

transformWhile

returns a boolean. I want the flow to complete after a failure or success is emitted, and what I have is more or less exactly like the example in the documentation for

transformWhile

.

My mistake, I thought that was just

transform

. Reading the doc:

Applies transform function to each value of the given flow while this function returns

true

.

This says that it applies the transforms function (emits your value) so long as you return true. In your case, you are returning

false

when the

Update

object is of type

Progress

, which is why it doesn't emit that. Also, you should consider using the

onComplete

and

catch

to emit your success or failures.

I'm not sure that's correct either. If update is

Update.Progress

then

update !is Update.Failure

is true and

update !is Update.Success

is also true, so it should continue emitting. Also, the

emit

call is made before it returns, which should guarantee at least one

Progress

gets emitted anyhow.

Is it possible you aren't emitting any Progress items? Add in an

onEach

between the filter and transform operators, and print out any Progress objects. Maybe even print your boolean logic there as well.

I've tried adding an

onEach

and I also debugged to make sure that I was emitting the

Progress

items and that they were being updated in the

StateFlow

, but still no luck

So

Progress

is not emitting with both operators removed?

Yep 😕

The only thing I can think of is that

StateFlow

is suppressing emits of

Progress

because they are the same when compared? Is the message the same for

Progress

on every call to

getUpdateStatus()

?

Some are, but in my tests there have always been at least two different values in the

message

field. I also tried adding an

Instant

that defaulted to

Instant.now()

but that didn't help either.

The

doUpdate()

doesn't need to be a suspend function. You can put

startUpdates()

inside an

onStart {}

operator, and just return the flow, that way you don't start updates prematurely before you start collecting.

Huh that's pretty clever, I had no idea you could do that. Unfortunately it also doesn't seem to work, even though my logging says that it is triggering the

startUpdates()

call...

I see the problem. Going through each broken case to explain: Broken1

suspend fun doLoop() {
    for (i in 0..5) {
        flow.emit(Status.Progress("Message #$i"))
        println("Emitted: Message #$i")
        delay(1000)
    }
    flow.emit(Status.Finish)
    println("Emitted: Finish")
}

fun exampleUseCase(): Flow<Status> {
    return flow
        .onStart {
            coroutineScope { launch { doLoop() } }
        }
        .transformWhile {
            emit(it)
            it !is Status.Finish
        }
}

In your

start

block, you are creating a new

coroutineScope

block to launch a new job. However, that

coroutineScope

block will suspend until all tasks in it complete, which means it'll iterate through all the

Progress

events, emitting them, but because

transformWhile

doesn't start until

onStart

finishes, you've consumed all events, and only get the last one.

Ohhhh, I was unaware of that behaviour for

coroutineScope

. Thank you for explaining. Do you know if there's any way to accomplish the desired behaviour without making the function extend

CoroutineScope

or otherwise having to pass the coroutine scope in as a parameter?