Introduce broadcast() extension to turn ReceiveChannel into BroadcastChannel

[elizarov]

We need a broadcast operator that turns a ReceiveChannel into BroadcastChannel.

[jcornaz]

Should the broadcast operator directly start to send elements? Or should it wait for the call on a method, like connect of the Rx publish operator?

[elizarov]

We are talking about hot channels here, so it makes sense to start sending them immediately. It would be an interesting design question for broadcast() (and for produce()) operators on cold streams (see #254), but I would also argue that for cold channels, too, the very invocation of broadcast() is a signal that you want to "meterialize" the source (start running it).

[jcornaz]

I agree, with hot channel there is no point for not starting immediately.

But for cold channel, I think it is more arguable. Even if I would be ok to start immediately, I think some user could find useful to at least decide how it behave depending on the use-case. With Rx for example the user may use connect, autoConnect, refCount and share to manage it.

However, may I ask what you would think about the following implementation?

fun <E> ReceiveChannel<E>.broadcast(capacity: Int): BroadcastChannel<E> {
  val broadcast = BroadcastChannel<E>(capacity)

  val job = launch(Unconfined) {
    try {
      consumeEach { broadcast.send(it) }
    } finally {
      broadcast.close()
    }
  }

  return object : BroadcastChannel<E> by broadcast {
    override fun close(cause: Throwable?): Boolean {
      job.cancel(cause)
      return broadcast.close(cause)
    }
  }
}

[elizarov]

@jcornaz you should propagate close cause to broadcast.close, that is add catch(e: Throwable) before finally to remember the cause.

It also seems better to also implement a broadcast { ... } builder for general use that performs this binding of broadcast channel to a job, and then implement .broadcast() operator via that broadcast { ... } builder in a straightforward way.

And it will good for now.

For a future, though, note that all current builders (like produce and actor) return a new object that delegates all the channel methods to the underlying channel (like by broadcast in your implementation). This is a suboptimal solution. Actually, if we introduce a linked job in every channel implementation (see discussion in #260), then channels returned by produce, actor, and broadcast will be more efficient as they will not require any delegation (because every channel will natively support a linked job).

[elizarov]

I'll do it myself.

[elizarov]

@sdeleuze Can you, please, elaborate a bit on your use-cases. I'm slightly worried that existing implementations of broadcast channel may not cover your needs, so we may need to add some new ones. In particular, I cannot figure out what kind of BroadcastChannel we shall create in broadcast() by default.

What kind of behavior is desired? When should happen to the produced items before the first invocation of openSubscription? Shall they be dropped or buffered? If buffered, then up to what limit? If they are buffered, then what should happen after the first openSubscription? Shall they be dropped or buffered (basically forever) for all the subsequent subscribers?

[jcornaz]

I cannot figure out what kind of BroadcastChannel we shall create in broadcast() by default

Couldn't it be a capacity argument, capable of accepting Channel.CONFLATED for example?

[sdeleuze]

The use case is to multicast elements from a hot stream to multiple clients, for example we can imagine a Coroutine based HTTP client requesting a remote JSON streaming endpoint on server-side and broadcasting to 100+ SSE endpoints, that's what we do with Flux + share(). No need to buffer, items could be lost for late subscribers.

[elizarov]

@sdeleuze Thanks. That is the same conclusion I came to -- mimic share by default. So, by default broadcast() is going to use CoroutineStart.LAZY and start only when the first subscriber comes in. That guarantees that first subscriber never looses elements. Late subscribers may loose them, though.

The tradeoff in this behavior, is that the producer of the stream will be suspended (backpressure) until the first subscriber appears. I really hope we are choosing the right default. Maybe we should have no default at all, but always require an explicit specification of what to do.