Angular 2 - How to change the interval of an RxJS Observable

Question

I'm using rxJS Observable Interval to refresh the data being fetched. I can't figure out the way to change the interval setting. I've seen something about using the Subject class provided by rxJS but I can't manage to get it to work.

I provided an simplified example in this plunk

In the AppComponent I have this method.

getTime() {
        this.timeService.getTime(this.refreshInterval)
          .subscribe(t => {
            this.currentTime = t;
            console.log('Refresh interval is: ' + this.refreshInterval);
          }
        );
}

And in the service component I currently have this code.

getTime(refreshInterval: number) {
  return Observable.interval(refreshInterval)
        .startWith(0)
        .map((res: any) => this.getDate())
        .catch(this.handleError)
}

Can someone perhaps provide me with a working example it would be great!

Answer 1

You do not need to destroy and recreate the entire Observable stream to change the refreshInterval. You only need to renew the part of the stream that depends on the changing interval.

First simplify your service's getTime() so it's not in charge of determining the frequency of outputs. All it does is return the time:

getTime() { return (new Date()).toString(); }

Now the calling code will determine the schedule. Just 3 simple steps:

1. A source function that adjusts to the desired interval:

/** Observable waits for the current interval, then emits once */
refreshObs() {return Observable.timer(this.refreshInterval)}

2. An observable chain that uses the repeat operator to continuously re-execute the stream:

getTime$ = Observable.of(null)
            .switchMap(e=>this.refreshObs()) // wait for interval, then emit
            .map(() => this.timeService.getTime()) // get new time
            .repeat(); // start over

3. A subscription to trigger the whole thing:

ngOnInit(){
    this.getTime$.subscribe(t => {
        this.currentTime = t;
        console.log('refresh interval = '+this.refreshInterval);
    });
}

This works because refreshObs() returns a new observable each time the stream is repeated, and that new observable will wait according to the currently set interval before emitting.

Live demo