буферный withLatestFrom

Question

мне нужно что-то похожее на withLatestFrom что соответствует диаграмме ниже:

---------A-----------------B-- 
-1-2-3------4------5-6-7-8---- 

--------[A,[123]]---------------[B,[45678]] 

Есть ли способ, чтобы реализовать такое поведение с RxJS?

Answer 1

Прошу прощения, что я не кодовый код javascript, но это работает в .NET. Надеюсь, вы сможете перевести.

var xs = new Subject<string>(); 
var ys = new Subject<string>(); 

var qs = 
    xs.Publish(pxs => 
     ys.Buffer(() => pxs) 
      .Zip(pxs, (numbers, letter) => new { letter, numbers })); 

.Publish(pxs => оператор принимает один наблюдаемый, выписывает ему только один раз и акций, что подписка в лямбда. Он предотвращает множественные подписки источника и синхронизирует производство значений от pxs в пределах лямбда.

ys.Buffer(() => pxs принимает все значения ys и преобразует значения в ряд списков, разбитых в точке, когда pxs производит значения.

И наконец, .Zip(...) принимает значения от pxs и связывает их со списками, произведенными .Buffer(...).

qs.Subscribe(q => Console.WriteLine(q)); 

ys.OnNext("1"); 
ys.OnNext("2"); 
ys.OnNext("3"); 
xs.OnNext("A"); 
ys.OnNext("4"); 
ys.OnNext("5"); 
ys.OnNext("6"); 
ys.OnNext("7"); 
ys.OnNext("8"); 
xs.OnNext("B"); 

ys.OnCompleted(); 
xs.OnCompleted(); 

Он производит:

---

Вот переведенный Javascript версия:

var qs = xs.publish(
    pxs => 
     ys.buffer(() => pxs).zip(pxs, (numbers, letter) => [ letter, numbers ]) 
); 

Answer 2

This, похоже, работает. Он основан на операторе sample, но есть и другие способы сделать то же самое.

var ta_message = document.getElementById('ta_message'); 
var ta_result = document.getElementById('ta_result'); 

function emits (who, who_) {return function (x) { 
who.innerHTML = [who.innerHTML, who_ + " emits " + JSON.stringify(x)].join("\n"); 
};} 

function pushValue (arr, el){arr.push(el); return arr;} 

var sourceCounter = 0; 
var samplerCounter = 0; 
var source$ = Rx.Observable 
    .fromEvent(document.getElementById('source'), 'click') 
    .map(function(){return sourceCounter++;}) 
    .do(emits(ta_source, 'source')) 
    .share(); 
var sampler$ = Rx.Observable 
    .fromEvent(document.getElementById('sampler'), 'click') 
    .map(function(){return String.fromCharCode(65+samplerCounter++);}) 
    .do(emits(ta_sampler, 'sampler')) 
    .share(); 

var bufferedSource$ = sampler$ 
    .startWith('0') 
    .flatMapLatest(function (x){ 
    console.log("x",x) 
    return source$.scan(pushValue,[]) 
    }) 
    .do(emits(ta_sampler, 'bufferedSource$')); 

var sampledSource$ = bufferedSource$ 
    .sample(sampler$) 
    .do(emits(ta_result, 'sampledSource')) 
    .withLatestFrom(sampler$, 
     function (bufferedValues, samplerValue){ 
      return "" +samplerValue + bufferedValues; 
     }) 
    .do(emits(ta_result, 'result')) 

//sampler$.connect(); 
sampledSource$.subscribe(function(){});