Function thresholdToggle

operators/src/thresholdToggle.ts:26–69

(
  workCreator: (event: T) => ObservableInput<R>,
  threshold: number = 2,
  mapper?: (_: T, inner: R) => S
)

 * ```
 */
 export function thresholdToggle<T, R, S = R>(
  workCreator: (event: T) => ObservableInput<R>,
  threshold: number = 2,
  mapper?: (_: T, inner: R) => S
): OperatorFunction<T, S> {
  type NewType = Observable<T>;

  return function (source: NewType) {
    return new Observable((notify) => {
      let innerSub: Subscription | null;
      let hitsTowardThreshold = 0;

      return source.subscribe({
        next(trigger) {
          hitsTowardThreshold += 1;

          if (hitsTowardThreshold >= threshold) {
            if (!innerSub || innerSub?.closed) {
              innerSub = from(workCreator(trigger)).subscribe({
                next: (inner) => {
                  const result = (mapper ? mapper(trigger, inner) : inner) as S;
                  notify.next(result);
                },
                error: (e) => notify.error(e)
              });
            } else {
              console.log(innerSub?.closed, hitsTowardThreshold);
              innerSub.unsubscribe();
              innerSub = null;
            }

            hitsTowardThreshold = 0;
          }
        },
        error(e) {
          notify.error(e);
        },
        complete() {
          notify.complete();
        }
      });
    });
  };
}