to_signal.ts

/**
 * @license
 * Copyright Google LLC All Rights Reserved.
 *
 * Use of this source code is governed by an MIT-style license that can be
 * found in the LICENSE file at https://angular.io/license
 */

import {assertInInjectionContext, assertNotInReactiveContext, computed, DestroyRef, inject, Injector, signal, Signal, WritableSignal, ɵRuntimeError, ɵRuntimeErrorCode} from '@angular/core';
import {Observable, Subscribable} from 'rxjs';

/**
 * Options for `toSignal`.
 *
 * @publicApi
 */
export interface ToSignalOptions {
  /**
   * Initial value for the signal produced by `toSignal`.
   *
   * This will be the value of the signal until the observable emits its first value.
   */
  initialValue?: unknown;

  /**
   * Whether to require that the observable emits synchronously when `toSignal` subscribes.
   *
   * If this is `true`, `toSignal` will assert that the observable produces a value immediately upon
   * subscription. Setting this option removes the need to either deal with `undefined` in the
   * signal type or provide an `initialValue`, at the cost of a runtime error if this requirement is
   * not met.
   */
  requireSync?: boolean;

  /**
   * `Injector` which will provide the `DestroyRef` used to clean up the Observable subscription.
   *
   * If this is not provided, a `DestroyRef` will be retrieved from the current [injection
   * context](/guide/dependency-injection-context), unless manual cleanup is requested.
   */
  injector?: Injector;

  /**
   * Whether the subscription should be automatically cleaned up (via `DestroyRef`) when
   * `toObservable`'s creation context is destroyed.
   *
   * If manual cleanup is enabled, then `DestroyRef` is not used, and the subscription will persist
   * until the `Observable` itself completes.
   */
  manualCleanup?: boolean;
}

// Base case: no options -> `undefined` in the result type.
export function toSignal<T>(source: Observable<T>|Subscribable<T>): Signal<T|undefined>;
// Options with `undefined` initial value and no `requiredSync` -> `undefined`.
export function toSignal<T>(
    source: Observable<T>|Subscribable<T>,
    options: ToSignalOptions&{initialValue?: undefined, requireSync?: false}): Signal<T|undefined>;
// Options with `null` initial value -> `null`.
export function toSignal<T>(
    source: Observable<T>|Subscribable<T>,
    options: ToSignalOptions&{initialValue?: null, requireSync?: false}): Signal<T|null>;
// Options with `undefined` initial value and `requiredSync` -> strict result type.
export function toSignal<T>(
    source: Observable<T>|Subscribable<T>,
    options: ToSignalOptions&{initialValue?: undefined, requireSync: true}): Signal<T>;
// Options with a more specific initial value type.
export function toSignal<T, const U extends T>(
    source: Observable<T>|Subscribable<T>,
    options: ToSignalOptions&{initialValue: U, requireSync?: false}): Signal<T|U>;

/**
 * Get the current value of an `Observable` as a reactive `Signal`.
 *
 * `toSignal` returns a `Signal` which provides synchronous reactive access to values produced
 * by the given `Observable`, by subscribing to that `Observable`. The returned `Signal` will always
 * have the most recent value emitted by the subscription, and will throw an error if the
 * `Observable` errors.
 *
 * With `requireSync` set to `true`, `toSignal` will assert that the `Observable` produces a value
 * immediately upon subscription. No `initialValue` is needed in this case, and the returned signal
 * does not include an `undefined` type.
 *
 * By default, the subscription will be automatically cleaned up when the current [injection
 * context](/guide/dependency-injection-context) is destroyed. For example, when `toObservable` is
 * called during the construction of a component, the subscription will be cleaned up when the
 * component is destroyed. If an injection context is not available, an explicit `Injector` can be
 * passed instead.
 *
 * If the subscription should persist until the `Observable` itself completes, the `manualCleanup`
 * option can be specified instead, which disables the automatic subscription teardown. No injection
 * context is needed in this configuration as well.
 *
 * @developerPreview
 */
export function toSignal<T, U = undefined>(
    source: Observable<T>|Subscribable<T>,
    options?: ToSignalOptions&{initialValue?: U}): Signal<T|U> {
  ngDevMode &&
      assertNotInReactiveContext(
          toSignal,
          'Invoking `toSignal` causes new subscriptions every time. ' +
              'Consider moving `toSignal` outside of the reactive context and read the signal value where needed.');

  const requiresCleanup = !options?.manualCleanup;
  requiresCleanup && !options?.injector && assertInInjectionContext(toSignal);
  const cleanupRef =
      requiresCleanup ? options?.injector?.get(DestroyRef) ?? inject(DestroyRef) : null;

  // Note: T is the Observable value type, and U is the initial value type. They don't have to be
  // the same - the returned signal gives values of type `T`.
  let state: WritableSignal<State<T|U>>;
  if (options?.requireSync) {
    // Initially the signal is in a `NoValue` state.
    state = signal({kind: StateKind.NoValue});
  } else {
    // If an initial value was passed, use it. Otherwise, use `undefined` as the initial value.
    state = signal<State<T|U>>({kind: StateKind.Value, value: options?.initialValue as U});
  }

  // Note: This code cannot run inside a reactive context (see assertion above). If we'd support
  // this, we would subscribe to the observable outside of the current reactive context, avoiding
  // that side-effect signal reads/writes are attribute to the current consumer. The current
  // consumer only needs to be notified when the `state` signal changes through the observable
  // subscription. Additional context (related to async pipe):
  // https://github.com/angular/angular/pull/50522.
  const sub = source.subscribe({
    next: value => state.set({kind: StateKind.Value, value}),
    error: error => state.set({kind: StateKind.Error, error}),
    // Completion of the Observable is meaningless to the signal. Signals don't have a concept of
    // "complete".
  });

  if (ngDevMode && options?.requireSync && state().kind === StateKind.NoValue) {
    throw new ɵRuntimeError(
        ɵRuntimeErrorCode.REQUIRE_SYNC_WITHOUT_SYNC_EMIT,
        '`toSignal()` called with `requireSync` but `Observable` did not emit synchronously.');
  }

  // Unsubscribe when the current context is destroyed, if requested.
  cleanupRef?.onDestroy(sub.unsubscribe.bind(sub));

  // The actual returned signal is a `computed` of the `State` signal, which maps the various states
  // to either values or errors.
  return computed(() => {
    const current = state();
    switch (current.kind) {
      case StateKind.Value:
        return current.value;
      case StateKind.Error:
        throw current.error;
      case StateKind.NoValue:
        // This shouldn't really happen because the error is thrown on creation.
        // TODO(alxhub): use a RuntimeError when we finalize the error semantics
        throw new ɵRuntimeError(
            ɵRuntimeErrorCode.REQUIRE_SYNC_WITHOUT_SYNC_EMIT,
            '`toSignal()` called with `requireSync` but `Observable` did not emit synchronously.');
    }
  });
}

const enum StateKind {
  NoValue,
  Value,
  Error,
}

interface NoValueState {
  kind: StateKind.NoValue;
}

interface ValueState<T> {
  kind: StateKind.Value;
  value: T;
}

interface ErrorState {
  kind: StateKind.Error;
  error: unknown;
}

type State<T> = NoValueState|ValueState<T>|ErrorState;