How do I create a timeout?


This inaugural post is in the format I hope to follow going forward: a question, a discussion of the question, a solution or several solutions (each linked to a demo), and a teaser for the next question.  The point of the teaser is to stimulate some discussion of possible solutions from readers, but I will be sincerely shocked if the first few posts even get readers, let alone discussion from them.  The assumption is that every reader knows the basics of RxJs.  On the left is a link to the documentation for RxJs; that may clarify some things.

If the question is, “How do I create an Operator that allows its source Observable to run for given time and then ends it?”, that’s pretty simple:

function timeout(n) {
  return takeUntil(timer(n));
}

Code is here.

But the question is, “How do I create an Operator that copies its source Observable, until the source fails to emit for given time?”, it’s somewhat more complicated.

The key insight is that switchMap() will unsubscribe from one inner Observable when the source emits the next value.  So:

function timeoutE(n) {
  const timeoutSentinel = Symbol();
  return pipe(
    switchMap(v => timer(n).pipe(
      mapTo(timeoutSentinel),
      startWith(v))),
    takeWhile(v => v !== timeoutSentinel),
  );             
}

Code is here.

I am not 100% happy with this solution.  The use of a sentinel is... iffy.  I am open to alternatives.

Next time: “How do I check if an Observable is still running?”

Comments

  1. Travis HeppeNovember 2, 2018 at 5:52 PM

    This seems like a special case of the "take until something external happens" problem. What's "iffy" about using a sentinal?

    ReplyDelete
    Replies
    1. AnonymousNovember 2, 2018 at 7:10 PM

      I'm just not happy with the type. If this were TypeScript, the type of the switchMap() would by T | Symbol. It might be possible to mark the takeWhile so the compiler realizes it's effectively a type-guard, but I dunno, it bugs me a little. Of course, I don't have a better idea...

      Delete

Post a Comment

Popular posts from this blog

How do I get RxViz to display several streams at once?

Image
If you have been reading these blog entries, you have seen some of the things RxViz can do — but perhaps, you have tried to display an array of Observables and been frustrated by this error:   Last expression must be an Observable Well, darn. An  Observable.  One.  I want to put a bunch on the screen, lined up top to bottom. Consider, though, higher-order Observables; that is, Observables that emit other Observables.  Like this ship that carries other ships: RxViz does a beautiful  job of displaying higher-order Observables.  See here for good example. And you might hope that therefore you could display several Observables just by packing them up in a single containing Observable, like this: of(p0, p1, p2) It almost works.  All three Observables are displayed, just not in any predictable order.  Sometimes,   p2  is at the bottom; sometimes at the top.  Why RxViz  has this problem I ...
Read more

How can I tell if an Observable is running or not?

You should think of Observables like pipes: you cannot tell what is happening inside of them by looking at the outside. The question should be rephrased as, given an Observable, how can I create another Observable that emits true when a the first Observable is still running and false when it is not. First, consider the following operator: function activity() {   return pipe(     mapTo(0),     startWith(1),     endWith(-1),     catchError(() => of(-1)),   ); } This will emit 1, then 0 every time the source Observable emits, and then -1 when it finalizes, successfully or not. In marble terms: source: ---a--b--c---. result  1--0--0--0--(-1). Why that weird protocol? Well, mostly because I have a theory that activity() will be useful later, but you can see that function runningSum() {   return scan((acc, v) => acc + v, 0); } source.pipe(runningSum()); will emit 1 whenever the source Observable...
Read more