Consider sharing a Flow through a ConnectableFlow

[streetsofboston]

Enhancement: Add ConnectableFlow to the Flow API.

Each time an observer of a Flow starts collecting, the source of the Flow is executed, much like a call to subscribe of a Flowable in RxJava executes the Flowable's source.

This change is to defer the execution of the source of the Flow until a specific point in time, possibly after one or more observers started collecting the 'shared' Flow.

The use-case for deferring the execution of the source of a Flow is for (cold) Flows whose data-source is a resource that should not be started/created or stopped/destroyed by each and every call to collect and should be explicitly managed by a call to a function (connect, for example) instead. It differs from using broadcastIn by the fact that publish will return a Flow, not a BroadcastChannel.

E.g.

val dataFlow = flowViaChannel<MyData> { channel ->
    val resource = getResource(channel)
    channel.invokeOnClose {
        resource.close()
    }
    resource.startReceivingData()
}

val sharedFlow = dataFlow.publish()
...
...

val observer1 = launch {
    sharedFlow.collect { ... }
}
...
...
val observer2 = launch {
    sharedFlow.collect { ... }
}
...
// Start the flow right now.
val connection = sharedFlow.connect(scope)
...
...
...
// Cancel the flow here.
// Note that when 'scope' is cancelled, this 'connection' would be canceled as well.
connection.close()
...

I propose creating these new classes and extension functions or something similar (they are modeled after RxJava ConnectableObserver):

/**
 * A [Flow] of type [T] that only starts emitting value after its [connect] method is called.
 * 
 * If this flow's [Connection] is still connected, the current [Connection] will be returned when 
 * [connect] is called and the flow will not be restarted. 
 *
 * When its [collect] method is called, this flow will not immediately start collecting. Only after
 * [connect] is called, the emission and actual collecting of values starts.
 */
interface ConnectableFlow<out T> : Flow<T> {
    /**
     * Connects this shared [Flow] to start emitting values.
     *
     * @param scope The [CoroutineScope] in which the emissions will take place.
     * @return The [Connection] that can be closed to stop this shared [Flow].
     */
    fun connect(scope: CoroutineScope): Connection
}

and

/**
 * A connection returned by a call to [ConnectableFlow.connect].
 */
interface Connection {
    /**
     * Returns true if this connection is connected and active.
     */
    suspend fun isConnected(): Boolean

    /**
     * Closes this connection in an orderly fashion.
     */
    suspend fun close()
}

/**
 * Publishes and shares an upstream [Flow] of type [T] and returns a [ConnectableFlow] of type [T].
 *
 * The upstream [Flow] begins emissions only after the [ConnectableFlow.connect] has been called.
 *
 * @return The [ConnectableFlow] that represents the shared [Flow] of this receiver.
 */
fun <T> Flow<T>.publish(): ConnectableFlow<T> 

/**
 * Creates a [Flow] of type [T] from this [ConnectableFlow] that automatically connects (i.e. calls
 * [ConnectableFlow.connect]) when the first [numberOfCollectors] observer starts collecting (i.e. calls [Flow.collect])
 * and automatically cancels this [ConnectableFlow] when the last observers stops collecting.
 *
 * @param scope The scope in which this [ConnectableFlow] will be connected.
 * @param numberOfCollectors The number of observers that need to start collecting before the connection (re)starts.
 * @return The shared referenced-counted [Flow].
 */
fun <T> ConnectableFlow<T>.refCount(scope: CoroutineScope, numberOfCollectors: Int = 1): Flow<T> =

/**
 * Shares this [Flow] of type [T] with multiple observers without restarting when each observer starts
 * collecting. This is the same as calling [Flow.publish] and then [ConnectableFlow.refCount].
 *
 * @param scope The scope in which this [ConnectableFlow] will be connected.
 * @return The new [Flow] that shares this [Flow]
 */
fun <T> Flow<T>.share(scope: CoroutineScope): Flow<T> = publish().refCount(scope)

