Function pendingUntilEvent

packages/core/rxjs-interop/src/pending_until_event.ts:21–64

(injector?: Injector)

 * @developerPreview 20.0
 */
export function pendingUntilEvent<T>(injector?: Injector): MonoTypeOperatorFunction<T> {
  if (injector === undefined) {
    ngDevMode && assertInInjectionContext(pendingUntilEvent);
    injector = inject(Injector);
  }
  const taskService = injector.get(PendingTasks);

  return (sourceObservable) => {
    return new Observable<T>((originalSubscriber) => {
      // create a new task on subscription
      const removeTask = taskService.add();

      let cleanedUp = false;
      function cleanupTask() {
        if (cleanedUp) {
          return;
        }

        removeTask();
        cleanedUp = true;
      }

      const innerSubscription = sourceObservable.subscribe({
        next: (v) => {
          originalSubscriber.next(v);
          cleanupTask();
        },
        complete: () => {
          originalSubscriber.complete();
          cleanupTask();
        },
        error: (e) => {
          originalSubscriber.error(e);
          cleanupTask();
        },
      });
      innerSubscription.add(() => {
        originalSubscriber.unsubscribe();
        cleanupTask();
      });
      return innerSubscription;
    });
  };
}