Observable.flatMapLatest: Отменить на более глубоком уровне?

Question

Я экспериментирую с реактивным программированием в первый раз, используя bacon.js, и ударяю по нечетному взаимодействию, которое меня раздражает. Я придумал рабочую версию этой цепочки событий, но хакерские манипуляции с дополнительными потоками делают ее отличной от остальной части кода - как вы увидите.

Я нашел FRP/Bacon.js для создания довольно чистого кода, поэтому мне кажется, что я просто не хватает некоторых очевидных операторов. К проблеме, не так ли?

Несколько источников рекомендуют использовать Observable.flatMapLatest для отбрасывания устаревших событий. Предположим, мы хотим завершить операцию ввода-вывода для каждого события во входном потоке. На следующей диаграмме 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. Если это не ясно теперь, это то, что я хочу сделать:

   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. Но не только это выглядит очень взломанным, он добавляет timestamp/id как требование для запросов, так что мы можем сортировать его позже.

Я думаю, мой вопрос в том, что: Что мне не хватает? Есть лучший способ сделать это? Как я могу использовать takeUntil каждый из потоков событий (запроса) flatMap вместо внешнего (ввода)?

Любая помощь приветствуется, спасибо!

Answer 1

Вот решение, которое использует максимальный идентификатор и сканирования:

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