first commit

liusuyi

2023-04-24 4737f1e038743ced243c9e52423404d9034d6107

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
import { Observable } from '../Observable';
import { innerFrom } from '../observable/innerFrom';
import { Subject } from '../Subject';
import { Subscription } from '../Subscription';
 
import { MonoTypeOperatorFunction, ObservableInput } from '../types';
import { operate } from '../util/lift';
import { createOperatorSubscriber } from './OperatorSubscriber';
 
/**
 * Returns an Observable that mirrors the source Observable with the exception of a `complete`. If the source
 * Observable calls `complete`, this method will emit to the Observable returned from `notifier`. If that Observable
 * calls `complete` or `error`, then this method will call `complete` or `error` on the child subscription. Otherwise
 * this method will resubscribe to the source Observable.
 *
 * ![](repeatWhen.png)
 *
 * ## Example
 *
 * Repeat a message stream on click
 *
 * ```ts
 * import { of, fromEvent, repeatWhen } from 'rxjs';
 *
 * const source = of('Repeat message');
 * const documentClick$ = fromEvent(document, 'click');
 *
 * const result = source.pipe(repeatWhen(() => documentClick$));
 *
 * result.subscribe(data => console.log(data))
 * ```
 *
 * @see {@link repeat}
 * @see {@link retry}
 * @see {@link retryWhen}
 *
 * @param notifier Function that receives an Observable of notifications with
 * which a user can `complete` or `error`, aborting the repetition.
 * @return A function that returns an `ObservableInput` that mirrors the source
 * Observable with the exception of a `complete`.
 * @deprecated Will be removed in v9 or v10. Use {@link repeat}'s {@link RepeatConfig#delay delay} option instead.
 * Instead of `repeatWhen(() => notify$)`, use: `repeat({ delay: () => notify$ })`.
 */
export function repeatWhen<T>(notifier: (notifications: Observable<void>) => ObservableInput<any>): MonoTypeOperatorFunction<T> {
  return operate((source, subscriber) => {
    let innerSub: Subscription | null;
    let syncResub = false;
    let completions$: Subject<void>;
    let isNotifierComplete = false;
    let isMainComplete = false;
 
    /**
     * Checks to see if we can complete the result, completes it, and returns `true` if it was completed.
     */
    const checkComplete = () => isMainComplete && isNotifierComplete && (subscriber.complete(), true);
    /**
     * Gets the subject to send errors through. If it doesn't exist,
     * we know we need to setup the notifier.
     */
    const getCompletionSubject = () => {
      if (!completions$) {
        completions$ = new Subject();
 
        // If the call to `notifier` throws, it will be caught by the OperatorSubscriber
        // In the main subscription -- in `subscribeForRepeatWhen`.
        innerFrom(notifier(completions$)).subscribe(
          createOperatorSubscriber(
            subscriber,
            () => {
              if (innerSub) {
                subscribeForRepeatWhen();
              } else {
                // If we don't have an innerSub yet, that's because the inner subscription
                // call hasn't even returned yet. We've arrived here synchronously.
                // So we flag that we want to resub, such that we can ensure finalization
                // happens before we resubscribe.
                syncResub = true;
              }
            },
            () => {
              isNotifierComplete = true;
              checkComplete();
            }
          )
        );
      }
      return completions$;
    };
 
    const subscribeForRepeatWhen = () => {
      isMainComplete = false;
 
      innerSub = source.subscribe(
        createOperatorSubscriber(subscriber, undefined, () => {
          isMainComplete = true;
          // Check to see if we are complete, and complete if so.
          // If we are not complete. Get the subject. This calls the `notifier` function.
          // If that function fails, it will throw and `.next()` will not be reached on this
          // line. The thrown error is caught by the _complete handler in this
          // `OperatorSubscriber` and handled appropriately.
          !checkComplete() && getCompletionSubject().next();
        })
      );
 
      if (syncResub) {
        // Ensure that the inner subscription is torn down before
        // moving on to the next subscription in the synchronous case.
        // If we don't do this here, all inner subscriptions will not be
        // torn down until the entire observable is done.
        innerSub.unsubscribe();
        // It is important to null this out. Not only to free up memory, but
        // to make sure code above knows we are in a subscribing state to
        // handle synchronous resubscription.
        innerSub = null;
        // We may need to do this multiple times, so reset the flags.
        syncResub = false;
        // Resubscribe
        subscribeForRepeatWhen();
      }
    };
 
    // Start the subscription
    subscribeForRepeatWhen();
  });
}