如何用RxJS管道操作符替代自定义Observable实现相同逻辑？

Absolutely! You can ditch the custom Observable and replicate your exact logic using a clean chain of RxJS pipeable operators. Here's the equivalent implementation:

import { of, EMPTY, throwError } from 'rxjs';
import { mergeMap } from 'rxjs/operators';

// ...

return this.anotherObservable().pipe(
  mergeMap(next => {
    if (this.condition1(next)) {
      return of(next);
    } else if (this.condition2(next)) {
      return EMPTY;
    } else {
      return throwError(() => 'there was an error');
    }
  })
);

How This Matches Your Original Logic #

Let's map each part of your custom Observable to the operator chain:

• Conditional value emission: When condition1(next) passes, of(next) emits the value to the subscriber (just like subscriber.next(next)). The stream stays open for more values from anotherObservable().
• Early completion: When condition2(next) passes, EMPTY emits an immediate complete notification, which terminates the entire stream—matching your subscriber.complete().
• Error handling: When neither condition is met, throwError(...) emits an error notification, identical to your subscriber.error('there was an error').
• Source Observable error/complete: Any error from anotherObservable() automatically propagates through the pipe to the subscriber, just like your error handler. Similarly, when anotherObservable() completes, that notification is passed along to the subscriber, matching your complete handler.
• Subscription cleanup: RxJS automatically manages the subscription to anotherObservable() when the resulting Observable is unsubscribed. You don't need to manually return an unsubscribe function—this matches the cleanup logic in your custom Observable.

Quick Note on Operator Choice #

Since all the Observables we return in mergeMap are synchronous, you could also use concatMap or switchMap here—they'll work exactly the same. mergeMap is a safe, general-purpose choice for this pattern.

内容的提问来源于stack exchange，提问作者Elie Faës