Flatten behaviour unexpected

[cefn]

I have found myself reimplementing the flatten behaviour, because it has unexpected consequences which I find hard to square with the 'contract' I was expecting from Kefir's lazy subscriptions. In particular, flatten() seems to fall between two models of synchrony.

Consider flattening the constant property [1,2,3]

In one model, the upstream subscription is made when a subscription is made to flatten, then flatten processes the resulting array events, splitting them into elements and passing them on.

Model1
<value:current> 1
<value:current> 2
<value:current> 3
<end:current>

On another model, the upstream subscription is made BY flatten(), flatten consumes all synchronously-available events. Subscribers receive only entries from future arrays which will be flattened (which is none in this case). So when you call flatten on a constant property, you receive nothing by the time you've subscribed, so this would only be useful e.g. for realtime streams or pool.

Model 2
<end:current>

In fact, the model is somewhere in between these, and it's hard to describe what it is. What actually happens is the following...

Model3
flatten <value:current> 3
flatten <end:current>

It seems that when it receives a downstream subscriber, it subscribes upstream, receives an array, then throws away a whole load of the entries, serving only the final entry.

This script demonstrates the issue, as well as providing a basic implementation of the kind of behaviour expected (noted as 'bespoke).

var Kefir = require("kefir");

var source = Kefir.constant([1,2,3]).toProperty();

var flattener = function(emitter, event){
    if(event.type==="value"){
        event.value.map(emitter.emit);
    }
    else{
        emitter.emitEvent(event);
    }
};

Kefir.stream(function(emitter){
    var handler = flattener.bind(null, emitter);
    source.onAny(handler);
    return function(){
        source.offAny(handler);
    }
}).log("bespoke");

source.withHandler(flattener).log("withHandler");

source.flatten().log("flatten");

Is there any way to achieve the expected Model1 behaviour through Kefir operations and flatten? The current behaviour seems surprising, but is also something I need to work around.

Is there a clear statement of the subscription contract which is implied by Kefir's derived streams which explains how flatten and withHandler should operate?

[rpominov]

The issue actually in property implementation. A property always has a single current value, if it receives multiple current values from its source, it just ignore all but the latest of them.

Also flatten returns an observable of same type, so since you apply it to a property it returns another property.

We have following options of fixing this:

1. remove that constraint from the property (that it always has a single current value)
2. make flatten always return a stream
3. add a conventional way for converting properties to streams, that preserves current value, so we could do prop.toStream().flatten() (we have prop.changes(), but it removes current)

I don't have a strong opinion on this yet, but perhaps (1) is what should be fixed. I haven't use Kefir for a while, and have lost the context of why that constraint was added.

[cefn]

I've been working on the assumption that flatten() wouldn't activate its upstream Kefir.constant() until the first onValue() subscription was made. Is this false? Is flatten() itself counted as a subscriber? Does this subscription behaviour apply to all observable transformations?

It's the fact that the value assignments of 1 and 2 are missing which is confusing, and doesn't seem to relate to the notion of Property directly. I think it would be fine that the property still has a single current value, but I'd expect in this case that the value is assigned first to 1, then 2, then 3 after the first onValue subscriber was added.

If you modify the example to replace the declaration var source = Kefir.constant([1,2,3]) with var source = Kefir.sequentially(0, [[1,2,3]]).toProperty(); all cases then conform to the expected 'model1' behaviour, suggesting the consumption of earlier events by flatten and withHandler doesn't relate directly to property nature.

However, there's a property-related explanation. The stream was already activated by the flatten() operation ('model2' behaviour) therefore all events '1', '2' and '3' have gone missing before first subscription, except that the property nature inherited by flatten() means it retained the last event '3'. I'm not sure this is ever what is wanted, though. On this analysis it seems that the design of flatten() means that it's actually impossible to catch all the events flattened from a Property which is surprising.

In the alternate sequentially(...) initialisation of the stream, an onValue subscriber is added before the first event was issued by the property, and therefore the flattened events are able to be detected by the onValue subscriber.

I thought that each step in the pipeline was only subscribed when it was activated (e.g. a downstream subscriber was added), and that transformations didn't themselves count as subscribers or activate the pipeline, so that's why the behaviour was so confusing to me.

In my use of Kefir Properties I very often employ them as syntactic sugar for an event which may have already happened, and may go on to happen several times in the future (meaning I don't have to handle the cases differently). For this reason, I definitely favour behaviours in which an event coming from a Property and an event coming from a Stream are normalised, preferably by default or if necessary by some workaround.

[cefn]

With respect to the alternative fixes you described...

I don't understand why Property nature would be inherited through observable transformations. At any point that you want to re-assert Property nature explicitly, you can do so with toProperty(), so it seems surprising. I understood toProperty as creating a derived stream which serves a single cached event to each new immediate subscriber, but that a Property continues to present itself as a stream without any additional affects on downstream subscribers beyond that serve-cache-on-immediate-subscribe behaviour. My preference would be to avoid property nature inheritance by transformed observables to avoid surprising model3 sequences, which are neither one or the other. However, I recognise I may not really understand the issues here.

The toStream() transformation (where this is understood to be equivalent to wiring a stream to the onAny of an observable and hence would received cached values from Properties) would have allowed me to debug the issue (and recreate the less surprising model2 behaviour), but it's still model1 flatten behaviour that I need which seems impossible by any transformation route at the moment.

I think the only chance of getting model1 behaviour is based on end-to-end lazy subscription (where the flatten() call doesn't itself activate the upstream source until a downstream onValue() is subscribed). If flatten() activates upstream, then the events have already gone. If the opposite assumption is baked into the Kefir library then I understand this won't be easy to achieve.

Sorry for lots of comments. If you're not actively working on Kefir it must be a pain, sorry. I find the library to be excellent, though (and well documented), so I'm very keen to continue building on it.

[rpominov]

I've been working on the assumption that flatten() wouldn't activate its upstream Kefir.constant() until the first onValue() subscription was made. Is this false?

It activates upstream only after first onValue()

Is flatten() itself counted as a subscriber?

Well, it activates upstream by subscribing to it :)

Does this subscription behaviour apply to all observable transformations?

Sort of, when you apply a transformation, you create a new observable, that will subscribe to original one on activation.

It's the fact that the value assignments of 1 and 2 are missing which is confusing, and doesn't seem to relate to the notion of Property directly. I think it would be fine that the property still has a single current value, but I'd expect in this case that the value is assigned first to 1, then 2, then 3 after the first onValue subscriber was added.

Probably there is a misunderstanding of what I meant by property having single current value. I call "current" all values, that an observable emits immediately (synchronously) to a new subscriber right after it passed to onValue. Currently it's guaranteed, that when you subscribe to a property, you'll get at most one such value.

If you modify the example to replace the declaration var source = Kefir.constant([1,2,3]) with var source = Kefir.sequentially(0, [[1,2,3]]).toProperty(); all cases then conform to the expected 'model1' behaviour, suggesting the consumption of earlier events by flatten and withHandler doesn't relate directly to property nature.

With .sequentially(0, [[1,2,3]]).toProperty() you create a property with empty current value — it won't emit [1,2,3] to the first subscriber immediately, but will emit it after 0 timeout as a normal value.

I thought that each step in the pipeline was only subscribed when it was activated (e.g. a downstream subscriber was added), and that transformations didn't themselves count as subscribers or activate the pipeline, so that's why the behaviour was so confusing to me.

Yeah, that correct. All observables subscribe to their sources only on activation, and unsubscribe on deactivation. Thant conventions followed pretty strictly in Kefir.

I don't understand why Property nature would be inherited through observable transformations.

Take a property that represents a current value of a form input. Then suppose you want a property, that represents the value of input multiplied by 2, you do mult2 = inputValue.map(x => x * 2) and get such new property. Does that makes sense? Properties are designed to represent values changing over time, when you always can say that a currently the value is X: input values, current scroll position, current mouse position, latest server response etc.

Sorry for lots of comments. If you're not actively working on Kefir it must be a pain, sorry. I find the library to be excellent, though (and well documented), so I'm very keen to continue building on it.

No problem, but I'd appreciate if questions were shorter. I indeed don't have much time to answer them :)

---

One more time what happening in constant([1,2,3]).flatten():

constant([1,2,3]) creates a property that will emit [1,2,3] to the subscriber immediately (what we call a current value, and what marked as current in .log).

prop.flatten() will create an inactive property, with prop as the source.

When you subscribe to prop.flatten():

• it'll activate itself;
• subscribe to prop;
• get [1,2,3] from it;
• apply transformation to [1,2,3], which results in 3 values 1, 2, and 3;
• as a result property will get 3 values in the immediate onValue subscriber call — i.e. 3 current values;
• property will ignore first 2 current values, and save 3 as its current;
• property will call its subscriber with 3.

[cefn]

Brilliant detail and thanks. I can see the core ideas, definitely.

Currently it's guaranteed, that when you subscribe to a property, you'll get at most one such value.

The intention that a downstream onValue from an upstream Property would have to re-impose Property contracts (e.g. it would have to be a property, and would have to issue no more than one event in onValue), could be the issue here. This seems optional and potentially costly in terms of the surprising behaviours it introduces.

If you think of a Property as a behaviour explicitly requested at a single point in a pipeline, (i.e. a request to 'cache here'), rather than as an observable nature which continues through transformations and automatically reappears at later points in the same pipeline, then I think a lot of these issues go away.

At any later point where another 'onValue cache' is required, you can put one in with another toProperty(). Before our discussion (and my understanding more of your intent) I would have imagined the following would be required for a multiplied-by-two property (based on the idea of Property being a cache in a stream pipeline). I think it is more explicit and less surprising.

var singleProp = ....toProperty();
var mult2Stream = singleProp.map(function(value){ return value * 2;});
var mult2Prop = mult2Stream.toProperty();

For many purposes, the mult2Stream would end up with Property-like behaviour anyway, since when mult2Stream had its first onValue() call, then it would call onValue on singleProp, and get issued with its first event immediately. However, if you wanted a later onValue call to your mult2Stream to also re-trigger the last multiplied value, then you'd have to be explicit about making that a property instead, like mult2Prop.

FOR INTEREST

Any Kefir.stream() strategy can be expected to deliver a series of events triggered by an onValue, so it's hard to know what committing to the 'single-event' aspect of the contract really buys you.

Incidentally, multiple events triggered a downstream onValue() call is typical of scenarios in my project work, which may be why I'm hitting these issues a fair amount.

As an example of using Kefir.stream triggered by onValue, I have this fragment (tweaked for readability) in the library I'm building to allow subscribers to traverse all available topics now and in the future through a 'catchup' stream, dedicated to a single subscriber, but which falls back to the generic shared topic stream after traversing existing values, meaning they get up to date, and then handle later events in synchrony with other handlers against the shared stream...

    pastTopicAdded:function(){
        var tree = this;
        return Kefir.stream(function(emitter){
            _.forEach(Object.keys(tree.items), emitter.emit);
            emitter.end();
        });
    },
    catchupTopicAdded:function(){
        return Kefir.concat([this.pastTopicAdded(), this.topicAdded()]);
    },

[rpominov]

If each transformation would return a stream, you'll lose propagation of current value, as stream provide current value only to the first subscriber, which consumes it.

function getScroll() {
  return document.body.scrollTop
}

var scrollTopProp = Kefir.fromEvents(window, 'scroll', getScroll).toProperty(getScroll)

var plus1 = scrollTopProp.map(function(x){return x+1})

plus1.log('a')
plus1.log('b')

// a <value:current> 753
// b <value:current> 753

If map would return a stream the output would be:

// a <value:current> 753

[cefn]

Certainly if map returns a stream, the output would be as you suggest, but I'm confused why this is a bad thing. It seems trivial to reintroduce Property contract at any point you choose in the chain. This is most likely where I've missed something huge, if this contract can't be consistently delivered later in the chain then that would explain the design decision and if so, then sorry again for useless noise.

To explore my own assumptions, I implemented the 'mapStream' alternative in the example below, which does indeed create a single event as you warned (log 'b' produces nothing when you run the second commented construction of plus1).

But I found simply extending the construction of my 'mapStream' with an explicit toProperty() call provides exactly the same as the original, but more explicitly and enabling us to avoid the fallout from Property nature inheritance in every Observable transformation (especially given there's no toStream() call available in the API). Am I missing something which means that this isn't always so trivial to do?

var Kefir = require("kefir"),
    events = require("events");

var evtSource = new events.EventEmitter();
function getScroll() { return 399; }

var scrollTopProp = Kefir.fromEvents(evtSource, 'scroll', getScroll).toProperty(getScroll)

scrollTopProp.mapStream = function(fun){
    var thisStream = this;
    return Kefir.stream(function(emitter){
        var handler = function(event){
            if(event.type ==="value"){
                emitter.emit(fun(event.value));
            }
            else{
                emitter.emitEvent(event);
            }
        }
        thisStream.onAny(handler);
        return function(){
            thisStream.offAny(handler);
        }
    })
};

//var plus1 = scrollTopProp.map(function(x){return x+1}) //original with property nature contagion
//var plus1 = scrollTopProp.mapStream(function(x){return x+1}) //stream-oriented alternative
var plus1 = scrollTopProp.mapStream(function(x){return x+1}).toProperty(); //stream-oriented alternative with explicit property request - equivalent result to property contagion case

plus1.log('a')
plus1.log('b')

Results in...

a <value:current> 400
b <value:current> 400

Currently we seem stuck with Properties once they are Properties through whatever transformations (especially in the absence of toStream()).

I don't fully get why there's a cost to having a stream result as the default from every Observable transformation, and there certainly seems to be a number of costs to having "Property-nature contagion" as the default. Surprising and unexpected results seem to follow from the expectation that the result of e.g. flatten should always implicitly satisfy Property nature just because the upstream observable has Property nature. 'Always results in a stream' seems parsimonious and harmless as far as I can see, although any breaking change to an API is a little terrifying.

Sorry if the word contagion sounds really negative, but it's intended to be purely descriptive - it seems to capture the irreversible downstream Propertyness when using the standard API. Once you've had contact with a Property, then everything you touch becomes a Property. At least I didn't mention zombies :)

MY OWN EXAMPLE

In my case, I'm relying on Properties to cache the last value of a remotely-served list of ids. I was expecting to use diff and flatten to be able to generate streams of events of 'single ids added', and 'single ids removed' even when multiple ids were actually added or removed by a remote update. Such updates could come either before or after subscription and activation. Of course because I'm using a Property, (so that any previously retrieved list can be accessed instantaneously without a round-trip), I'm now concerned there will be unexpected side-effects in the behaviour of diff and flatten. Having the confidence that all transformations produce a realtime stream without a cache or Property contract enforcement (and assuming that the 'toProperty()' trick always works in case a cache is actually needed) would give me a lot of confidence to predict how the pipelines I'm creating would behave.

[rpominov]

Certainly if map returns a stream, the output would be as you suggest, but I'm confused why this is a bad thing. It seems trivial to reintroduce Property contract at any point you choose in the chain. This is most likely where I've missed something huge, if this contract can't be consistently delivered later in the chain then that would explain the design decision and if so, then sorry again for useless noise.

Yeah, this is the case. It can't be consistently delivered later. For example (supposed map returns a stream):

var prop1 = ...
var stream1 = prop.map(...)
var prop2 = stream1.toProperty()

stream1.onValue(f1) // will consume the current
prop2.onValue(f2) // won't get the current

Also I find, than in most cases I want to get a property automatically, so I'd have to do p.map(...).toProperty() all the time.

And returning same type after transformation aligns well with the algebraic data types semantics e.g., when we apply map to a functor, we get a functor of same type, same for flatMap and monads. And in Kefir both streams and properties can be counted as monads and functors.

Closing as this is more like a question issue, that can't be fully resolved. But feel free to continue discussion. Aslo see https://github.com/rpominov/kefir/issues/142

[rpominov]

Also #144

[rpominov]

Also #146 :)

[cefn]

I think the notion of Property we're working with may be two totally different concepts so when you say 'I want to get a property automatically' I'm probably unclear as to the contract you're expecting it to meet which can't be met by the later toProperty().

The contract I'm expecting is what I thought was summarised in the existing documentation - that it provides a cache of the last-served value.

So if the Property wasn't activated at the time a value was issued, then the value wasn't served, and hence it wasn't cached. I understood this was pretty common in cases of creating Properties already, (the toProperty() calls in my library are very often toProperty().onValue(_.noop) to force values to be served for this reason.

The example you describe where property behaviour can't be delivered later seems like a case of not being subscribed at the right time, which I understood was a common userland 'bug' and Kefir feature, as per e.g. https://github.com/rpominov/kefir/issues/125

The idea userland bugs like this should be solved through defaults in the Observables API (the 'current' value reaching forward through the pipeline) appears a challenge to consistency.

I am definitely missing something fundamental and should probably puzzle more before bothering you again as all the cases you describe as a problem (if Property contagion wasn't in place) seem absolutely the right behaviour to me which must mean I have the wrong model.