This is my first stab at an initial/draft/try-out implementation:
https://gist.github.com/streetsofboston/39c3f8c882c9891b35e0ebc3cd812381

Update:
I took autoConnect out: This is more for 'replay' and 'caching'. If needed, this should be addressed in a separate issue.

[elizarov]

Can you please add some actual use-case as in "here is the what application is trying to do and that is what it wants to achive" without tying this use-case to the actual solution in the form of ConnectableFlow. There are tons of methods here. All of them need use-case (refCounting, autoConnect), etc.

[streetsofboston]

Use case for ConnectableFlow:

Cold Streams are often Unicast. When an observer/consumer starts observing, the source of the Stream is started again. The Flow API currently allows for cold unicasts, where its source is (re)started each time collect is called.

Hot Streams are often Multicast where observers/consumers can come and go without them restarting anything.
The Channel and BroadcastChannel APIs support this and the method broadcastIn already exists.

Sometimes, it is desirable to have Cold Streams that are Multicast. The source of the stream may not always be active (it may be expensive to have the stream being active all the time or starting a new one each time), and starting the stream does not depend on whether any observers/consumers are actually observing. Starting and stopping the cold multicast stream needs to be managed explicitly.

The proposed ConnectableFlow would implement such Cold Multicast Flow, where the source of the Flow (re)starts each time when its connect method is called and where the source of the Flow stops when this connection is closed.

Examples of such cold multicast streams are BLE (Bluetooth Low Energy) characteristics that notify the observer of data changing on an external device, e.g. a BLE thermometer or any other continuous monitoring device. Starting a characteristic keeps a connection open between the observer and the BLE device and this can be somewhat expensive. It is best to manage this explicitly, e.g. have the user click a 'connect' and 'disconnect' button or to manage it implicitly by only starting the connection when observers on the UI are observing the device.

Make a cold unicast Flow a cold multicast ConnectableFlow:
fun <T> Flow<T>.publish(): ConnectableFlow<T>

Implicitly manage the connection of a ConnectableFlow by reference-counting:
This allows a cold multicast Flow to be active only when observers are listening/collecting.
E.g. Keep a BLE Characteristic notification stream active if the user is looking at a UI that needs it.
fun <T> ConnectableFlow<T>.refCount(scope: CoroutineScope, numberOfCollectors: Int = 1): Flow<T>

[elizarov]

Sometimes, it is desirable to have Cold Streams that are Multicast. The source of the stream may not always be active (it may be expensive to have the stream being active all the time or starting a new one each time), and starting the stream does not depend on whether any observers/consumers are actually observing. Starting and stopping the cold multicast stream needs to be managed explicitly.

Can you, please, provide an example with an actual application scenario (as in "here is the actual application I'm writing and here is why I need it") where a cold stream needs to be mutlicast, but it cannot be always active (so you cannot just use always active BroadcastChannel) and when, at the same time, starting/stopping stream does not depend on the presence of observers, so that it needs to be managed explicitly.

[LouisCAD]

@streetsofboston To me, that Bluetooth GATT notifications use case is not a cold stream at all, but a hot one with manual start and stop, so effectively just a channel (possibly broadcast), and two custom functions to start/subscribe and stop/unsubscribe.

[streetsofboston]

@LouisCAD You're correct.
But adding the two custom functions to start and stop the steam could be also handled by the ConnectableFlow's connect method and the close method on the returned Connection, much like a ConnectableObserver in Rx (connect and dispose). The connect call would make the underlying stream hot (i.e. it will start it and the stream will emit values until close on the returned connection is called).

I also do believe the addition of ConnectableFlow to the core Flow API is not required and stuff can be done manually (I implemented a ConnectableFlow myself in the gist I linked, using the public Flow api). But it can be convenient to other devs. Maybe ConnectableFlow can be part of an extension library?

