Function createPlatformTestObservable

packages/test/src/test-sources.ts:133–182

(
  controller: VirtualTimeController,
  messages: readonly TestMessage<T>[]
)

}

export function createPlatformTestObservable<T>(
  controller: VirtualTimeController,
  messages: readonly TestMessage<T>[]
): TestPlatformObservable<T> {
  const ObservableConstructor = getObservableConstructor();
  const subscriptions: MutableSubscriptionLog[] = [];
  const observable = new ObservableConstructor<T>((subscriber) => {
    const log = openSubscriptionLog(controller, subscriptions);
    const activatedAt = controller.now();
    let closed = false;
    const close = (): void => {
      if (!closed) {
        closed = true;
        closeSubscriptionLog(controller, log);
      }
    };
    subscriber.addTeardown(close);

    for (const message of messages) {
      controller.scheduleAt(
        () => {
          if (!subscriber.active) {
            return;
          }
          switch (message.notification.kind) {
            case 'N':
              subscriber.next(message.notification.value);
              break;
            case 'E':
              subscriber.error(message.notification.error);
              close();
              break;
            case 'C':
              subscriber.complete();
              close();
              break;
          }
        },
        activatedAt + message.frame,
        { signal: subscriber.signal }
      );
    }
  });

  return Object.assign(observable, {
    kind: 'observable' as const,
    messages,
    subscriptions,
  });
}

class IndependentTestSubscriber<T> implements Subscriber<T> {
  readonly #observer: Partial<Observer<T>>;