feat(core): prototype implementation of @angular/core/rxjs-interop (#49154)

alxhub · atscott · commit 8997bdc03bd3 · 2023-03-30T09:44:38.000-07:00
This commit adds the basic sketch for the implementation of `fromObservable` and `fromSignal`, the two basic primitives which form the RxJS interop layer with signals. PR Close #49154
diff --git a/goldens/public-api/core/rxjs-interop/index.md b/goldens/public-api/core/rxjs-interop/index.md
@@ -4,6 +4,24 @@
 
 ```ts
 
+import { Injector } from '@angular/core';
+import { Observable } from 'rxjs';
+import { Signal } from '@angular/core';
+
+// @public
+export function fromObservable<T>(source: Observable<T>): Signal<T>;
+
+// @public
+export function fromObservable<T, U extends T | null | undefined>(source: Observable<T>, initialValue: U): Signal<T | U>;
+
+// @public
+export function fromSignal<T>(source: Signal<T>, options?: FromSignalOptions): Observable<T>;
+
+// @public
+export interface FromSignalOptions {
+    injector?: Injector;
+}
+
 // (No @packageDocumentation comment for this package)
 
 ```
diff --git a/packages/core/rxjs-interop/src/from_observable.ts b/packages/core/rxjs-interop/src/from_observable.ts
@@ -0,0 +1,118 @@
+/**
+ * @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, computed, DestroyRef, inject, signal, Signal, WritableSignal} from '@angular/core';
+import {Observable} from 'rxjs';
+
+/**
+ * Get the current value of an `Observable` as a reactive `Signal`.
+ *
+ * `fromObservable` 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.
+ *
+ * The subscription will last for the lifetime of the current injection context. That is, if
+ * `fromObservable` is called from a component context, the subscription will be cleaned up when the
+ * component is destroyed. When called outside of a component, the current `EnvironmentInjector`'s
+ * lifetime will be used (which is typically the lifetime of the application itself).
+ *
+ * If the `Observable` does not produce a value before the `Signal` is read, the `Signal` will throw
+ * an error. To avoid this, use a synchronous `Observable` (potentially created with the `startWith`
+ * operator) or pass an initial value to `fromObservable` as the second argument.
+ *
+ * `fromObservable` must be called in an injection context.
+ */
+export function fromObservable<T>(source: Observable<T>): Signal<T>;
+
+/**
+ * Get the current value of an `Observable` as a reactive `Signal`.
+ *
+ * `fromObservable` 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.
+ *
+ * The subscription will last for the lifetime of the current injection context. That is, if
+ * `fromObservable` is called from a component context, the subscription will be cleaned up when the
+ * component is destroyed. When called outside of a component, the current `EnvironmentInjector`'s
+ * lifetime will be used (which is typically the lifetime of the application itself).
+ *
+ * Before the `Observable` emits its first value, the `Signal` will return the configured
+ * `initialValue`. If the `Observable` is known to produce a value before the `Signal` will be read,
+ * `initialValue` does not need to be passed.
+ *
+ * `fromObservable` must be called in an injection context.
+ *
+ * @developerPreview
+ */
+export function fromObservable<T, U extends T|null|undefined>(
+    // fromObservable(Observable<Animal>) -> Signal<Cat>
+    source: Observable<T>, initialValue: U): Signal<T|U>;
+export function fromObservable<T, U = never>(source: Observable<T>, initialValue?: U): Signal<T|U> {
+  assertInInjectionContext(fromObservable);
+
+  // 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 (initialValue === undefined && arguments.length !== 2) {
+    // No initial value was passed, so initially the signal is in a `NoValue` state and will throw
+    // if accessed.
+    state = signal({kind: StateKind.NoValue});
+  } else {
+    // An initial value was passed, so use it.
+    state = signal<State<T|U>>({kind: StateKind.Value, value: initialValue!});
+  }
+
+  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".
+  });
+
+  // Unsubscribe when the current context is destroyed.
+  inject(DestroyRef).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:
+        // TODO(alxhub): use a RuntimeError when we finalize the error semantics
+        throw new Error(`fromObservable() signal read before the Observable emitted`);
+    }
+  });
+}
+
+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;
diff --git a/packages/core/rxjs-interop/src/from_signal.ts b/packages/core/rxjs-interop/src/from_signal.ts
@@ -0,0 +1,54 @@
+/**
+ * @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, effect, inject, Injector, Signal} from '@angular/core';
+import {Observable} from 'rxjs';
+
+/**
+ * Options for `fromSignal`.
+ *
+ * @developerPreview
+ */
+export interface FromSignalOptions {
+  /**
+   * The `Injector` to use when creating the effect.
+   *
+   * If this isn't specified, the current injection context will be used.
+   */
+  injector?: Injector;
+}
+
+/**
+ * Exposes the value of an Angular `Signal` as an RxJS `Observable`.
+ *
+ * The signal's value will be propagated into the `Observable`'s subscribers using an `effect`.
+ *
+ * `fromSignal` must be called in an injection context.
+ *
+ * @developerPreview
+ */
+export function fromSignal<T>(
+    source: Signal<T>,
+    options?: FromSignalOptions,
+    ): Observable<T> {
+  !options?.injector && assertInInjectionContext(fromSignal);
+  const injector = options?.injector ?? inject(Injector);
+
+  // Creating a new `Observable` allows the creation of the effect to be lazy. This allows for all
+  // references to `source` to be dropped if the `Observable` is fully unsubscribed and thrown away.
+  return new Observable(observer => {
+    const watcher = effect(() => {
+      try {
+        observer.next(source());
+      } catch (err) {
+        observer.error(err);
+      }
+    }, {injector, manualCleanup: true});
+    return () => watcher.destroy();
+  });
+}
diff --git a/packages/core/rxjs-interop/src/index.ts b/packages/core/rxjs-interop/src/index.ts
@@ -6,4 +6,5 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-export {};
+export {fromObservable} from './from_observable';
+export {fromSignal, FromSignalOptions} from './from_signal';
diff --git a/packages/core/rxjs-interop/test/BUILD.bazel b/packages/core/rxjs-interop/test/BUILD.bazel
@@ -0,0 +1,38 @@
+load("//tools:defaults.bzl", "jasmine_node_test", "karma_web_test_suite", "ts_library")
+load("//tools/circular_dependency_test:index.bzl", "circular_dependency_test")
+
+circular_dependency_test(
+    name = "circular_deps_test",
+    entry_point = "angular/packages/core/rxjs-interop/index.mjs",
+    deps = ["//packages/core/rxjs-interop"],
+)
+
+ts_library(
+    name = "test_lib",
+    testonly = True,
+    srcs = glob(["**/*.ts"]),
+    deps = [
+        "//packages:types",
+        "//packages/core",
+        "//packages/core/rxjs-interop",
+        "//packages/core/src/signals",
+        "//packages/core/testing",
+        "//packages/private/testing",
+        "@npm//rxjs",
+    ],
+)
+
+jasmine_node_test(
+    name = "test",
+    bootstrap = ["//tools/testing:node"],
+    deps = [
+        ":test_lib",
+    ],
+)
+
+karma_web_test_suite(
+    name = "test_web",
+    deps = [
+        ":test_lib",
+    ],
+)
diff --git a/packages/core/rxjs-interop/test/from_observable_spec.ts b/packages/core/rxjs-interop/test/from_observable_spec.ts
@@ -0,0 +1,105 @@
+/**
+ * @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 {EnvironmentInjector, Injector, runInInjectionContext} from '@angular/core';
+import {fromObservable} from '@angular/core/rxjs-interop';
+import {BehaviorSubject, Subject} from 'rxjs';
+
+import {test} from './util';
+
+describe('fromObservable()', () => {
+  it('should reflect the last emitted value of an Observable', test(() => {
+       const counter$ = new BehaviorSubject(0);
+       const counter = fromObservable(counter$);
+
+       expect(counter()).toBe(0);
+       counter$.next(1);
+       expect(counter()).toBe(1);
+       counter$.next(3);
+       expect(counter()).toBe(3);
+     }));
+
+  it('should notify when the last emitted value of an Observable changes', test(() => {
+       let seenValue: number = 0;
+       const counter$ = new BehaviorSubject(1);
+       const counter = fromObservable(counter$);
+
+       expect(counter()).toBe(1);
+
+       counter$.next(2);
+       expect(counter()).toBe(2);
+     }));
+
+  it('should propagate an error returned by the Observable', test(() => {
+       const counter$ = new BehaviorSubject(1);
+       const counter = fromObservable(counter$);
+
+       expect(counter()).toBe(1);
+
+       counter$.error('fail');
+       expect(counter).toThrow('fail');
+     }));
+
+  it('should unsubscribe when the current context is destroyed', test(() => {
+       const counter$ = new BehaviorSubject(0);
+       const injector = Injector.create({providers: []}) as EnvironmentInjector;
+       const counter = runInInjectionContext(injector, () => fromObservable(counter$));
+
+       expect(counter()).toBe(0);
+       counter$.next(1);
+       expect(counter()).toBe(1);
+
+       // Destroying the injector should unsubscribe the Observable.
+       injector.destroy();
+
+       // The signal should have the last value observed.
+       expect(counter()).toBe(1);
+
+       // And this value should no longer be updating (unsubscribed).
+       counter$.next(2);
+       expect(counter()).toBe(1);
+     }));
+
+  describe('with no initial value', () => {
+    it('should throw if called before a value is emitted', test(() => {
+         const counter$ = new Subject<number>();
+         const counter = fromObservable(counter$);
+
+         expect(() => counter()).toThrow();
+         counter$.next(1);
+         expect(counter()).toBe(1);
+       }));
+
+    it('should not throw if a value is emitted before called', test(() => {
+         const counter$ = new Subject<number>();
+         const counter = fromObservable(counter$);
+
+         counter$.next(1);
+         expect(() => counter()).not.toThrow();
+       }));
+  });
+
+  describe('with an initial value', () => {
+    it('should return the initial value if called before a value is emitted', test(() => {
+         const counter$ = new Subject<number>();
+         const counter = fromObservable(counter$, null);
+
+         expect(counter()).toBeNull();
+         counter$.next(1);
+         expect(counter()).toBe(1);
+       }));
+
+    it('should not return the initial value if called after a value is emitted', test(() => {
+         const counter$ = new Subject<number>();
+         const counter = fromObservable(counter$, null);
+
+         counter$.next(1);
+         expect(counter()).not.toBeNull();
+       }));
+  });
+});
diff --git a/packages/core/rxjs-interop/test/from_signal_spec.ts b/packages/core/rxjs-interop/test/from_signal_spec.ts
diff --git a/packages/core/rxjs-interop/test/util.ts b/packages/core/rxjs-interop/test/util.ts
diff --git a/packages/core/src/signals/BUILD.bazel b/packages/core/src/signals/BUILD.bazel
