feat(core): support streaming resources (#59573)

alxhub · AndrewKushnir · commit bc2ad7bfd37a · 2025-01-21T09:55:32.000-08:00
This commit adds support for creating `resource()`s with streaming response data. A streaming resource is defined by a `stream` option instead of a `loader`, with `stream` being a function returning `Promise<Signal<{value: T}|{error: unknown}>>`. Once the streaming loader resolves to a `Signal`, it can continue to update that signal over time, and the values (or errors) will be delivered to via the resource's state. `rxResource()` is updated to leverage this new functionality to handle multiple responses from the underlying Observable. PR Close #59573
diff --git a/goldens/public-api/core/index.api.md b/goldens/public-api/core/index.api.md
@@ -184,6 +184,13 @@ export interface AttributeDecorator {
     new (name: string): Attribute;
 }
 
+// @public
+export interface BaseResourceOptions<T, R> {
+    equal?: ValueEqualityFn<T>;
+    injector?: Injector;
+    request?: () => R;
+}
+
 // @public
 export function booleanAttribute(value: unknown): boolean;
 
@@ -1436,6 +1443,11 @@ export class PlatformRef {
 // @public
 export type Predicate<T> = (value: T) => boolean;
 
+// @public
+export interface PromiseResourceOptions<T, R> extends BaseResourceOptions<T, R> {
+    loader: ResourceLoader<T, R>;
+}
+
 // @public
 export function provideAppInitializer(initializerFn: () => Observable<unknown> | Promise<unknown> | void): EnvironmentProviders;
 
@@ -1610,13 +1622,8 @@ export interface ResourceLoaderParams<R> {
     request: Exclude<NoInfer<R>, undefined>;
 }
 
-// @public
-export interface ResourceOptions<T, R> {
-    equal?: ValueEqualityFn<T>;
-    injector?: Injector;
-    loader: ResourceLoader<T, R>;
-    request?: () => R;
-}
+// @public (undocumented)
+export type ResourceOptions<T, R> = PromiseResourceOptions<T, R> | StreamingResourceOptions<T, R>;
 
 // @public
 export interface ResourceRef<T> extends WritableResource<T> {
@@ -1633,6 +1640,13 @@ export enum ResourceStatus {
     Resolved = 4
 }
 
+// @public
+export type ResourceStreamingLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLike<Signal<{
+    value: T;
+} | {
+    error: unknown;
+}>>;
+
 // @public
 export const RESPONSE_INIT: InjectionToken<ResponseInit | null>;
 
@@ -1747,6 +1761,11 @@ export interface StaticClassSansProvider {
 // @public
 export type StaticProvider = ValueProvider | ExistingProvider | StaticClassProvider | ConstructorProvider | FactoryProvider | any[];
 
+// @public
+export interface StreamingResourceOptions<T, R> extends BaseResourceOptions<T, R> {
+    stream: ResourceStreamingLoader<T, R>;
+}
+
 // @public
 export abstract class TemplateRef<C> {
     abstract createEmbeddedView(context: C, injector?: Injector): EmbeddedViewRef<C>;
diff --git a/goldens/public-api/core/rxjs-interop/index.api.md b/goldens/public-api/core/rxjs-interop/index.api.md
@@ -4,14 +4,14 @@
 
 ```ts
 
+import { BaseResourceOptions } from '@angular/core';
 import { DestroyRef } from '@angular/core';
 import { Injector } from '@angular/core';
 import { MonoTypeOperatorFunction } from 'rxjs';
 import { Observable } from 'rxjs';
 import { OutputOptions } from '@angular/core';
 import { OutputRef } from '@angular/core';
 import { ResourceLoaderParams } from '@angular/core';
-import { ResourceOptions } from '@angular/core';
 import { ResourceRef } from '@angular/core';
 import { Signal } from '@angular/core';
 import { Subscribable } from 'rxjs';
@@ -30,7 +30,7 @@ export function pendingUntilEvent<T>(injector?: Injector): MonoTypeOperatorFunct
 export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T | undefined>;
 
 // @public
-export interface RxResourceOptions<T, R> extends Omit<ResourceOptions<T, R>, 'loader'> {
+export interface RxResourceOptions<T, R> extends BaseResourceOptions<T, R> {
     // (undocumented)
     loader: (params: ResourceLoaderParams<R>) => Observable<T>;
 }
diff --git a/packages/core/rxjs-interop/src/rx_resource.ts b/packages/core/rxjs-interop/src/rx_resource.ts
@@ -8,20 +8,21 @@
 
 import {
   assertInInjectionContext,
-  ResourceOptions,
   resource,
   ResourceLoaderParams,
   ResourceRef,
+  Signal,
+  signal,
+  BaseResourceOptions,
 } from '@angular/core';
-import {Observable, Subject} from 'rxjs';
-import {take, takeUntil} from 'rxjs/operators';
+import {Observable, Subscription} from 'rxjs';
 
 /**
  * Like `ResourceOptions` but uses an RxJS-based `loader`.
  *
  * @experimental
  */
-export interface RxResourceOptions<T, R> extends Omit<ResourceOptions<T, R>, 'loader'> {
+export interface RxResourceOptions<T, R> extends BaseResourceOptions<T, R> {
   loader: (params: ResourceLoaderParams<R>) => Observable<T>;
 }
 
@@ -35,22 +36,37 @@ export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T |
   opts?.injector || assertInInjectionContext(rxResource);
   return resource<T, R>({
     ...opts,
-    loader: (params) => {
-      const cancelled = new Subject<void>();
-      params.abortSignal.addEventListener('abort', () => cancelled.next());
-
-      // Note: this is identical to `firstValueFrom` which we can't use,
-      // because at the time of writing, `core` still supports rxjs 6.x.
-      return new Promise<T>((resolve, reject) => {
-        opts
-          .loader(params)
-          .pipe(take(1), takeUntil(cancelled))
-          .subscribe({
-            next: resolve,
-            error: reject,
-            complete: () => reject(new Error('Resource completed before producing a value')),
-          });
+    stream: (params) => {
+      let sub: Subscription;
+
+      // Track the abort listener so it can be removed if the Observable completes (as a memory
+      // optimization).
+      const onAbort = () => sub.unsubscribe();
+      params.abortSignal.addEventListener('abort', onAbort);
+
+      // Start off stream as undefined.
+      const stream = signal<{value: T} | {error: unknown}>({value: undefined as T});
+      let resolve: ((value: Signal<{value: T} | {error: unknown}>) => void) | undefined;
+      const promise = new Promise<Signal<{value: T} | {error: unknown}>>((r) => (resolve = r));
+
+      function send(value: {value: T} | {error: unknown}): void {
+        stream.set(value);
+        resolve?.(stream);
+        resolve = undefined;
+      }
+
+      sub = opts.loader(params).subscribe({
+        next: (value) => send({value}),
+        error: (error) => send({error}),
+        complete: () => {
+          if (resolve) {
+            send({error: new Error('Resource completed before producing a value')});
+          }
+          params.abortSignal.removeEventListener('abort', onAbort);
+        },
       });
+
+      return promise;
     },
   });
 }
diff --git a/packages/core/rxjs-interop/test/rx_resource_spec.ts b/packages/core/rxjs-interop/test/rx_resource_spec.ts
@@ -6,7 +6,7 @@
  * found in the LICENSE file at https://angular.dev/license
  */
 
-import {of, Observable} from 'rxjs';
+import {of, Observable, BehaviorSubject} from 'rxjs';
 import {TestBed} from '@angular/core/testing';
 import {ApplicationRef, Injector, signal} from '@angular/core';
 import {rxResource} from '@angular/core/rxjs-interop';
@@ -55,6 +55,27 @@ describe('rxResource()', () => {
     await appRef.whenStable();
     expect(unsub).toBe(true);
   });
+
+  it('should stream when the loader returns multiple values', async () => {
+    const injector = TestBed.inject(Injector);
+    const appRef = TestBed.inject(ApplicationRef);
+    const response = new BehaviorSubject(1);
+    const res = rxResource({
+      loader: () => response,
+      injector,
+    });
+    await appRef.whenStable();
+    expect(res.value()).toBe(1);
+
+    response.next(2);
+    expect(res.value()).toBe(2);
+
+    response.next(3);
+    expect(res.value()).toBe(3);
+
+    response.error('fail');
+    expect(res.error()).toBe('fail');
+  });
 });
 
 async function waitFor(fn: () => boolean): Promise<void> {
diff --git a/packages/core/src/resource/api.ts b/packages/core/src/resource/api.ts
@@ -160,12 +160,21 @@ export interface ResourceLoaderParams<R> {
  */
 export type ResourceLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLike<T>;
 
+/**
+ * Streaming loader for a `Resource`.
+ *
+ * @experimental
+ */
+export type ResourceStreamingLoader<T, R> = (
+  param: ResourceLoaderParams<R>,
+) => PromiseLike<Signal<{value: T} | {error: unknown}>>;
+
 /**
  * Options to the `resource` function, for creating a resource.
  *
  * @experimental
  */
-export interface ResourceOptions<T, R> {
+export interface BaseResourceOptions<T, R> {
   /**
    * A reactive function which determines the request to be made. Whenever the request changes, the
    * loader will be triggered to fetch a new value for the resource.
@@ -174,11 +183,6 @@ export interface ResourceOptions<T, R> {
    */
   request?: () => R;
 
-  /**
-   * Loading function which returns a `Promise` of the resource's value for a given request.
-   */
-  loader: ResourceLoader<T, R>;
-
   /**
    * Equality function used to compare the return value of the loader.
    */
@@ -189,3 +193,33 @@ export interface ResourceOptions<T, R> {
    */
   injector?: Injector;
 }
+
+/**
+ * Options to the `resource` function, for creating a resource.
+ *
+ * @experimental
+ */
+export interface PromiseResourceOptions<T, R> extends BaseResourceOptions<T, R> {
+  /**
+   * Loading function which returns a `Promise` of the resource's value for a given request.
+   */
+  loader: ResourceLoader<T, R>;
+}
+
+/**
+ * Options to the `resource` function, for creating a resource.
+ *
+ * @experimental
+ */
+export interface StreamingResourceOptions<T, R> extends BaseResourceOptions<T, R> {
+  /**
+   * Loading function which returns a `Promise` of a signal of the resource's value for a given
+   * request, which can change over time as new values are received from a stream.
+   */
+  stream: ResourceStreamingLoader<T, R>;
+}
+
+/**
+ * @experimental
+ */
+export type ResourceOptions<T, R> = PromiseResourceOptions<T, R> | StreamingResourceOptions<T, R>;
diff --git a/packages/core/src/resource/resource.ts b/packages/core/src/resource/resource.ts
diff --git a/packages/core/test/resource/resource_spec.ts b/packages/core/test/resource/resource_spec.ts
