RxJS emit array items over time?

Three ways to do it, with RxJS version 6 :

1. Using concatMap

import { from, of, pipe } from 'rxjs'; import { concatMap, delay } from 'rxjs/operators'; const array = [1, 2, 3, 4, 5]; from(array) .pipe( concatMap(val => of(val).pipe(delay(1000))), ) .subscribe(console.log); 

2. Using zip and interval

import { from, pipe, interval } from 'rxjs'; import { delay, zip} from 'rxjs/operators'; const array = [1, 2, 3, 4, 5]; from(array) .pipe( zip(interval(1000), (a, b) => a), ) .subscribe(console.log); 

3. Using interval as source

import { interval, pipe } from 'rxjs'; import { map, take } from 'rxjs/operators'; const array = [1, 2, 3, 4, 5]; interval(1000) .pipe( take(array.length), map(i => array[i]) ) .subscribe(console.log); 

This is how I would do it:

var fruits = ['apple', 'orange', 'banana', 'apple']; var observable = Rx.Observable.interval(1000).take(fruits.length).map(t => fruits[t]); observable.subscribe(t => { console.log(t); document.body.appendChild(document.createTextNode(t + ', ')); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/2.5.2/rx.all.min.js"></script>

var arrayList = [1,2,3,4,5]; var source = Rx.Observable .interval(500/* ms */) .timeInterval() .take(arrayList.length); source.subscribe(function(idx){ console.log(arrayList[idx]); //or document.write or whatever needed }); 

Pretty late but a simpler solution would be :

const arr = ["Hi,", "how", "may", "I", "help", "you?"]; Rx.Observable.interval(500) .takeWhile(_ => _ < arr.length) .map(_ => arr[_]) .subscribe(_ => console.log(_)) 

I find Weichhold technique to be the best but that it would gain in clarity of intent by extracting the zipped value outside of the zip:

// assume some input stream of values: var inputs = Obs.of(1.2, 2.3, 3.4, 4.5, 5.6, 6.7, 7.8); // emit each value from stream at a given interval: var events = Obs.zip(inputs, Obs.interval(1000)) .map(val => val[0]) .forEach(console.log); 

If you want to release samples over time, you can do something like this

const observable = interval(100).pipe( scan((acc, value) => [value, ...acc], []), sampleTime(10000), map((acc) => acc[0]) ); 

I had a little different requirement, my array kept updating over time too. So basically I had to implement a queue which I can dequeue at a regular interval, but I didn't want to use an Interval.

If somebody has a need for something like this then probably this solution can help:

I have a function createQueue() that takes the array as an input and returns an Observable which we subscribe for listening to events from the Array at a regular interval. The function also modifies the 'push()' method of the passes array so that whenever any item is pushed in the array, the Observable would emit.

createQueue(queue: string[]) { return Observable.create((obs: Observer<void>) => { const arrayPush = queue.push; queue.push = (data: string) => { const returnVal = arrayPush.call(queue, data); obs.next(); return returnVal; } }).pipe(switchMap(() => { return from([...queue]) .pipe( concatMap(val => of(val) .pipe(delay(1000))) ); }), tap(_ => queue.shift())) } 

Lets say that the array is: taskQueue = [];

So, we need to pass it to the above function and subscribe to it.

 createQueue(taskQueue).subscribe((data) => { console.log('Data from queue => ', data); }); 

Now, every time we do taskQueue.push('<something here>'), the subscription will trigger after a delay of "1000ms".

Please note: we should not be assigning a new array to the taskQueue after createQueue() has been called, or else we will loose the modified push().

Here is a dummy example for the above implementation: Test Example