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
import { ActorRef, EmittedFrom } from '.';
interface WaitForOptions {  /**   * How long to wait before rejecting, if no emitted   * state satisfies the predicate.   *   * @default 10_000 (10 seconds)   */  timeout: number;}
const defaultWaitForOptions: WaitForOptions = {  timeout: 10_000 // 10 seconds};
/** * Subscribes to an actor ref and waits for its emitted value to satisfy * a predicate, and then resolves with that value. * * @example * ```js * const state = await waitFor(someService, state => { *   return state.hasTag('loaded'); * }); * * state.hasTag('loaded'); // true * ``` * * @param actorRef The actor ref to subscribe to * @param predicate Determines if a value matches the condition to wait for * @param options * @returns A promise that eventually resolves to the emitted value * that matches the condition */export function waitFor<TActorRef extends ActorRef<any, any>>(  actorRef: TActorRef,  predicate: (emitted: EmittedFrom<TActorRef>) => boolean,  options?: Partial<WaitForOptions>): Promise<EmittedFrom<TActorRef>> {  const resolvedOptions: WaitForOptions = {    ...defaultWaitForOptions,    ...options  };  return new Promise((res, rej) => {    let done = false;    if (process.env.NODE_ENV !== 'production' && resolvedOptions.timeout < 0) {      console.error(        '`timeout` passed to `waitFor` is negative and it will reject its internal promise immediately.'      );    }    const handle =      resolvedOptions.timeout === Infinity        ? undefined        : setTimeout(() => {            sub.unsubscribe();            rej(new Error(`Timeout of ${resolvedOptions.timeout} ms exceeded`));          }, resolvedOptions.timeout);
    const dispose = () => {      clearTimeout(handle!);      done = true;      sub?.unsubscribe();    };
    const sub = actorRef.subscribe({      next: (emitted) => {        if (predicate(emitted)) {          dispose();          res(emitted);        }      },      error: (err) => {        dispose();        rej(err);      },      complete: () => {        dispose();        rej(new Error(`Actor terminated without satisfying predicate`));      }    });    if (done) {      sub.unsubscribe();    }  });}