I'm currently not working on any BLE app right now, but have been in the past on an apps that used plain callbacks the RxAndroidBLE library. There we make use of ConnectableObservables. But that has been a while and I need to dig into the past a little to get a good use-case. :-)

[LouisCAD]

@streetsofboston Actually, I made a library for Bluetooth Low Energy with coroutines a while ago (and I keep it updated). Notifications support works with channels, although I've not needed notifications support myself. If you think that part of the API may be improved, feel free to open an issue there!

[matejdro]

To me, that Bluetooth GATT notifications use case is not a cold stream at all, but a hot one with manual start and stop, so effectively just a channel (possibly broadcast), and two custom functions to start/subscribe and stop/unsubscribe.

All that is true, but wrapping GATT into cold stream provides benefits of automatic state management. Programmer can easily forget to call manual start/stop method (especially stop). But with flow, this is done automatically (start on collect call, stop when coroutine is cancelled).

Another use case:

Developing a mobile app that uses GPS location at various points. Location is exposed as flow stream. Whenever part of app needs access to Location, it starts collect on stream, GPS receiver activates and location is transmitted. If another part needs access to Location, it can collect on the same stream. Since stream is already active, it would just multicast new location data to all subscribers. When part does not need access to Location anymore, it cancels the collecting coroutine. Once all producers are cancelled, GPS receiver shuts off, saving battery.

[inorichi]

This last example might be somewhat related to #1097, but using Flows instead. Not sure if they could share a common implementation, though.

Edit: although it's only the analogue of the .refCount() operator. .connect(N) must be cancelled manually, or with the scope (what OP has asked for).

[elizarov]

Let me summarize what I'm getting out of this thread so far. I see a bunch of use-cases here for an operator that automatically actives a flow on a first collector, shares the emitted events with all the other collectors, and cancels the flow instance as soon as the last collector is done. Easy, usefull, no chance of resource leakage, no need to introduce any new types like ConnectableFlow -- it is just an operator. The only question is how do we have name it. Can we name it just share()?

A kind of manual activation/deactivation of the flow sounds like a use-cases for a channel to me. You can already do flow.produceIn(scope) to active a flow and we might even provide a scope-less flow.produce() variant. produce activates a flow and returns a channel. You can cancel this channel when you no longer need it -- that is your manual activation even when you don't have any collectors.

Does this sound like a plan?

[matejdro]

For my use case, this sounds perfect.

[nhaarman]

Next to just share()ing the subscription, new subscribers often want to immediately receive the most recent value that was emitted. In Rx this is done with the replay(1) operator.

Take for example the way Firebase Database provides lists of items. Firebase emits events like 'item added', 'item removed', 'item moved', etc. Clients can construct the entire list from these events.
A Flow can scan these events and construct the list, sharing the result:

databaseEvents
  .scan(emptyList()) { list, event ->
     list.apply(event)
  }
  .replay(1).refCount()

Replacing replay(1) with publish(), new subscribers potentially never receive any emissions.

[streetsofboston]

@elizarov
I updated my gist with some examples, use-cases for ConnectableFlow and share().
https://gist.github.com/streetsofboston/39c3f8c882c9891b35e0ebc3cd812381
(see the 2nd file in this gist called Example.kt)

I also tried to implement the UseCasesForConnectableFlow.manage_expensive_start_and_stop_of_resource() use-case/example using the produceIn function instead, and even tried it using the broadcastIn function. I was not able to do this in a way that produced the same results when using ConnectableFlow from my implementation. In this use-case, the ExpensiveResource should be started about 3 seconds after the call to fun manage_expensive_start_and_stop_of_resource(), not earlier.

The main issue with using produceIn is that the listeners (collectors/consumers) need to be able to get a reference to a Flow (or a ReceiveChannel, BroadcastChannel, etc) before the cold stream is activated, but produceIn returns an activated ReceiveChannel...

Maybe I'm overlooking something and missing an implementation that works without the introduction of something new like a ConnectableFlow.