RxJs - повторить все события после каждого всплеска событий

Question

Как вы это делаете? RxJs по-прежнему остается загадкой для меня.

Я пытался вещи, как:

filterChanges 
    .delay(400) 
    .replay() 
    .reduce(function(acc,x) { return acc.concat(x) }, []) 
    .subscribe(function(changes) { 
     console.log(changes); 
     ... 

Или

filterChanges.subscribe(function() { 
    filterChanges.aggregate(function(changes) { 
      ... 

Я действительно очень потерял здесь. Поведение, которое я хочу: некоторые действия пользователя приводят к нескольким изменениям фильтра. Я не хочу обрабатывать их по одному, но только когда завершение изменений завершено. Но когда я их обрабатываю, я хочу, чтобы все изменения фильтра начинались с начала потока.

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

1. Подпишитесь на рассылку событий.
2. Подпишитесь на конец всплеска событий и запишите все события из начала потока.

Answer 1

Если я правильно понял, используйте buffer в сочетании с debounce. Буфер предоставит вам список событий, произошедших во время окна буфера, и debounce определит, когда окно буфера должно быть закрыто, так вы можете получить все события в рывке. Чтобы получить все события с самого начала потока, вы получите scan, чтобы скопировать все списки из буферов.

var hotFilterChanges = filterChanges.share(); 

hotFilterChanges 
    .buffer(hotFilterChanges.debounce(200)) 
    .scan((acc, curr) => acc.concat(curr)) 
    .subscribe(x => console.log(x)); 

Мы используем .share() сделать Наблюдаемые быть "горячим". Это необходимо, так что hotFilterChange.buffer(...) и hotFilterChanges.debounce(200) относятся к точному исполнению.

See the JSBin.