Variant of takeUntil() which completes the source Observable when the notifier Observable completes

[kalgon]

Would a variation of takeUntil() which would complete the source Observable when the notifier Observable completes make sense?

In the following case, we wouldn't need to call next() on the notifier anymore:

notifier$ = new Subject();
source$.pipe(takeUntil(notifier$),...);
notifier$.next(); // not needed anymore
notifier$.complete();

This could be a new operator (takeUntilCompletes(notifier$)) or the existing one could be augmented (takeUntil<T>(notifier: ObservableInput<any>, on: 'emits' | 'completes' = 'emits'): MonoTypeOperatorFunction<T>)

[josepot]

You could do:

source$.pipe(takeUntil(
  notifier$.pipe(endWith(null))
))

[benlesh]

FYI: it's not a party foul to write your own operators. But if you don't want to do that, and you don't want to test them, then go with @josepot's suggestion.

You could do this:

import { OperatorFunction, ObservableInput, from } from 'rxjs';

export function takeUntilComplete<T>(notifier: ObservableInput<unknown>): OperatorFunction<T, T> {
  return (source) => new Observable(subscriber => {
    subscriber.add(
      from(notifier).subscribe({
        complete: () => subscriber.complete(),
        error: (err) => subscriber.error(err)
      })
    )
  
    source.subscribe(subscriber)
  })
}

Which in RxJS 8 (alpha) still works 100%, and always will, but if you did it with the new operate function, it doesn't look much different:

import { OperatorFunction, ObservableInput, from, operate, noop } from 'rxjs';

export function takeUntilComplete<T>(notifier: ObservableInput<unknown>): OperatorFunction<T, T> {
  return (source) => new Observable(destination => {
    from(notifier).subscribe(operate({
      destination,
      next: noop,
    }))
  
    source.subscribe(subscriber)
  })
}