Nested onValue on a property

[fxmouthuy]

Hello,

We encountered a strange behavior of Kefir while nesting onValue calls on a Kefir property. See an example below:

var e = Kefir.sequentially(500,[1,2,3,4]);

var p = e.toProperty();

p.log();

function fn(v1) {
    p.onValue(function(v2) { console.log(v1, v2); });
}

p.onValue(fn);

Actual result

[sequentially.toProperty] <value> 1
[sequentially.toProperty] <value> 2
2 1
1 2
[sequentially.toProperty] <value> 3
3 2
1 3
2 3
[sequentially.toProperty] <value> 4
4 3
1 4
2 4
3 4
[sequentially.toProperty] <end>

Expected result

[sequentially.toProperty] <value> 1
1 1
[sequentially.toProperty] <value> 2
2 2
1 2
[sequentially.toProperty] <value> 3
3 3
1 3
2 3
[sequentially.toProperty] <value> 4
4 4
1 4
2 4
3 4
[sequentially.toProperty] <end>

Regards,

[rpominov]

Hi, @fxmouthuy. If we subscribe in an .onValue handler, the new listener misses the train of the current event dispatch. Also if the property has a cached value, a subscriber is called immediately with that value. And the cached value updates after current listeners are notified. Taking that into account the result seems reasonable to me.

Although maybe this part isn't right indeed:

And the cached value updates after current listeners are notified.

If we fix this, the result will be almost as your expected result, except for the first 1, 1 log — it still won't be present.

This is a breaking change, that we should consider making in 3.0.0.

[rpominov]

Also, I'd recommend you to consider doing something like the following instead of nested subscribes:

var e = Kefir.sequentially(500,[1,2,3,4])
var p = e.toProperty()

function fn(arr) {
  var last = arr[arr.length - 1]
  return arr.map(function(x) {
    return [x, last]
  })
}

p.log()
p.slidingWindow(1000000).map(fn).flatten().log('')

[fxmouthuy]

Thanks for the answer.

One of my colleague forked and fixed the following part:

And the cached value updates after current listeners are notified.

And it does fix both issues.

Regards

[olivierguerriat]

Hi, @fxmouthuy's colleague here.

My approach is probably naive, but it does produce the expected results and all tests are passing. Are you aware of any situations where this modification might cause some problems?

[rpominov]

Hi, @olivierguerriat. Yeah, the change probably just it — move the line to before dispatching. But strictly speaking it's a breaking change, and introducing this change in a not major release may break someone's code, if they rely on that behavior.

Also I have a feeling that it may introduce some unexpected side effects. But it's just a gut feeling :smile: Did you try running tests, btw?

[olivierguerriat]

I was wondering if it was breaking any documented behaviours, but I agree it is a breaking change anyway.

I did run npm test which reported 960 tests, 1289 assertions, 0 failures, 0 skipped, but I didn't use my version in a "real life" project yet.

[Macil]

I just ran into what I think is this same issue.

var Kefir = require('kefir');

var seq = Kefir.sequentially(1000, [1, 2]);
var odds = seq.filter(x => x%2===1);
var evens = seq.filter(x => x%2===0);
var lastWasOdd = odds.awaiting(evens);
lastWasOdd.log('lastWasOdd');

setTimeout(() => {
  // Wait until lastWasOdd is true or 5 seconds have passed, then wait until
  // lastWasOdd is false.
  Kefir.merge([
      lastWasOdd.filter(x => x),
      Kefir.later(5000)
    ])
    .take(1)
    .flatMap(() =>
      lastWasOdd.filter(x => !x).take(1)
    )
    .onValue(value => {
      console.log('done', value);
    });
}, 0 /*3000*/);

The above gives the following output incorrectly:

lastWasOdd <value:current> false
lastWasOdd <value> true
done false
lastWasOdd <value> false
lastWasOdd <end>

I worked around it for now by changing

   Kefir.merge([
-      lastWasOdd.filter(x => x),
+      lastWasOdd.filter(x => x).delay(0),
       Kefir.later(5000)

[rpominov]

@AgentME Does it work as expected if you swap lines in Kefir source like here?

[Macil]

@rpominov That change fixes it.

[rpominov]

Reduced it to this:

var prop = Kefir.sequentially(1000, [true, false]).toProperty(function() {return false})
prop.log('[subscribed earlier]')

firstTrue = prop.filter(function(x) {return x}).take(1)
firstFalse = prop.filter(function(x) {return !x}).take(1)
firstTrue.flatMap(function() {return firstFalse}).log('[subscribed later]')

[subscribed earlier] <value:current> false
[subscribed earlier] <value> true
[subscribed later] <value> false
[subscribed later] <end>
[subscribed earlier] <value> false
[subscribed earlier] <end>

And thought that the problem was that, the more later subscriber gets false earlier than the more earlier one. But actually problem has another dimension — these are different false's:

var prop = Kefir
 .sequentially(1000, [{x: true, meta: 1}, {x: false, meta: 2}])
 .toProperty(function() {return {x: false, meta: 3}})
prop.log('[subscribed earlier]')

firstTrue = prop.filter(function(x) {return x.x}).take(1)
firstFalse = prop.filter(function(x) {return !x.x}).take(1)
firstTrue.flatMap(function() {return firstFalse}).log('[subscribed later]')

[subscribed earlier] <value:current> Object {x: false, meta: 3}
[subscribed earlier] <value> Object {x: true, meta: 1}
[subscribed later] <value> Object {x: false, meta: 3}
[subscribed later] <end>
[subscribed earlier] <value> Object {x: false, meta: 2}
[subscribed earlier] <end>

Now it's clear to me what is happening, and I can confirm this is the same issue: when we subscribe as [subscribed later] to firstFalse (via flatMap, but still to firstFalse basically) in response to true value from firstTrue, we still immediately get false from prop, as it still technically the "current" value of the property, although true already being dispatching.

---

This new case makes me want to fix this faster and to wait at the same time. Want to fix because it causes problems in more legitimate cases (than subscribing to the same property in onValue). But this case also shows, that the fix may indeed introduce unexpected side effects I was worried about.

Need to think more, perhaps we should cross fingers and fix.

[rpominov]

A PR into 3.0 branch is welcomed (see #138)