Is there a nice way to convert replaying Flow into...

# coroutines

[dimsuz]

Is there a nice way to convert replaying Flow into a non replaying? A naive implementation using drop(1) makes subscriber loose first emitted item:

val flow = MutableSharedFlow<Int>(replay = 1)
  fun get(replayLast: Boolean): Flow<Int> = flow.drop(1) // naive impl

  launch { get(replayLast = false).collect { println(it} } } // doesn't receive 1
  launch { delay(1000); flow.emit(1) }

Is having two separate flows: one with replay and one witout - the only option?

[bezrukov]

you can probably use flow.replayCache for this, but most likely it won't 100% robust

fun get(replayLast: Boolean): Flow<Int> = flow { 
   flow.drop(flow.replayCache?.size ?: 0).collect {
      emit(it)
   }
}

The problem is that it's unclear what to do with items emitted between flow.replayCache and .collect calls

[dimsuz]

Nice, thank you!

I can do without collect even:

fun get(replayLast) = flow.drop(if (replayLast) 0 else flow.replayCache.size))

[bezrukov]

you're returning a cold flow (what if a caller will start collecting it after a few more emits?) so you have to wrap it to flow builder

[dimsuz]

yep, cold flow is my intention here, so it's ok.

[bezrukov]

No, I mean replayCache.size may change between calling get function and collect:

// Nothing were emitted to `flow` yet
val cold = get(replayLatest = false) // you made a cold flow with drop(0)
flow.emit(1)
cold.collect { } // you will receive 1

[dimsuz]

ah, oh, you right, thanks!