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

Ответ 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 секунды, пока оно не повторится по всему массиву.

Ответ 2

Я думаю, что использование zip дает лучший и читаемый код, все еще используя только 3 наблюдаемых.

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

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

Ответ 3

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

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
  });

Ответ 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, но немного короче, используя .zip как метод экземпляра.

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

Ответ 5

Для RxJS 5:

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

Ответ 6

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

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

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