fix(core): Prevent an error on cleanup when an rxResource stream threw before returning an Observable (angular#63342)

Before this commit, it was wrongly assumed that the stream subscription could not be `undefined`.

Fixes angular#63341

PR Close angular#63342

Author: wartab

packages/core/rxjs-interop/src/rx_resource.ts

@@ -55,11 +55,11 @@ export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T |
     ...opts,
     loader: undefined,
     stream: (params) => {
-      let sub: Subscription;
+      let sub: Subscription | undefined;
 
       // Track the abort listener so it can be removed if the Observable completes (as a memory
       // optimization).
-      const onAbort = () => sub.unsubscribe();
+      const onAbort = () => sub?.unsubscribe();
       params.abortSignal.addEventListener('abort', onAbort);
 
       // Start off stream as undefined.

packages/core/rxjs-interop/test/rx_resource_spec.ts

@@ -77,6 +77,17 @@ describe('rxResource()', () => {
     expect(res.error()).toEqual(jasmine.objectContaining({cause: 'fail'}));
     expect(res.error()!.message).toContain('Resource');
   });
+
+  it('should cleanup without error when the stream function threw an error', async () => {
+    const appRef = TestBed.inject(ApplicationRef);
+    const res = rxResource({
+      stream: () => {
+        throw 'oh no';
+      },
+      injector: appRef.injector,
+    });
+    await appRef.whenStable();
+  });
 });
 
 async function waitFor(fn: () => boolean): Promise<void> {