fix(signals): use Injector of rxMethod instance caller if available

modules/signals/rxjs-interop/spec/rx-method.spec.ts

@@ -1,13 +1,20 @@
 import {
+  Component,
   createEnvironmentInjector,
   EnvironmentInjector,
+  inject,
   Injectable,
+  Injector,
+  OnInit,
   signal,
 } from '@angular/core';
 import { TestBed } from '@angular/core/testing';
-import { BehaviorSubject, pipe, Subject, tap } from 'rxjs';
+import { BehaviorSubject, finalize, pipe, Subject, tap } from 'rxjs';
 import { rxMethod } from '../src';
 import { createLocalService } from '../../spec/helpers';
+import { provideRouter } from '@angular/router';
+import { provideLocationMocks } from '@angular/common/testing';
+import { RouterTestingHarness } from '@angular/router/testing';
 
 describe('rxMethod', () => {
   it('runs with a value', () => {
@@ -231,4 +238,170 @@ describe('rxMethod', () => {
     TestBed.flushEffects();
     expect(counter()).toBe(4);
   });
+
+  it('completes on manual destroy with Signals', () => {
+    TestBed.runInInjectionContext(() => {
+      let completed = false;
+      const counter = signal(1);
+      const fn = rxMethod<number>(finalize(() => (completed = true)));
+      TestBed.flushEffects();
+      fn(counter);
+      fn.unsubscribe();
+      expect(completed).toBe(true);
+    });
+  });
+
+  /**
+   * This test suite verifies that the internal effect of
+   * an RxMethod instance is executed with the correct injector
+   * and is destroyed at the specified time.
+   *
+   * Since we cannot directly observe the destruction of the effect from the outside,
+   * we test it indirectly.
+   *
+   * Components use the globalSignal from GlobalService and pass it
+   * to the `log` method. If the component is destroyed but a subsequent
+   * Signal change still increases the `globalSignalChangerCounter`,
+   * it indicates that the internal effect is still active.
+   */
+  describe('Internal effect for Signal tracking', () => {
+    @Injectable({ providedIn: 'root' })
+    class GlobalService {
+      globalSignal = signal(1);
+      globalSignalChangeCounter = 0;
+
+      log = rxMethod<number>(pipe(tap(() => this.globalSignalChangeCounter++)));
+    }
+
+    @Component({
+      selector: `app-storeless`,
+      template: ``,
+      standalone: true,
+    })
+    class WithoutStoreComponent {}
+
+    function setup(WithStoreComponent: new () => unknown): GlobalService {
+      TestBed.configureTestingModule({
+        providers: [
+          provideRouter([
+            { path: 'with-store', component: WithStoreComponent },
+            {
+              path: 'without-store',
+              component: WithoutStoreComponent,
+            },
+          ]),
+          provideLocationMocks(),
+        ],
+      });
+
+      return TestBed.inject(GlobalService);
+    }
+
+    it('it tracks the Signal when component is active', async () => {
+      @Component({
+        selector: 'app-with-store',
+        template: ``,
+        standalone: true,
+      })
+      class WithStoreComponent {
+        store = inject(GlobalService);
+
+        constructor() {
+          this.store.log(this.store.globalSignal);
+        }
+      }
+
+      const globalService = setup(WithStoreComponent);
+
+      await RouterTestingHarness.create('/with-store');
+      expect(globalService.globalSignalChangeCounter).toBe(1);
+
+      globalService.globalSignal.update((value) => value + 1);
+      TestBed.flushEffects();
+      expect(globalService.globalSignalChangeCounter).toBe(2);
+
+      globalService.globalSignal.update((value) => value + 1);
+      TestBed.flushEffects();
+      expect(globalService.globalSignalChangeCounter).toBe(3);
+    });
+
+    it('destroys with component injector when rxMethod is in root and RxMethod in component', async () => {
+      @Component({
+        selector: 'app-with-store',
+        template: ``,
+        standalone: true,
+      })
+      class WithStoreComponent {
+        store = inject(GlobalService);
+
+        constructor() {
+          this.store.log(this.store.globalSignal);
+        }
+      }
+
+      const globalService = setup(WithStoreComponent);
+
+      const harness = await RouterTestingHarness.create('/with-store');
+
+      // effect is destroyed → Signal is not tracked anymore
+      await harness.navigateByUrl('/without-store');
+      globalService.globalSignal.update((value) => value + 1);
+      TestBed.flushEffects();
+
+      expect(globalService.globalSignalChangeCounter).toBe(1);
+    });
+
+    it("falls back to rxMethod's injector when RxMethod's call is outside of injection context", async () => {
+      @Component({
+        selector: `app-store`,
+        template: ``,
+        standalone: true,
+      })
+      class WithStoreComponent implements OnInit {
+        store = inject(GlobalService);
+
+        ngOnInit() {
+          this.store.log(this.store.globalSignal);
+        }
+      }
+
+      const globalService = setup(WithStoreComponent);
+
+      const harness = await RouterTestingHarness.create('/with-store');
+
+      // Signal is still tracked because RxMethod injector is used
+      await harness.navigateByUrl('/without-store');
+      globalService.globalSignal.update((value) => value + 1);
+      TestBed.flushEffects();
+
+      expect(globalService.globalSignalChangeCounter).toBe(2);
+    });
+
+    it('provides the injector for RxMethod on call', async () => {
+      @Component({
+        selector: `app-store`,
+        template: ``,
+        standalone: true,
+      })
+      class WithStoreComponent implements OnInit {
+        store = inject(GlobalService);
+        injector = inject(Injector);
+
+        ngOnInit() {
+          this.store.log(this.store.globalSignal, { injector: this.injector });
+        }
+      }
+
+      const globalService = setup(WithStoreComponent);
+
+      const harness = await RouterTestingHarness.create('/with-store');
+
+      // effect is destroyed → Signal is not tracked anymore
+      await harness.navigateByUrl('/without-store');
+      globalService.globalSignal.update((value) => value + 1);
+      TestBed.flushEffects();
+
+      expect(globalService.globalSignalChangeCounter).toBe(1);
+    });
+  });
 });

modules/signals/rxjs-interop/src/rx-method.ts

@@ -11,7 +11,8 @@ import {
 import { isObservable, noop, Observable, Subject, Unsubscribable } from 'rxjs';
 
 type RxMethod<Input> = ((
-  input: Input | Signal<Input> | Observable<Input>
+  input: Input | Signal<Input> | Observable<Input>,
+  config?: { injector?: Injector }
 ) => Unsubscribable) &
   Unsubscribable;
 
@@ -30,14 +31,21 @@ export function rxMethod<Input>(
   const sourceSub = generator(source$).subscribe();
   destroyRef.onDestroy(() => sourceSub.unsubscribe());
 
-  const rxMethodFn = (input: Input | Signal<Input> | Observable<Input>) => {
+  const rxMethodFn = (
+    input: Input | Signal<Input> | Observable<Input>,
+    config?: { injector?: Injector }
+  ) => {
     if (isSignal(input)) {
+      const callerInjector = getCallerInjectorIfAvailable();
+      const customInjector = config?.injector;
+      const instanceInjector = customInjector ?? callerInjector ?? injector;
+
       const watcher = effect(
         () => {
           const value = input();
           untracked(() => source$.next(value));
         },
-        { injector }
+        { injector: instanceInjector }
       );
       const instanceSub = { unsubscribe: () => watcher.destroy() };
       sourceSub.add(instanceSub);
@@ -59,3 +67,11 @@ export function rxMethod<Input>(
 
   return rxMethodFn;
 }
+
+function getCallerInjectorIfAvailable(): Injector | null {
+  try {
+    return inject(Injector);
+  } catch (e) {
+    return null;
+  }
+}