Feature Request: catchComplete(Function) operator

[Trequetrum]

Feature Request

Like catchError it would be nice if there were a native way to catch a complete stream and replace it with another stream. This lets you chain a new observable into the completion of a previous observable.

Here is how I've implemented it in the past.

function catchComplete<T, R>(fn: () => Observable<R>): OperatorFunction<T, T|R>{
  return s => new Observable(observer => {
    let iSub: Subscription;
    const oSub = s.subscribe({
      next: x => observer.next(x),
      error: err => observer.error(err),
      complete: () => {
        iSub = fn().subscribe(observer);
      }
    });
    return { unsubscribe: () => {
      oSub?.unsubscribe();
      iSub?.unsubscribe();
    }}
  });
}

or

function catchComplete<T, R>(fn: () => Observable<R>): OperatorFunction<T, T|R> {
  return s => concat(s, defer(fn));
}

---

Here is a simple example of this in use:

from([1,2,3,4,5]).pipe(
  catchComplete(() => from([5,4,3,2,1]))
).subscribe(console.log);
// Output: 1 2 3 4 5 5 4 3 2 1

Alternatives do exist,

Generally:

stream1.pipe(
  catchComplete(() => stream2)
)

Can be re-written as:

concat(
  stream1, 
  defer(() => stream2)
)

---

I would argue that catchComplete is more ergonomic and easier to understand than the equivalent use of concat

Consider the following two snippets of code:

stream1.pipe(
  mergeMap(this.serviceCall),
  catchComplete(() => of("this.finalValue"))
  map(msg => msg + "!")
)

vs

concat(
  stream1.pipe(
    mergeMap(this.serviceCall)
  ),
  defer(() => of("this.finalValue"))
).pipe(
  map(msg => msg + "!")
)

[benlesh]

I think given how trivial it was for you to compose this yourself with:

function catchComplete<T, R>(fn: () => Observable<R>): OperatorFunction<T, T|R> {
  return s => concat(s, defer(fn));
}

That's probably your answer.

If we add this as a feature to RxJS, then we're on the hook for maintaining it nearly indefinitely. Meanwhile, we've provided enough tools for people to roll this themselves with a couple of lines of code.

[Trequetrum]

Fair enough!

I'd argue that operators like pluck, mapTo, 'timestamp`, and a few others find themselves in the same position. They're useful but they're also easily composed with a line or two.

I find it strange that of the three types of "events" an RxJS observable emits (next value, complete, & error), only one of them is lacking just about any native operators that deal with the emission explicitly. endWith is the only one that jumps out at me.

Either way, you're not wrong, We roll out our own logging operator(s) and can do the same with something like catchComplete, it's not really a problem at all.

The question is: "What is idiomatic RxJS?". I wouldn't expect an Observable library to replace/match whatever logging utilities I use. It's easy enough to interop anyhow. With the current state of affairs you can react to values, errors, and (with finalize) errors/completion/unsubscription - but not really to just completion. Unlike logging, that's something I would expect of an Observable library like RxJS.

"C'est la vie"! :)

[kwonoj]

I'd argue that operators like pluck, mapTo, 'timestamp`, and a few others find themselves in the same position

this is very valid argument and fyi, for those reason pluck is planned to be deprecated and we're continously evaluting other chances as well.

[cartant]

Aside from the fact that the implementation is trivial, I think the name is misleading. I would expect an operator with 'catch' in its name to have something to do with error notifications. This is unrelated to errors; performing concatenation instead. IMO, this would be an issue for anyone reading code that used this operator. Note that catchError and throwError are named as they are because catch and throw are keywords - not because they need to be qualified regarding what is caught or thrown.

[Trequetrum]

That's a good point.