Отдельные наблюдаемые значения по конкретному времени в RxJS

Question

Каким будет самый идиоматический способ получения значений Наблюдаемого на определенное количество времени? Например, допустим, у меня есть Observable, созданный из большого массива, и я хочу дать значение каждые 2 секунды. Является ли комбинация interval и selectMany лучшим способом?

Answer 1

Для вашего конкретного примера, идея заключается в том, чтобы отобразить значение каждого из массива наблюдаемой, что даст свой результат после задержки, а затем сцепить результирующий поток наблюдаемых:

var delayedStream = Rx.Observable 
    .fromArray([1, 2, 3, 4, 5]) 
    .map(function (value) { return Rx.Observable.return(value).delay(2000); }) 
    .concatAll(); 

Другие примеры могут действительно используйте timer или interval. Это просто зависит.

Например, если ваш массив действительно очень большой, то вышеизложенное вызовет достаточное количество давления в памяти (потому что он создает N наблюдаемых для действительно больших N). Вот альтернатива, которая использует interval лениво ходить массив:

var delayedStream = Rx.Observable 
    .interval(2000) 
    .take(reallyBigArray.length) // end the observable after it pulses N times 
    .map(function (i) { return reallyBigArray[i]; }); 

Это один даст следующее значение из массива каждые 2 секунды, пока она не итерированные по всему массиву.

Answer 2

В то время как ответ Брэндона получает суть идеи, вот версия, которая немедленно дает первый элемент, а затем помещает время между следующими элементами.

var delay = Rx.Observable.empty().delay(2000); 

var items = Rx.Observable.fromArray([1,2,3,4,5]) 
    .map(function (x) { 
    return Rx.Observable.return(x).concat(delay); // put some time after the item 
    }) 
    .concatAll(); 

Обновлен для нового RxJS:

var delay = Rx.Observable.empty().delay(2000); 

var items = Rx.Observable.fromArray([1,2,3,4,5]) 
    .concatMap(function (x) { 
    return Rx.Observable.of(x).concat(delay); // put some time after the item 
    }); 

Answer 3

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

var items = ['A', 'B', 'C']; 

Rx.Observable.zip(
    Rx.Observable.fromArray(items), 
    Rx.Observable.timer(2000, 2000), 
    function(item, i) { return item;} 
) 

Answer 4

Согласитесь, что zip - это чистый подход. Вот функция многократного использования, чтобы генерировать поток интервал для массива:

function yieldByInterval(items, time) { 
    return Rx.Observable.from(items).zip(
    Rx.Observable.interval(time), 
    function(item, index) { return item; } 
); 
} 

// test 
yieldByInterval(['A', 'B', 'C'], 2000) 
    .subscribe(console.log.bind(console)); 

Это основано на farincz's answer, но немного короче, используя .zip как метод экземпляра.

Кроме того, я использовал Rx.Observable.from(), потому что Rx.Observable.fromArray() является deprecated.

Answer 5

Для RxJS 5:

Rx.Observable.from([1, 2, 3, 4, 5]) 
    .zip(Rx.Observable.timer(0, 2000), x => x) 
    .subscribe(x => console.log(x)); 

Answer 6

Поскольку это не было упомянуто, я думаю, что в сочетании с concatMapdelay довольно читаемым.

Rx.Observable.fromArray([1, 2, 3, 4, 5]) 
    .concatMap(x => Rx.Observable.of(x).delay(1000)); 

https://codepen.io/jmendes/pen/EwaPzw См