javascript - rxjs - 具有下一个和错误回调的运算符？

Question

我正在寻找 rxjs 中的 promise.then(onNextCallback,onErrorCallback) 的等价物。

有这样的东西吗？

• pipe(concatMap(),catchError) 替代方案不是我要找的。

Answer 1

I also asked for this as well并最终写了我自己的。

import * as RxJS from 'rxjs';

/**
 * Like `promise.then(onFulfilled, onRejected)`. This is *not* the same as
 * `map(onNext).catchError(onError)`, because this operator will only catch errors resulting from
 * the original observable—not errors resulting from `onNext`.
 */
export const mapOrCatchError = <T, B>(
  onNext: (value: T) => B,
  onError: (value: unknown) => B,
): RxJS.OperatorFunction<T, B> => ob$ =>
  new RxJS.Observable<B>(observer =>
    ob$.subscribe({
      next: t => {
        let next: B;
        try {
          next = onNext(t);
        } catch (error) {
          observer.error(error);
          return;
        }
        observer.next(next);
      },
      error: error => {
        let next: B;
        try {
          next = onError(error);
        } catch (newError) {
          observer.error(newError);
          return;
        }
        observer.next(next);
        observer.complete();
      },
      complete: () => {
        observer.complete();
      },
    }),
  );

测试:

import { marbles } from 'rxjs-marbles/jest';
import { mapOrCatchError } from '../operators';

describe('mapOrCatchError', () => {
  it(
    'should map',
    marbles(m => {
      const source$ = m.cold('--(a|)', { a: 1 });
      const expected = '      --(b|)';

      const actual$ = source$.pipe(
        mapOrCatchError(
          a => a + 1,
          _error => 0,
        ),
      );
      m.expect(actual$).toBeObservable(expected, { b: 2 });
    }),
  );
  it(
    'should catch',
    marbles(m => {
      const source$ = m.cold('--#');
      const expected = '      --(a|)';

      const actual$ = source$.pipe(
        mapOrCatchError(
          a => a + 1,
          _error => 0,
        ),
      );
      m.expect(actual$).toBeObservable(expected, { a: 0 });
    }),
  );
  it(
    'should error if error handler throws',
    marbles(m => {
      const source$ = m.cold('--#');
      const expected = '      --#';

      const error = new Error('foo');
      const actual$ = source$.pipe(
        mapOrCatchError(
          a => a + 1,
          _error => {
            throw error;
          },
        ),
      );
      m.expect(actual$).toBeObservable(expected, undefined, error);
    }),
  );
  it(
    'should not catch errors thrown by map function',
    marbles(m => {
      const source$ = m.cold('--(a|)');
      const expected = '      --#';

      const error = new Error('foo');
      const actual$ = source$.pipe(
        mapOrCatchError(
          () => {
            throw error;
          },
          _error => 'caught error',
        ),
      );
      m.expect(actual$).toBeObservable(expected, undefined, error);
    }),
  );
});