readme.md

RxJS: finalCheck operator

Operator that will be runned on stream completition and throw error if callback return something.

Reason: finalize operator cannot pass error to a stream.

const finalCheck = (checkFn: () => any) => <T>(source: Observable<T>) =>
  new Observable<T>(observer => {
    return source.subscribe({
      next(x) {
        observer.next(x);
      },
      error(err) { observer.error(err); },
      complete() { 
        const err = checkFn();
        if (err) {
          observer.error(err);
        }
        observer.complete();
      }
    })
  });

Usage:

...pipe(
  ...
  finalCheck(() => {
    if (isError) {
      return 'Error message';
    }
  }),