Is there an easy way convert a `SharedFlow` only to emit to

Question

Is there an easy way convert a `SharedFlow` only to emit to its first subscriber

Michael Marshall · Answer

What s the purpose behind doing this

Arun Joseph · Answer

The flow is sort of a side effect flow taking data from one repository and deposits in another repository the subscribers are UI elements and keeps this data transfer alive and emitting I use `callbackFlow` and `shareIn` to keep the listeners registered as long as there are subscribers But only would like to have only one subscriber collects at a time

Michael Marshall · Answer

Perhaps sharing some of the code would help slightly smiling face

Arun Joseph · Answer

```val locationFlow Flow lt Either lt GeoLocationEntity LocationFailure gt gt = callbackFlow lt Either lt GeoLocationEntity LocationFailure gt gt val onLocationUpdateListener = OnLocationUpdateListener location gt trySend location toGeoLocationEntity toOk locationEngine addOnLocationUpdateListener onLocationUpdateListener locationEngine enable awaitClose locationEngine removeOnLocationUpdateListener onLocationUpdateListener locationEngine disable shareIn shareInScope replay = 0 started = SharingStarted WhileSubscribed replayExpirationMillis = 0 private fun keepEngineRunning locationFlow Flow lt Either lt GeoLocationEntity LocationFailure gt gt Flow lt Either lt Unit Failure gt gt = locationFlow map location gt location andThen geoLocationEntity gt engine updateLocation geoLocationEntity fun UI element 1 = runBlocking launch keepEngineRunning collect fun UI element 2 = runBlocking launch keepEngineRunning collect ```

Arun Joseph · Answer

UI elements Here UI elements are view models do not use any data from `keepEngineRunning` but keeps collecting it to make sure engine is running But I would like to prevent engine gets same location twice

Nick Allen · Answer

Use `shareIn` on `keepEngineRunning` as well It ll only run once for any number of subscribers since it s shared

Arun Joseph · Answer

I my app `keepEngineRunning` is in domain layer clean arch and doesn t receive a scope for `shareIn`

Arun Joseph · Answer

I could move to Repository level then i would need `engine` and `location` repository as one

Michael Marshall · Answer

Yeah using the UI elements to collect the flow directly rather than within a piece of business logic or viewmodel seems odd

Michael Marshall · Answer

Using a `Channel` with `receiveAsFlow` will ensure only one engine gets each emission but they will be evenly distributed among engines using a fan out method

Arun Joseph · Answer

Does this mean i shouldn t be using shared flow at all Is there any easy method to map `SharedFlow` to `Channel` Then i can reuse existing `SharedFlow` for other purposes

Michael Marshall · Answer

Using a channel might look like ``` private val locationChannel = Channel lt Either lt GeoLocationEntity LocationFailure gt gt Flow will emit one element to one collector only val locationFlow Flow lt Either lt GeoLocationEntity LocationFailure gt gt = locationChannel receiveAsFlow init callbackFlow lt Either lt GeoLocationEntity LocationFailure gt gt val onLocationUpdateListener = OnLocationUpdateListener location gt trySend location toGeoLocationEntity toOk locationEngine addOnLocationUpdateListener onLocationUpdateListener locationEngine enable awaitClose locationEngine removeOnLocationUpdateListener onLocationUpdateListener locationEngine disable Using shareIn ensures you re only setting up this listener once and sharing the emissions shareIn scope = shareInScope replay = 0 started = SharingStarted WhileSubscribed onEach locationChannel send This sends each emission into the channel ``` There may be a better way using just flows though

Michael Marshall · Answer

bump in case you didn t see my edit slightly smiling face

Arun Joseph · Answer

Hmmm This way `SharingStarted WhileSubscribed ` can t be used because there are no subscribers for callbackFlow

Arun Joseph · Answer

I tried `SharingStarted Eagerly` but then location updates happen even when there are no subscribers

Michael Marshall · Answer

Okay this is hacky but what if you did ``` private val locationChannel = Channel lt Either lt GeoLocationEntity LocationFailure gt gt Flow will emit one element to one collector only val locationFlow Flow lt Either lt GeoLocationEntity LocationFailure gt gt = locationChannel receiveAsFlow Subscribe to this to start collection but get the actual result from the other flow val locationFlowToSubscribeTo = callbackFlow lt Either lt GeoLocationEntity LocationFailure gt gt val onLocationUpdateListener = OnLocationUpdateListener location gt trySend location toGeoLocationEntity toOk locationEngine addOnLocationUpdateListener onLocationUpdateListener locationEngine enable awaitClose locationEngine removeOnLocationUpdateListener onLocationUpdateListener locationEngine disable Using shareIn ensures you re only setting up this listener once and sharing the emissions onEach locationChannel send This sends each emission into the channel shareIn scope = shareInScope replay = 0 started = SharingStarted WhileSubscribed fun UI element 1 = runBlocking launch locationFlowToSubscribeTo collect keepEngineRunning collect ```

Michael Marshall · Answer

You might be able to use `combine` to make it look like a single flow to the subscribers

Arun Joseph · Answer

``` private val locationChannel = Channel lt Either lt GeoLocationEntity LocationFailure gt gt Flow will emit one element to one collector only val locationFlow Flow lt Either lt GeoLocationEntity LocationFailure gt gt = locationChannel receiveAsFlow Subscribe to this to start collection but get the actual result from the other flow val locationFlowToSubscribeTo = callbackFlow lt Either lt GeoLocationEntity LocationFailure gt gt val onLocationUpdateListener = OnLocationUpdateListener location gt trySend location toGeoLocationEntity toOk locationEngine addOnLocationUpdateListener onLocationUpdateListener locationEngine enable awaitClose locationEngine removeOnLocationUpdateListener onLocationUpdateListener locationEngine disable Using shareIn ensures you re only setting up this listener once and sharing the emissions onEach locationChannel trySend This sends each emission into the channel shareIn scope = shareInScope replay = 0 started = SharingStarted WhileSubscribed private fun keepEngineRunning locationFlow Flow lt Either lt GeoLocationEntity LocationFailure gt gt Flow lt Either lt Unit Failure gt gt = locationFlowToSubscribeTo flowAlso locationFlow map location gt location andThen geoLocationEntity gt engine updateLocation geoLocationEntity Extension to run another flow as long as this flow is collecting fun lt S T gt Flow lt S gt flowAlso another Flow lt T gt = flow coroutineScope launch another collect this flowAlso collect emit it ```

Arun Joseph · Answer

This works Just needed to use `onEach locationChannel trySend `