rxjs: perform some action regulary with specific delay in between

Question

Client applications sends request to server, that could potentially take long to complete. Once request is finished or failed, client should wait some period of time (i.e. 10 seconds) and then again send the request.

Current working solution is this:

appRequest = new Subject();

ngOnInit(): void {
  this.appRequest.delay(10000).subscribe(() => this.refresh());
  this.refresh();
}

refresh() {
  this.api.getApplications().subscribe(a => {
      this.updateApplications(a);
      this.appRequest.next();
    },() => this.appRequest.next()
  );
}

Is there a more elegant solution for this?

EDIT:

I could use timer with regular intervals but I don't want to send new request unless previous request has finished. Only after previous request has finished, I want to wait 10 seconds and do send request again. This should repeat indefinitely.

getApplications() function is generated by swagger and it internally uses angular's http client library. Current observation is that unless you subscribe to Observable returned by getApplications(), it will not send request to server.

Cagey

Answer 1

The repeatWhen() operator seems designed for this, but there is a lack of examples and docs in the rxjs flavor.

Here's a doc for RxJava (description also applies to RxJs) RxJava's repeatWhen and retryWhen, explained.

Uses
Poll for data periodically using repeatWhen + delay:
source.repeatWhen(completed => completed.delay(5000))

Your version might be

stopRequesting = new Subject();

ngOnInit() {
  this.api.getApplications()
    .repeatWhen(completed => completed.delay(10000))
    .takeUntil(stopRequesting)
    .subscribe(a => this.updateApplications(a))
} 

ngOnDestroy() {
  this.stopRequesting.next(true);
}

Demo

// log to html output
log = function(x) { document.write(x + "<br />"); };

const stop = new Rx.Subject();

Rx.Observable.interval(500)
  .take(2)
  .repeatWhen(completed => completed.delay(1000))
  .takeUntil(stop)
  .subscribe(
    x => log(`Next: ${x}`),
    err => log(`Error: ${err}`),
    () => log('Completed')
  );

setTimeout(() => stop.next(true), 10000)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"></script>

Anikaanil

Answer 2

You can use .expand() to recursively call your function.

var recursiveObservable = this.api.getApplications().delay(10000);

//recursively call
recursiveObservable
    .expand((res) => {
        //can do something with your res if you want to.
        return recursiveObservable;
    })
    .subscribe();

The above solution will:

1. Call this.api.getApplications()
2. delay for 10000 ms
3. Repeat step 1 and 2

Note that as are all recursive functions, .expand() will repeat itself indefinitely until you actually supply it with a terminating condition. For example you can specify how many times you want it to repeat, by using take operator:

recursiveObservable
    .expand((res) => {
        //can do something with your res if you want to.
        return recursiveObservable;
    })
    .take(10) //stop after 10 repetitions
    .subscribe()

Or the terminal condition can just be based on a boolean:

var shouldContinue = true;

recursiveObservable
    .expand((res) => {
        //you can modify shouldContinue based on your res results.
        return shouldContinue? recursiveObservable : Observable.empty();
    }) // will terminate when shouldContinue is false
    .subscribe()

Burp

This looks mind blowing, but unfortunately it does not work. Maybe I forgot to mention that getApplications() function is generated by swagger and it uses angular's http client library. Unless I subscribe to observable provided by getApplications() it does not even send request to server. – Cagey

@Cagey yes, you need to subscribe() to the observable as you normally would, after the .expand operator – Burp

Answer 3

Solution with repeat and delay looks for me more understandable:

unsubscribe = new Subject();

ngOnInit() {
  this.api.getApplications()
    .pipe(
        delay(10000),
        repeat(),
        takeUntil(this.unsubscribe))
    .subscribe(a => this.updateApplications(a))
} 

ngOnDestroy() {
  this.unsubscribe.next(true);
}

Outland

Answer 4

You're looking for the interval operator.

From the link above:

// emit value in sequence every 1 second
const source = Rx.Observable.interval(1000);
// output: 0,1,2,3,4,5....
const subscribe = source.subscribe(val => console.log(val));

Minardi

Problem with this approach is that it does not wait for previous request to finish. I need it to wait for previous request to finish, then wait 10 seconds and then make new request. – Cagey

Answer 5

This solution still use Subject (and more than once) but it doesn't need the other functions to know or call refresh():

const retry$ = new Subject();
const autoRetryRequest$ = Observable.of('Kick start')
  .merge(retry$)
  .mergeMap(() => {
    const next$ = new Subject()

    this.api.getApplications().delay(10000).subscribe(
      value => next$.next(value),
      () => retry$.next(),
      () => retry$.next()
    );

    return next$;
  })
  .subscribe((data) => { this.updateApplications(data); });

Note:

• Remember to use takeUntil, unsubscribe,... to stop it. Preferably takeUntil just before the mergeMap.
• The expand solution is very elegant and should be considered if possible. But in your case, it doesn't fit because the request must be repeat on error, and it will blow if your request returns multiple value. Will need special handling on those cases for it to works properly.

Francklin