Observable.flatMapLatest: Undo at a deeper level?

Question

I am experimenting with reactive programming for the first time using bacon.js and hit the odd interaction annoying me. I came up with a working version of this chain of events, but hacking manipulations with additional threads make it different from the rest of the code - as you will see.

I found FRP / Bacon.js to create fairly clean code, so it seems to me that I'm just missing some obvious operators. To the problem, will we?

Several sources recommend using Observable.flatMapLatestto remove obsolete events. Suppose we want to complete an I / O operation for each event in the input stream. The following diagram irepresents the events of the input stream, rNrepresents the events of each of the request flows, and xes indicate the point in time when the request flow was canceled. These are most likely streams with a single event, therefore, it .means that the stream is actually closed after the event (if it has not been canceled before!).

             input stream:  i──i────────i───i─────────>   
                  request:  │  │        │   └────r4.──>   
                  request:  │  │        └───x─r3.─────>   
                  request:  │  └──r2.───x─────────────>   
                  request:  └──x─────r1.──────────────>   
                                                        + 
i.flatMapLatest(requests):  ──────r2─────────────r4───>   

This is perfectly normal for a few use cases, but pay attention to r3: it is never accepted, although it was still a valid answer until it reaches r4. In the worst case, many requests disappear:

             input stream:  i──i──i───i──i──────>   
                  request:  │  │  │   └──x─r4.──>   
                  request:  │  │  └───x─r3.─────>   
                  request:  │  └──x─r2.─────────>   
                  request:  └──x─────r1.────────>   
                                                  + 
i.flatMapLatest(requests):  ────────────────────>   

, throttle debounce , () . , ( 0,5 -5 +) .

, , , FRP. , , :

             input stream:  i──i────────i───i─────────>   
                  request:  │  │        │   └────r4.──>   
                  request:  │  │        └─────r3.─x───>   
                  request:  │  └──r2.──────────x──────>   
                  request:  └──────x─r1.──────────────>   
                                                        + 
i.flatMap_What?(requests):  ──────r2──────────r3─r4───>   

flatMapLatest one:

var wrong = function (interval, max_duration) {
    var last_id = 0,
        interval = interval || 1000,
        max_duration = max_duration || 5000;

    Bacon.interval(interval, 'bang').log('interval:')
        .flatMapLatest(function () {
            return Bacon.later(Math.random()*max_duration, last_id++).log(' request:');
        })
        .log('accepted: %s **********');
};

var right = function (interval, max_duration) {
    var last_id = 0,
        interval = interval || 1000,
        max_duration = max_duration || 5000;

    Bacon.interval(interval, 'bang').log('interval:')
        .flatMap(function () {
            return Bacon.later(Math.random()*max_duration, last_id++).log(' request:');
        })
        .diff(-1, function (a, b) { return b > a ? b : -1 })
        .filter(function (id) { return id !== -1 })
        .log('accepted: %s **********');    
};

wrong , , right flatMap diff filter. , timestamp/id , .

, , : ? ? takeUntil flatMap () ()?

, !

Answer 1

, max id scan:

var right = function (interval, max_duration) {
    var bang_id = 0, last_id = 0,
        interval = interval || 1000,
        max_duration = max_duration || 5000;

    Bacon.interval(interval, 'bang')
        .map(function() { return 'Bang'+(bang_id++) })
        .flatMap(function (val) {
            return Bacon.later(Math.random()*max_duration, {value: val, id: last_id++});
        })
        .scan({id: -1, value: null}, function(memo, v) {
          if(v.id <= memo.id) {
            return memo;
          }
          return v;
        })
        .changes()
        .skipDuplicates(function(a, b) { return a.id === b.id; })
        .map('.value')
        .log();    
};

http://jsbin.com/doreqa/3/edit?js,console