创建一个自定义RXJS管道，该管道接受一个可观察对象并等待，直到该可观察对象满足某个条件

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

// 自定义管道函数
function customPipe<T>(condition: (value: T) => boolean) {
  return (source: Observable<T>) =>
    new Observable<T>(observer => {
      const subscription = source
        .pipe(
          filter(value => condition(value)), // 使用filter操作符过滤满足条件的值
          take(1) // 使用take操作符只取第一个满足条件的值
        )
        .subscribe({
          next: value => {
            observer.next(value); // 将满足条件的值传递给下游观察者
            observer.complete(); // 完成观察
          },
          error: err => observer.error(err),
          complete: () => observer.complete()
        });

      return () => {
        subscription.unsubscribe(); // 取消订阅
      };
    });
}

// 创建一个可观察对象
const observable = new Observable<number>(observer => {
  let count = 0;
  const intervalId = setInterval(() => {
    observer.next(count++);
  }, 1000);

  return () => {
    clearInterval(intervalId); // 清除定时器
  };
});

// 使用自定义管道等待可观察对象满足条件
observable
  .pipe(
    customPipe<number>(value => value > 5) // 自定义条件为值大于5
  )
  .subscribe({
    next: value => console.log(value),
    complete: () => console.log('Complete')
  });