自定义RxJS筛选器运算符中的不可为空类型保护

import { Observable, throwError } from 'rxjs';
import { filter } from 'rxjs/operators';

// 自定义筛选器运算符
function notNull<T>(): OperatorFunction<T, T> {
  return (source: Observable<T>) => {
    return new Observable<T>(observer => {
      return source.subscribe({
        next(value) {
          if (value !== null && value !== undefined) {
            observer.next(value);
          } else {
            observer.error(new Error('Value cannot be null or undefined.'));
          }
        },
        error(error) {
          observer.error(error);
        },
        complete() {
          observer.complete();
        }
      });
    });
  };
}

// 使用自定义筛选器运算符
const source$ = of(1, null, 2, undefined, 3).pipe(
  notNull(),
  filter(value => value > 1)
);

source$.subscribe({
  next(value) {
    console.log(value);
  },
  error(error) {
    console.error(error);
  },
  complete() {
    console.log('Complete');
  }
});