Function toSignal

packages/core/rxjs-interop/src/to_signal.ts:128–225

(
  source: Observable<T> | Subscribable<T>,
  options?: ToSignalOptions<T | U> & {initialValue?: U},
)

 * @see [RxJS interop with Angular signals](ecosystem/rxjs-interop)
 */
export function toSignal<T, U = undefined>(
  source: Observable<T> | Subscribable<T>,
  options?: ToSignalOptions<T | U> & {initialValue?: U},
): Signal<T | U> {
  typeof ngDevMode !== 'undefined' &&
    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;

  if (ngDevMode && requiresCleanup && !options?.injector) {
    assertInInjectionContext(toSignal);
  }

  const cleanupRef = requiresCleanup
    ? (options?.injector?.get(DestroyRef) ?? inject(DestroyRef))
    : null;

  const equal = makeToSignalEqual(options?.equal);

  // 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},
      {equal, ...(ngDevMode ? createDebugNameObject(options?.debugName, 'state') : undefined)},
    );
  } 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},
      {equal, ...(ngDevMode ? createDebugNameObject(options?.debugName, 'state') : undefined)},
    );
  }

  let destroyUnregisterFn: (() => void) | undefined;

  // 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});
      destroyUnregisterFn?.();
    },
    complete: () => {
      destroyUnregisterFn?.();
    },