Add Subscribable interfaces

[jcornaz]

Resolve #245

This PR add the following:

• interface Subscribable:
  Represent anything from which it is possible to open a subscription
• interface SubscribableValue:
  Represent a Subscriblable from which it is always possible to get a current value.
• interface SubscriblableVariable:
  Mutable SubscribableValue allowing to set the current value
• factory functions SubscribableValue(value)and SubscribableVariable(intialValue):
  Which create instances for the corresponding interfaces.

This PR also make BroadcastChannel a sub interface of Subscribable.

Decisions made:

• We do not support late initialization. Therefore ConflatedBroadacstChannel do not implement SusbcribableVariable. This makes the overall usage simpler and safer to use.
• Operators such as map and combine or adapter for JavaFX ObservableValue won't be part of this PR, but may be made and discussed in another issue/PR

[fvasco]

We should provide builders for SubscribableValue and SubscribableVariable.

[fvasco]

offer should be enough.

[jcornaz]

offer would never throw any exception. But I could do if (!offer(value)) throw IllegalStateException().

Or do we prefer to silently fail if the broadcast channel is closed ?

[fvasco]

Are you sure?

kotlinx.coroutines/core/kotlinx-coroutines-core/src/main/kotlin/kotlinx/coroutines/experimental/channels/ConflatedBroadcastChannel.kt

Line 201 in fc6e31e

