javascript - Observable.flatMap最新: Cancel at a deeper level?

Question

我第一次尝试响应式编程，使用 bacon.js ，并遇到了一个让我很烦的奇怪的互动。我想出了这个事件链的一个工作版本，但是 hackish 的额外流操作使它从其余代码中脱颖而出——如您所见。

我发现 FRP/Bacon.js 生成了一些非常干净的代码，所以感觉我只是缺少一些明显的运算符。对于这个问题，我们好吗？

多个来源建议使用 Observable.flatMapLatest丢弃过时的事件。假设我们要为一个输入流中的每个事件完成一个IO操作。在下图中，i表示输入流事件，rN s 代表每个请求流的事件和 x es 标记请求流被取消的时间点。这些很可能是单事件流，因此 .意味着流实际上在事件之后关闭(如果之前没有取消!)。

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

这对于几个用例来说完全没问题，但要注意 r3 : 它永远不会被接受，即使它在 r4 之前仍然是一个有效的响应到达的。在最坏的情况下，很多请求都被浪费了:

             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 研究也很有值(value)。如果现在还不清楚，这就是我想要做的:

             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一:

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 .但这不仅看起来非常 hacky，它还添加了时间戳/id 作为请求的要求，以便我们稍后对其进行排序。

我想我的问题是:我错过了什么？有一个更好的方法吗？我怎样才能takeUntil每个 flatMap事件流(请求)而不是外部事件流(输入)？

感谢任何帮助，谢谢!

Answer 1

这是一个使用最大 ID 和扫描的解决方案:

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