Closed fluidsonic closed 4 years ago
What is your use-case? What kind of application scenario calls for this kind of behavior?
At the moment I would be beneficial in the sign-in and sign-out processes of an Android app.
AuthenticationManager
performs sign-in/sign-out operations (verify credentials with API, save token, etc.).UserSignedIn
/ UserSignedOut
) to subscribers. They now have a chance for set-up work (e.g. fetch & save user object to DB on sign-in) or clean-up work (e.g. wipe all user data in DB, memory & disk on sign-out) before the AuthenticationManager
completes the process and gives control back to the UI.If I write that without subscriptions, the AuthenticationManager
(or an intermediate component) would have to know all components that need to perform some work synchronously upon sign-in/sign-out.
If I use a synchronous Flow, synchronous event bus or similar then the AuthenticationManager
just needs to know that there are some components that need to do some related work and waits for them to finish.
I've heard that a Flow.share()
operator is planned that I think covers such a use case already?
I'm not sure that Flow
(which is cold and reactive) is a good fit for this kind of use-case to start with. Have you tried just doing a regular suspending callbacks? E.g. define the property in your AuthenticationManager
:
val authProcessors = ArrayList<suspend (AuthEvent) -> Unit>()
Here, I suppose that AuthEvent
is a super class for UserSignedIn
and UserSignedOut
. Now, in your subscribers that should do some required processing of those events, write:
authManager.authProcessors += { event -> ... process event }
In the code of the AuthenticationManager
itself, when you have an authentication event and need to wait until every component has processed it, you just write:
authProcessors.forEach { it(event) }
This looks pretty straigtforward and simple to me. Would it work for you?
That for sure would work.
But I'd have to build things on top which I have for free in Flow
, like unsubscribing when the CoroutineScope
is closed for example.
I just wonder why I can use Flow
to
but not
It just feels weird that I have to make an exception here and use Flow
for all the other cases.
Maybe Flows
aren't a good idea for events in general?
Or the synchronous single-consumer case isn't something I should rely on?
Flow for events is a complicated story, indeed. There was a big discussion in #1082 and one outcome for that is that we might introduce some kind of EventFlow
that would exactly cover your use-case of flowing to many consumers synchronously.
Here is a design for SharedFlow
that covers this use-case for suspending broadcast flow emission. See #2034
It looks like it is now safe to close this issue, as its use-cases are fully accounted for by #2034.
I propose to add a
Flow
operation that broadcasts upstream values downstream without involving aBroadcastChannel
and thus suspending the emitter until the broadcast for a value is complete.Status quo on suspending emitters
emit()
ing a value to aFlow
suspends the emitter until the collector has returned from the collection lambda for that value.Flow
values using.broadcastIn(scope).asFlow()
the emitter is no longer suspended until the collector(s) have returned from the collection lambda for that value.emit()
a value to multiple collectors where the emitter remains suspended until all collectors have processed that value.Example output
(1) Emitting two values [1, 2] to one collector looks like this:
(2) Emitting two values [1, 2] to multiple collectors through a
BroadcastChannel
looks like this:(3) The desired behavior is that two values [1, 2] emitted to multiple collectors looks like this:
Use case
My use case for (3) are events where the event emitter has to wait for all event subscribers to fully process the event.
Example
AuthenticationManager
has aFlow<WipeDataEvent>
.WipeDataEvent
is emitted by theAuthenticationManager
all subscribers must wipe the data they have stored.AuthenticationManager
can continue with its work, e.g. the sign-out/sign-in process..broadcast()
example for (3)The following code is very naive. It merely illustrates how the behavior in the example output for (3) can be achieved.
.broadcast()
..broadcastIn(scope).asFlow()
but not involve a channel.A fully working example which creates all of the output mentioned above can be found here: https://gist.github.com/fluidsonic/f7b2b0084f184932ea3be4cf0074496a