angular · JeanMeche · Mar 2, 2026 · Mar 3, 2026 · alan-agius4 · Mar 3, 2026
diff --git a/goldens/public-api/core/index.api.md b/goldens/public-api/core/index.api.md
@@ -186,6 +186,7 @@ export interface BaseResourceOptions<T, R> {
     equal?: ValueEqualityFn<T>;
     injector?: Injector;
     params?: (ctx: ResourceParamsContext) => R;
+    transferCacheKey?: (params: R) => StateKey<T>;
 }
 
 // @public
@@ -1725,7 +1726,7 @@ export type ResourceSnapshot<T> = {
 export type ResourceStatus = 'idle' | 'error' | 'loading' | 'reloading' | 'resolved' | 'local';
 
 // @public
-export type ResourceStreamingLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLike<Signal<ResourceStreamItem<T>>>;
+export type ResourceStreamingLoader<T, R> = (param: ResourceLoaderParams<R>) => Signal<ResourceStreamItem<T>> | PromiseLike<Signal<ResourceStreamItem<T>>> | undefined;
 
 // @public (undocumented)
 export type ResourceStreamItem<T> = {

diff --git a/packages/common/http/src/resource.ts b/packages/common/http/src/resource.ts
@@ -434,6 +434,7 @@ class HttpResourceImpl<T>
       equal,
       debugName,
       injector,
+      undefined,
       getInitialStream,
     );
     this.client = injector.get(HttpClient);

diff --git a/packages/core/rxjs-interop/src/rx_resource.ts b/packages/core/rxjs-interop/src/rx_resource.ts
@@ -6,19 +6,19 @@
  * found in the LICENSE file at https://angular.dev/license
  */
 
+import {Observable, Subscription} from 'rxjs';
 import {
   assertInInjectionContext,
+  BaseResourceOptions,
   resource,
   ResourceLoaderParams,
   ResourceRef,
+  ResourceStreamItem,
   Signal,
   signal,
-  BaseResourceOptions,
   ɵRuntimeError,
   ɵRuntimeErrorCode,
-  ResourceStreamItem,
 } from '../../src/core';
-import {Observable, Subscription} from 'rxjs';
 import {encapsulateResourceError} from '../../src/resource/resource';
 
 /**
@@ -75,8 +75,7 @@ export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T |
         resolve = undefined;
       }
 
-      // TODO(alxhub): remove after g3 updated to rename loader -> stream
-      const streamFn = opts.stream ?? (opts as {loader?: RxResourceOptions<T, R>['stream']}).loader;
+      const streamFn = opts.stream;
       if (streamFn === undefined) {
         throw new ɵRuntimeError(
           ɵRuntimeErrorCode.MUST_PROVIDE_STREAM_OPTION,
@@ -103,6 +102,10 @@ export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T |
         },
       });
 
+      if (resolve === undefined) {
+        return stream;
+      }
+
       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,21 +6,24 @@
  * found in the LICENSE file at https://angular.dev/license
  */
 
-import {of, Observable, BehaviorSubject, throwError} from 'rxjs';
+import {BehaviorSubject, EMPTY, Observable, of, Subscriber, throwError} from 'rxjs';
+import {ApplicationRef, Injector, makeStateKey, signal, TransferState} from '../../src/core';
 import {TestBed} from '../../testing';
-import {ApplicationRef, Injector, signal} from '../../src/core';
 import {rxResource} from '../src';
 
 describe('rxResource()', () => {
   it('should fetch data using an observable loader', async () => {
     const injector = TestBed.inject(Injector);
-    const appRef = TestBed.inject(ApplicationRef);
     const res = rxResource({
       stream: () => of(1),
       injector,
     });
-    await appRef.whenStable();
+
+    TestBed.tick();
+
+    // Value should be available synchronously (because the observable emits synchronously)
     expect(res.value()).toBe(1);
+    expect(res.status()).toBe('resolved');
   });
 
   it('should cancel the fetch when a new request comes in', async () => {
@@ -29,7 +32,7 @@ describe('rxResource()', () => {
     const request = signal(1);
     let unsub = false;
     let lastSeenRequest: number = 0;
-    rxResource({
+    const res = rxResource({
       params: request,
       stream: ({params: request}) => {
         lastSeenRequest = request;
@@ -47,12 +50,23 @@ describe('rxResource()', () => {
       injector,
     });
 
+    // The stream isn't evaluated eagerly. We have to wait for the effect to run to see the first request.
+    expect(lastSeenRequest).toBe(0);
+
+    TestBed.tick();
+
+    expect(res.status()).toBe('loading');
+    expect(lastSeenRequest).toBe(1);
+
     // Wait for the resource to reach loading state.
     await waitFor(() => lastSeenRequest === 1);
 
     // Setting request = 2 should cancel request = 1
     request.set(2);
+    // The stream is updated asynchronously because we're waiting for the effect to fire.
+    expect(lastSeenRequest).toBe(1);
     await appRef.whenStable();
+    expect(lastSeenRequest).toBe(2);
     expect(unsub).toBe(true);
   });
 
@@ -113,10 +127,158 @@ describe('rxResource()', () => {
 
     expect(() => rxRes.value()).toThrowError(/This is a FooError/);
   });
+
+  it('should reuse observable', async () => {
+    let count = 0;
+    let sub!: Subscriber<number>;
+    const obs = new Observable<number>((s) => {
+      sub = s;
+      count++;
+    });
+
+    const res = rxResource({
+      stream: () => obs,
+      injector: TestBed.inject(Injector),
+    });
+    // Hasn't subscribed to the observable yet
+    expect(count).toBe(0);
+
+    TestBed.tick();
+    expect(count).toBe(1);
+    expect(res.status()).toBe('loading');
+    sub.next(1);
+    await new Promise((resolve) => setTimeout(resolve, 0));
+
+    expect(count).toBe(1);
+  });
+
+  it('should report error synchronously (after tick)', () => {
+    const injector = TestBed.inject(Injector);
+    const res = rxResource({
+      stream: () => EMPTY,
+      injector,
+    });
+    TestBed.tick();
+    expect(res.status()).toBe('error');
+    expect(res.error()).toBeInstanceOf(Error);
+    expect(() => res.value()).toThrowError(/Resource completed before producing a value/);
+  });
+
+  it('should report sync error synchronously (after tick) ', () => {
+    const injector = TestBed.inject(Injector);
+    const res = rxResource({
+      stream: () => throwError(() => new Error('bad news')),
+      injector,
+    });
+    TestBed.tick();
+    expect(res.status()).toBe('error');
+    expect(res.error()).toBeInstanceOf(Error);
+    expect(() => res.value()).toThrowError(/bad news/);
+  });
+
+  describe('with TransferState', () => {
+    let transferState: TransferState;
+
+    beforeEach(() => {
+      TestBed.configureTestingModule({providers: [TransferState]});
+      transferState = TestBed.inject(TransferState);
+    });
+
+    afterEach(() => {
+      (globalThis as any).ngServerMode = undefined;
+    });
+
+    it('should read from TransferState if a key is present', async () => {
+      const key = makeStateKey<number>('test-key');
+      transferState.set(key, 123);
+
+      const injector = TestBed.inject(Injector);
+      const testResource = rxResource({
+        stream: () => of(456),
+        transferCacheKey: () => key,
+        injector,
+      });
+
+      // Should be synchronously resolved from cache
+      expect(testResource.status()).toBe('resolved');
+      expect(testResource.value()).toBe(123);
+
+      // Should prevent loader from running
+      await flushMicrotasks();
+      expect(testResource.value()).toBe(123);
+    });
+
+    it('should write to TransferState on server when resolved (sync)', async () => {
+      (globalThis as any).ngServerMode = true;
+      const key = makeStateKey<number>('server-key');
+
+      const injector = TestBed.inject(Injector);
+      const testResource = rxResource({
+        stream: () => of(789),
+        transferCacheKey: () => key,
+        injector,
+      });
+
+      expect(testResource.status()).toBe('loading');
+
+      await flushMicrotasks();
+
+      expect(testResource.status()).toBe('resolved');
+      expect(testResource.value()).toBe(789);
+      expect(transferState.get(key, null!)).toBe(789);
+    });
+
+    it('should write to TransferState on server when resolved (async)', async () => {
+      (globalThis as any).ngServerMode = true;
+      const key = makeStateKey<number>('server-async-key');
+
+      const injector = TestBed.inject(Injector);
+      const testResource = rxResource({
+        stream: () =>
+          new Observable<number>((sub) => {
+            Promise.resolve().then(() => {
+              sub.next(101112);
+              sub.complete();
+            });
+          }),
+        transferCacheKey: () => key,
+        injector,
+      });
+
+      expect(testResource.status()).toBe('loading');
+
+      await waitFor(() => testResource.status() === 'resolved');
+
+      expect(testResource.value()).toBe(101112);
+      expect(transferState.get(key, null!)).toBe(101112);
+    });
+
+    it('should not write to TransferState on client when resolved', async () => {
+      (globalThis as any).ngServerMode = false;
+      const key = makeStateKey<number>('client-key');
+
+      const injector = TestBed.inject(Injector);
+      const testResource = rxResource({
+        stream: () => of(131415),
+        transferCacheKey: () => key,
+        injector,
+      });
+
+      await flushMicrotasks();
+
+      expect(testResource.status()).toBe('resolved');
+      expect(testResource.value()).toBe(131415);
+      expect(transferState.hasKey(key)).toBeFalse();
+    });
+  });
 });
 
 async function waitFor(fn: () => boolean): Promise<void> {
   while (!fn()) {
     await new Promise((resolve) => setTimeout(resolve, 1));
   }
 }
+
+function flushMicrotasks(): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, 0));
+}
diff --git a/packages/core/src/resource/api.ts b/packages/core/src/resource/api.ts
@@ -9,6 +9,7 @@
 import {Injector} from '../di/injector';
 import {Signal, ValueEqualityFn} from '../render3/reactivity/api';
 import {WritableSignal} from '../render3/reactivity/signal';
+import {StateKey} from '../transfer_state';
 
 /** Error thrown when a `Resource` dependency of another resource errors. */
 export class ResourceDependencyError extends Error {
@@ -200,7 +201,7 @@ export type ResourceLoader<T, R> = (param: ResourceLoaderParams<R>) => PromiseLi
  */
 export type ResourceStreamingLoader<T, R> = (
   param: ResourceLoaderParams<R>,
-) => PromiseLike<Signal<ResourceStreamItem<T>>>;
+) => Signal<ResourceStreamItem<T>> | PromiseLike<Signal<ResourceStreamItem<T>>> | undefined;
 
 /**
  * Options to the `resource` function, for creating a resource.
@@ -231,6 +232,11 @@ export interface BaseResourceOptions<T, R> {
    * Overrides the `Injector` used by `resource`.
    */
   injector?: Injector;
+
+  /**
+   * The transfer cache key used to cache the resource data in the `TransferState` during server-side rendering and to retrieve it on the client side.
+   */
+  transferCacheKey?: (params: R) => StateKey<T>;
 }
 
 /**