class Presenter {
   init {
      presenterScope.launch {
         syncer.syncingFinishedEvent
	   .collect {
	      navigateSomewhere()
	   }
      }
      syncer.sync()
   }
}

class Syncer {
   fun sync() {
      syncerScope.launch {
         ...
         syncFinishedEvent.emit(Unit)
      }
   }
}

Is there a way to only call

syncer.sync()

after

syncer.syncingFinishedEvent

is for sure subscribed to? I cannot miss a event

You can use

start = Undispatched

in the first launch, in this case you will subscribe to the flow in the same call frame (but in this case it will be executed on the presenter's constructor thread)

Trying to manipulate internal behavior to get desired behavior is fragile at best. A

flowOn

could easily break this approach. Just because you subscribed to a

Flow

does not mean that the upstream source

Flow

has been subscribed. For this particular scenario, I don't even see the need for any

Flow

. If you want to do something and wait for it to finish, just make it a suspend method.

suspend fun sync()

would make it a lot simpler. If you really need/want it to run in it's own scope, then return the Job so you can

join

it.

Well, thats the shape of my api, many callsites can call it so syncs run on own scope, ui just sends requests to it. I dont want to expose suspend fun as then I'd need a mutex, only to return right away after resuming after contention, and then thats needs a flow of results anyways, as I dont want to rerun the sync for the 2nd mutex contender etc Also, exposing a Job.. someone might cancel it, I'm not a huge fan of that What I'd best like here is

doOnSubscribe

from rxJava, but not sure how to implement it as line after

collect,

when creating custom operator, only runs after the flow completes 😔 To be honest I find it silly that I cannot guarantee im collecting a flow, when the same component is initiating work which writes to the flow, so logically, it is present .. why are the hacks needed.. it just worked with rxjava

Seems like you may benefit from something like:

val syncJob = syncerScope.launch(start = LAZY) { ... }
suspend fun sync() = syncJob.join()

To be honest it was state, but the thing why im dealing with this is because sometimes I got backpressure in the collector, so it skipped the SYNCING state, then SYNCED got emitted, but lets say SYNCED was default, so its last value the collector seen, so the last SYNCED wont get delivered because of the

distinctUntilChanged

behavior of StateFlow Anyways,

SharedFlow.onSubscription

looks like what I want, but it feels odd to expose flow subclass as public api, or its just me?

it feels odd to expose flow subclass as public api, or its just me?

Yes it's weird, imo, for this use case. In general it's not weird to expose SharedFlow or StateFlow. If you describe a bit more of your use case, I'm guessing a better solution can be found.

I have a Syncer which has its own scope so I can throttle/serialize/return requests to sync from many call sites, and sync should not be canceled when call site dies. I expose

Sycner.syncInProgress: Flow<Boolean>

Now this usecase came where I need to navigate to some blocking ui, trigger a sync, wait for it to finish, then navigate to next screen. My solution was, as it was driven by StateFlow, to drop(1) to get ride of the cached value, and to request the sync. Given its dropped 1, I should only receive "live" values, wait for

filter { it == false

}`` as my signal to navigate to next screen. What however sometimes happens is that, if sync is fast, and collector (main thread drvien) is busy, it skips the

true

emit. And the when state changes back to false, its equal to the initial cached emit in eyes of the collector, so a no-emit. And now my ui is stuck.

I'm reading that as "I want to do something and know when it's done". That's the use case that a suspend method is for. You can do the work on

syncerScope

, and then `join`/`await` it.