override fun offer(element: E): Boolean {

[jcornaz]

Yes sorry, I was wrong. I'll use offer. At least I learned something :-)

[fvasco]

Is it already defined in SubscribableValue?

[jcornaz]

Yes, but Kotlin compiler enforce to redeclare it, because it is inherited from two interfaces (SubscribableValue and ReadWriteProperty)

[jcornaz]

https://kotlinlang.org/docs/reference/classes.html#overriding-rules

[fvasco]

Ok

[fvasco]

We should support lateinit?
Is Deferred<SubscribableValue> a good replacement?

[jcornaz]

I think yes, we should support it, because this is the actual current behavior of ConflatedBroadcastChannel.

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-conflated-broadcast-channel/value.html

[fvasco]

Ok, discard this point.

[fvasco]

We should provide also

fun SubscribableValue.map(...) : SubscribableValue

However this implementation should be lazy and GC friendly.

[jcornaz]

We should provide builders for SubscribableValue and SubscribableVariable.

Yes, I'll propose something.

[jcornaz]

Definitely yes for the map operator. I was just not sure if it should be part of this PR or not.

If a map operator is provided, I think we should also provide:

fun <T1, T2, R> SubscribableValue<T1>.combineWith(other: SubscribableValue<T2>, combine: (T1, T2) -> R): SusbcribableValue<R>

Should I also provide an adapter for JavaFx: fun <T> SubscriblableValue<T>.asObservableValue(): ObservableValue<T> in the module kotlinx-coroutines-javafx? To me it makes a lot of sense and I would write it in any of my JavaFx application anyway. This is something that would make interoperability between coroutines and JavaFx seamless and very pleasant. But perhaps, I should do it in a separate PR?

[jcornaz]

We should provide builders for SubscribableValue and SubscribableVariable.

Here is a wild API suggestion. Is it looking like what you imagined @fvasco ?

interface SubscribableValueBuilderScope<in T> : CoroutineScope {
  suspend fun yield(value: T)
}

interface SubscribableValueJob<T> : SubscribableValue<T>, Job

fun <T> SubscribableVariable(initialValue: T): SubscribableVariable<T> = ConflatedBroadcastChannel(initialValue)

fun <T> buildSubscribableValue(
    context: CoroutineContext = DefaultDispatcher,
    builderAction: suspend SubscribableValueBuilderScope<T>.() -> Unit
): SubscribableValueJob<T> = TODO()

fun usageExample() {
  val variable = SubscribableVariable<Int>(0)

  variable.value = 42

  val currentTime: SubscribableValueJob<Instant> = buildSubscribableValue {
    while (isActive) {
      delay(1, TimeUnit.SECONDS)
      yield(Instant.now())
    }
  }

  currentTime.cancel()
}

[fvasco]

buildSubscribableValue is really looks like the actor builder, I seen your code and probably the ConflatedBroadcastChannel class is enough, at least for my use case.

Instead I am considering another kind of actor

fun <E> SubscribableValue<E>.launchObserver(
        context: CoroutineContext = DefaultDispatcher,
        parent: Job? = null,
        block: suspend ActorScope<E>.() -> Unit
): Job

This requires explicit cancel so I am sceptic about it.
However this fits my use case, can you share your considerations?

[jcornaz]

buildSubscribableValue is really looks like the actor builder

Actually more like produce than actor. But creating broadcast instead of a channel.

probably the ConflatedBroadcastChannel class is enough, at least for my use case

For my use case too. So I won't add buildSubscribableValue in this PR. Should I add the SubscribableVariable() function, which would only be an alias to ConflatedBroadcastChannel constructor?

About the launchObserver, we implemented the following in our code-base:

/**
 * Start a job that consume each elements of the channel and execute [onEach] of each of them.
 */
fun <E> ReceiveChannel<E>.launchConsumer(context: CoroutineContext = DefaultDispatcher, onEach: suspend (E) -> Unit): Job =
    launch(context) { consumeEach { onEach(it) } }

I feel the same Idea than behind your launchObserver. The difference is that this one can be used for any ReceiveChannel regardless if it is coming from a broadcast or not.

I am not particulary found of the SubscribableValue.launchObserver you proposed for the exact reason you mentioned: "it requires explicit cancel".

And anyway, I do not need it personally. Because I use JavaFx, I would most of the time simply bind a JavaFx component to the result of SubscribableValue.asObservableValue(). With this pattern when the JavaFx component would unsubscribe, everything would be cancelled. So it is easier and safer to use (in the context of JavaFX).

And for other usages, it is not really an hassle to do:

launch {
  observableValue.openSubscription().consumeEach {
    // do something usefull
  }
}

But this is purely personal opinion. I will add it, if you guys want it.

[fvasco]

Well,
so limit this pull request to "Subscribable interfaces", probably it is not useful dump all ideas here.

[jcornaz]

Ok, I added some comments and a test for ConflatedBroadcastChannel.setValue

What should we do for BroadcastChannel.openSubscription()? We cannot remove it without breaking binary compatibility. But we cannot deprecate it either, because it is still the correct method to use.

Let me know if I should correct or do something.

I'll open new PRs for the map and combine operators and for the JavaFx adapter.

[fvasco]

What should we do for BroadcastChannel.openSubscription()?

Mark it for removal using a :todo: comment.

[fvasco]

Resume public modifier.

[fvasco]

Resume public modifier.

[fvasco]

Resume public modifier.

[fvasco]

Add public modifier.

[fvasco]

Add public modifier.

[fvasco]

Add public modifier.

[fvasco]

openSubscription().first()?

[fvasco]

What happen if channel is closed?

[fvasco]

Can you test some negative paths?

[jcornaz]

Can you test some negative paths?

Yes, good Idea, I'll add some for ConflatedBroadcastChannel

[jcornaz]

Ok, I added some test for negative paths (channel with no value set yet, channel closed, channel closed with a cause)

[bohsen]

Just wanted to say really great work to you guys.
The ConflatedBroadcastChannel IMO is a really powerful tool.
I've used something similar where I had to combine it with Android Architecture Components and a custom LiveData class (look below, boilerplate removed), but having this in the library will be SO good. Could also help solving #232 and #258.

/**
 * [LiveData] object for observing changes in viewstate. Uses kotlin coroutine [channel]s for
 * communicating changes
 */
class UiStateLiveData : LiveData<UiStateModel>() {

    private val channel = ConflatedBroadcastChannel<UiStateModel>(UiStateModel.Error(IllegalStateException("Illegal state")))
    private lateinit var subscription: SubscriptionReceiveChannel<UiStateModel>

    override fun onInactive() {
        subscription.close()
        super.onInactive()
    }

    override fun onActive() {
        super.onActive()
        subscription = channel.openSubscription()
        async { subscription.consumeEach { value -> postValue(value) } }
    }

    fun setUiState(state: UiStateModel) {
        async { channel.send(state) }
    }
}

[fvasco]

Should we support better lazy initialization?
Probably valueOrNull isn't good enough (null is a valid value for T).

public interface SubscribableValue<out T> : Subscribable<T>, ReadOnlyProperty<Any?, T> {

    public fun isInitialized(): Boolean

    public val value: T

    ...
}

public suspend fun <T> SubscribableValue<T>.awaitValue() =
        when {
            isInitialized() -> value
            else -> openSubscription().first()
        }

[dave08]

@fvasco Maybe value should have been lazy if no initial value was passed, since I think in most cases, a subscribable value should really have an initial value... At least in my use case for Andoid MVP or MVVM it would always be that way... So it would make the more common use case easier to use... BTW nice work guys! This could be very very helpful!

So if value can't be lazy, maybe enforce providing an initial value? I guess that making a val into suspending is not possible or recommended... Or maybe also have a lambda as a conflated channel initializer...

[jcornaz]

@bohsen and @dave08: thank you. It's good to hear that it'll be helpful.

@fvasco:

Should we support better lazy initialization?
Probably valueOrNull isn't good enough (null is a valid value for T).

As it is already supported in the way that it is possible create a ConflatedBroadacstChannel with no initial value, I think yes it would be a good Idea to provide a isInitialized property.

But how should it behave if the channel is closed? Should it fail? Or should it return false?

[fvasco]

The idea is to provide a better support for the lazy initialization adding isInitialized and awaitValue.

(sorry for unglish :)

[jcornaz]

The idea is to provide a better support for the lazy initialization adding isInitialized and awaitValue.

Yes I got it. Sorry if I wasn't clear ;-)

I added isInitialized and awaitValue with a test for it.

EDIT: Note, that I choose to make isInitialized fail if the channel is closed. Let me know what you think about it.

[fvasco]

I choose to make isInitialized fail if the channel is closed. Let me know what you think about it

It might be acceptable.

Tip: remember to change SubscribableValue.value and SubscribableVariable.value documentations

[jcornaz]

Tip: remember to change SubscribableValue.value and SubscribableVariable.value documentations

Yes good catch. I did it.

It might be acceptable.

I'm not especially found of it, but returning false is not semantically correct IMO (as it may have been initialized before being closed). So I guess we have to choose the less bad option.

[jcornaz]

And what if we rename isInitialized by hasValue? If so, it would be semantically correct to return false if the channel is closed.

[fvasco]

Sorry @jcornaz,
I back to my steps (#245 (comment)).
I consider difficult to understand why a SubscribableValue may have no value.

Can you consider this design instead?

public interface SubscribableValue<out T> : Subscribable<T>, ReadOnlyProperty<Any?, T> {

    public val value: T

    public override fun getValue(thisRef: Any?, property: KProperty<*>): T = value
}

public interface SubscribableVariable<T> : SubscribableValue<T>, ReadWriteProperty<Any?, T> {

    public override var value: T

    public override fun getValue(thisRef: Any?, property: KProperty<*>): T = value

    public override fun setValue(thisRef: Any?, property: KProperty<*>, value: T) {
        this.value = value
    }
}

fun <T> SubscribableValue(value: T) = SubscribableVariableImpl(value)
fun <T> SubscribableVariable(value: T) = SubscribableVariableImpl(value)

internal class SubscribableVariableImpl<T>(value: T) : SubscribableVariable<T> {

    private val channel = ConflatedBroadcastChannel(value)

    public override var value: T
        get() = channel.value
        set(value) {
            channel.offer(value)
        }

    public override fun openSubscription() = channel.openSubscription()
}

No errors, no checks, no corner cases.
It is possible cover more use case using Deferred<SubscribableValue>> or SubscribableValue<Option>.

Take your time.

[jcornaz]

I think you make a very good point.

No errors, no checks, no corner cases.

That's appealing indeed... It would be simpler and safer. Sounds very good.

In the other hand, I do need lazy initialization in practice. But it doesn't have to be in kotlinx.coroutines. I could write my own LazySusbcribableValue for my use-case (which we already have in our code-base). Or if it appears other have interest in LazySubscribableValue we could have an explicit interface for it, different from SubscribableValue, to avoid confusion.

[jcornaz]

Here is the API I'd propose:

(implementation code of ReadOnlyProperty and ReadWriteProperty is not shown)

public interface Subscribable<out T> {
	fun openSubscription(): SubscribtionReceiveChannel<T>
}

public interface SubscribableValue<out T> : Subscribable<T>, ReadOnlyProperty<Any?, T> {
    
	// Never throw, always succeed
	public val value: T
}

public interface SubscribableVariable<T> : SubscribableValue<T>, ReadWriteProperty<Any?, T> {

	// Never throw, always succeed
       public override var value: T
}

public interface LateinitSubscribableValue<out T> : Subscribable<T>, ReadOnlyProperty<Any?, T> {

	// never throw
	public val isInitialized: Boolean
    
	// throw only if no value set yet
        public val value: T
	
	// never throw
	public val valueOrNull: T?
}

// Never throw. May suspend until a value is available
suspend fun <T> LateinitSubscribableValue<T>.awaitValue(): T = openSubscription().first()

public interface LateinitSubscribableVariable<T> : LateinitSubscribableValue<T>, ReadWriteProperty<Any?, T> {

	// only the getter throw if no value set yet
       public override var value: T
}

Edit: I renamed Lazy by Lateinit. It is more appropriate I think.

[fvasco]

It is possible to replace LateinitSubscribableValue with Deferred<SubscribableValue> to cover the same use case?

[elizarov]

So I had this crazy idea toady. Instead of Subscribable interface that exposes openSubscription, do Consumable interface that exposes consumeEach (make it a member). That would let us unify all 3 kinds of abstractions: channels, broadcast channels, and cold streams (#254) and then provide a single set of (efficiently implemented!) operators like fitler, map, etc on all the types of streams with one piece of code (#285). I'll leave it for a while here to think though all the implications.

The key observation is that you try to write fitler or map on top of Subscribable, then it is bound to be inefficient, since it'll have to "pull" data via receive (or even worse -- via iterator's slow hasNext/next protocol). However, via Consumable the data is "pushed" down the pipe in an extremely efficient way.

[jcornaz]

This is very interesting an sounds promising. However, consumeEach is actually specific to an iteration over all values. And some terminal operators (those which don't need to iterate over all values) could be hard to implement efficiently. (For example, first, any, contains etc.)

But I still think it is a good Idea, and it is going in a very good direction.

So here is an alternative proposition:

What if we make ChannelIterator implement Closeable and add aConsumable interface (as you propose) but exposing a fun iterator(): ChannelIterator<E> function instead of consumeEach ?

Here is a quick draft of the idea: master...jcornaz:spike/consumable

[elizarov]

Iterators are inherently slow. My idea for consumeEach was that terminal operations that do not need to scan all elements would throw CancellationException when they are done. We can use a shared instance and that should be pretty efficient, however I'll need to do more benchmarking around this particular case.

[smaldini]

Sounds like something we are planning to add in reactor-core as an extension. fromAsyncIterable/toAsyncIterable (in addition to fromCoroutine/toCoroutine). We might be able to use that new contract instead of creating a new AsyncIterable of our own.

[fvasco]

I wrote a little example using exit to stop consuming.

interface Consumable<E> {
    fun consumeEach(consumer: suspend ConsumerScope<E>.(E) -> Unit): E?
}

interface ConsumerScope<E> {
    suspend fun exit(exitValue: E? = null): Nothing
}

fun Consumable<Int>.sum(): Int {
    var sum = 0
    consumeEach { sum += it }
    return sum
}

fun <E> Consumable<E>.first() = consumeEach { exit(it) }

[jcornaz]

Just to clarify, should I update the PR to provide and use Consumable instead of Subscribable ?

Or is this PR only waiting for design decision?

[elizarov]

We clearly need a prototype to make this decision. It is tightly bound to the decision on cold streams abstraction in #254, since it is basically the same "subscribable" abstraction that we are talking about here. It does not look that we have place for two different abstractions here. We already have too many of them.

The proof of concept should demonstrate that all the interesting operators like filter, map, first, take, etc that are currently defined as extensions to ReceiveChannel can be rewritten as extension to the new subscribable/coldStreams abstraction while keeping the code simple. We also need POC to measure the performance of the resulting code.

[SolomonSun2010]

Are you going to support Subscribable as dataflow concurrency in Oz ?https://www.info.ucl.ac.be/~pvr/GeneralOverview.pdf

I like it , Data-Driven Concurrency is cool !
Also, may I propose to rename these names similar with Groovy GPars ?
http://gpars.org/webapp/quickstart/index.html#__strong_dataflow_concurrency_strong
@jcornaz @fvasco @elizarov

see also:
Building Scalable, Highly Concurrent & Fault Tolerant Systems - Lessons Learned
https://www.slideshare.net/jboner/building-scalable-highly-concurrent-fault-tolerant-systems-lessons-learned?from_action=save

Dataflow Concurrency
• Deterministic
• Declarative
• Data-driven
• Threads are suspended until data is available
• Lazy & On-demand
• No difference between:
• Concurrent code
• Sequential code
• Examples: Akka & GPars

[jcornaz]

@SolomonSun2010 data-flow concurrency is already possible with channels. And there is active design discussion about how to provide cold stream in #254.

[qwwdfsad]

When #254 is almost here, I think it is time to close this PR.

During our design phase, we have a lot of discussions on whether Channel should implement Flow or not and decided not to; and this separate interface carries more burden than a value when living side-by-side with Flow.

There is also a natural desire to use channels as subjects and for this, we have Flow.broadcastIn and BroadcastChannel.asFlow, so I'd better focus our efforts on shaping this API

[fvasco]

No problem, all these consideration can be cherry picked when needed.