fix(fetch): don't leak event listeners added to passed-in signals

spec/observables/dom/fetch-spec.ts

@@ -206,7 +206,7 @@ describe('fromFetch', () => {
     expect(mockFetch.calls[0].init!.method).to.equal('HEAD');
   });
 
-  it('should pass in a signal with the init object without mutating the init', done => {
+  it('should add a signal to internal init object without mutating the passed init object', done => {
     const myInit = {method: 'DELETE'};
     const fetch$ = fromFetch('/bar', myInit);
     fetch$.subscribe({
@@ -249,4 +249,19 @@ describe('fromFetch', () => {
     // The subscription will not be closed until the error fires when the promise resolves.
     expect(subscription.closed).to.be.false;
   });
+
+  it('should not leak listeners added to the passed in signal', done => {
+    const controller = new MockAbortController();
+    const signal = controller.signal as any;
+    const fetch$ = fromFetch('/foo', { signal });
+    const subscription = fetch$.subscribe();
+    subscription.add(() => {
+      try {
+        expect(signal._listeners).to.be.empty;
+        done();
+      } catch (error) {
+        done(error);
+      }
+    });
+  });
 });

src/internal/observable/dom/fetch.ts

@@ -1,4 +1,5 @@
 import { Observable } from '../../Observable';
+import { Subscription } from '../../Subscription';
 
 /**
  * Uses [the Fetch API](https://developer.mozilla.org/en-US/docs/Web/API/Fetch_API) to
@@ -54,23 +55,32 @@ export function fromFetch(input: string | Request, init?: RequestInit): Observab
   return new Observable<Response>(subscriber => {
     const controller = new AbortController();
     const signal = controller.signal;
-    let outerSignalHandler: () => void;
     let abortable = true;
     let unsubscribed = false;
 
+    const subscription = new Subscription();
+    subscription.add(() => {
+      unsubscribed = true;
+      if (abortable) {
+        controller.abort();
+      }
+    });
+
     let perSubscriberInit: RequestInit;
     if (init) {
       // If a signal is provided, just have it teardown. It's a cancellation token, basically.
       if (init.signal) {
         if (init.signal.aborted) {
           controller.abort();
         } else {
-          outerSignalHandler = () => {
+          const outerSignal = init.signal;
+          const outerSignalHandler = () => {
             if (!signal.aborted) {
               controller.abort();
             }
           };
-          init.signal.addEventListener('abort', outerSignalHandler);
+          outerSignal.addEventListener('abort', outerSignalHandler);
+          subscription.add(() => outerSignal.removeEventListener('abort', outerSignalHandler));
         }
       }
       // init cannot be mutated or reassigned as it's closed over by the
@@ -92,11 +102,6 @@ export function fromFetch(input: string | Request, init?: RequestInit): Observab
       }
     });
 
-    return () => {
-      unsubscribed = true;
-      if (abortable) {
-        controller.abort();
-      }
-    };
+    return subscription;
   });
 }