Repeat request until condition is met and return intermediate values

user_1675492 0
Viewed 8909

I am trying to write a (generic) function run<ID, ENTITY>(…): Observable<ENTITY> which takes the following arguments:

  • A function init: () => Observable<ID> which is an initializing request to start a backend process.
  • A function status: (id: ID) => Observable<ENTITY> which takes the generated ID and queries the status for it in the backend.
  • A function repeat: (status: ENTITY) => boolean which determines whether the status request must be repeated.
  • Two integer values initialDelay and repeatDelay.

So run should execute init, then wait for initialDelay seconds. From now on it should run status every repeatDelay seconds until repeat() returns false.

However, there are two important things that need to work:

  • repeatDelay should only be counted starting when status has emitted its value as to avoid race conditions if status takes longer than repeatDelay
  • The intermediate values emitted by the calls to status must also be emitted to the caller.

The following (not very pretty) version does everything except for the last thing I mentioned: it doesn't wait for the network response before retrying status.

run<ID, ENTITY>(…): Observable<ENTITY> {
    let finished = false;
    return init().mergeMap(id => {
        return Observable.timer(initialDelay, repeatDelay)
            .switchMap(() => {
                if (finished) return Observable.of(null);
                return status(id);
            })
            .takeWhile(response => {
                if (repeat(response)) return true;
                if (finished) return false;

                finished = true;
                return true;
            });
    });
}

My second version is this, which again works for all but one detail: the intermediate values of the status calls aren't emitted, but I do need them in the caller to show the progress:

run<ID, ENTITY>(…): Observable<ENTITY> {
    const loop = id => {
        return status(id).switchMap(response => {
            return repeat(response)
                ? Observable.timer(repeatDelay).switchMap(() => loop(id))
                : Observable.of(response);
        });
    };

    return init()
        .mergeMap(id => Observable.timer(initialDelay).switchMap(() => loop(id)));
}

Admittedly, the latter one also is a bit of a kludge. I'm sure rxjs can solve this problem in a much neater way (and, more importantly, solve it at all), but I can't seem to figure out how.

rxjs5 rxjs
3 Answers

The repeatWhen operator looks tempting on its own, but it only provides a void stream of onComplete notifications, so you can't decide to repeat based on the values without some outside help. Here, BehaviorSubject is probably your best bet:

function run(/* ... */): Observable<ENTITY> {
  const last_value$ = new BehaviorSubject();
  const delayed$ = init()
    .delay(initialDelay)
    .flatMap(id => 
      status(id).repeatWhen(completions => 
         completions.delay(repeatDelay)
                    .takeWhile(_ => repeat(last_value$.getValue()))
      )
    ).share();
  delayed$.subscribe(last_value$);
  return delayed$;
}

Here, repeatWhen only resubscribes to the cold source of requests if the last value from the last request is a status that signifies repeating.

Try the fiddle.

Caveat: there might be a race condition risk when repeatDelay is small between the notifier (the function passed to repeatWhen) executing and the BehaviorSubject receiving the corresponding status. For a given observer, we're guaranteed that all onNext notifications happen before an onCompleted notification, but here our BehaviorSubject and repeatWhen are separated. From a glance at the source, it looks like the onNext notifications are passed right through repeatWhen. I suspect the same is true with concat, but I'm not sure.

user_39255070

Update: Observable supports recursion natively with expand, also shown in @IngoBürk's answer. This lets us write the recursion even more concisely:

function run<ENTITY>(/* ... */): Observable<ENTITY> {
  return init().delay(initialDelay).flatMap(id =>
    status(id).expand(s => 
      repeat(s) ? Observable.of(null).delay(repeatDelay).flatMap(_ => status(id)) : Observable.empty()
    )
  )
}

Fiddle.

If recursion is acceptable, then you can do things more concisely still:

function run(/* ... */): Observable<ENTITY> {
  function recurse(id: number): Observable<ENTITY> {
    const status$ = status(id).share();
    const tail$ = status$.delay(repeatDelay)
                         .flatMap(status => repeat(status) ? recurse(id, repeatDelay) : Observable.empty());
    return status$.merge(tail$);
  }
  return init().delay(initialDelay).flatMap(id => recurse(id));
}

Try the fiddle.

user_39255070

So I've come up with this which appears to work. It uses expand to create an infinite sequence of retries and takeWhile to determine how long to take from it.

The takeWhile hack could be removed by writing a custom operator takeUntil taking a predicate function. This currently doesn't exist, see rxjs#2420.

Fiddle: https://jsfiddle.net/2mhcvnog/1/

run<ID, ENTITY>(...): Observable<ENTITY> {
    /* This little hack is needed to also emit the final item in the takeWhile() loop. */
    let finished = false;

    const delayStatus = (id, delay) => {
        return Observable.of(null)
            .delay(delay)
            .switchMap(() => status(id))
            .map(status => [id, status]);
    };

    return init()
        .mergeMap(id => delayStatus(id, initialDelay))
        .expand(([id]) => {
            if (finished) {
                return Observable.of([id, null]);
            }

            return delayStatus(id, repeatDelay);
        })
        .takeWhile(([_, status]) => {
            if (repeat(status)) {
                return true;
            }

            if (finished) {
                return false;
            }

            finished = true;
            return true;
        })
        .map(([_, status]) => status);
}
user_16754920
Related