feat(signals): add rxjs-interop subpackage (#4061)

modules/signals/index.ts

@@ -1,7 +1 @@
-/**
- * DO NOT EDIT
- *
- * This file is automatically generated at build
- */
-
-export * from './public_api';
+export * from './src/index';

modules/signals/package.json

@@ -21,7 +21,13 @@
   },
   "homepage": "https://github.com/ngrx/platform#readme",
   "peerDependencies": {
-    "@angular/core": "^16.0.0"
+    "@angular/core": "^16.0.0",
+    "rxjs": "^6.5.3 || ^7.4.0"
+  },
+  "peerDependenciesMeta": {
+    "rxjs": {
+      "optional": true
+    }
   },
   "schematics": "./schematics/collection.json",
   "sideEffects": false,

modules/signals/project.json

@@ -35,7 +35,9 @@
       "options": {
         "lintFilePatterns": [
           "modules/signals/*/**/*.ts",
-          "modules/signals/*/**/*.html"
+          "modules/signals/*/**/*.html",
+          "modules/signals/rxjs-interop/**/*.ts",
+          "modules/signals/rxjs-interop/**/*.html"
         ]
       },
       "outputs": ["{options.outputFile}"]

modules/signals/rxjs-interop/ng-package.json

@@ -0,0 +1,5 @@
+{
+  "lib": {
+    "entryFile": "index.ts"
+  }
+}

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

@@ -0,0 +1,177 @@
+import { Injectable, signal } from '@angular/core';
+import { TestBed } from '@angular/core/testing';
+import { BehaviorSubject, pipe, Subject, tap } from 'rxjs';
+import { rxMethod } from '../src';
+import { createLocalService, testEffects } from '../../spec/helpers';
+
+describe('rxMethod', () => {
+  it('runs with a value', () => {
+    const results: number[] = [];
+    const method = TestBed.runInInjectionContext(() =>
+      rxMethod<number>(pipe(tap((value) => results.push(value))))
+    );
+
+    method(1);
+    expect(results.length).toBe(1);
+    expect(results[0]).toBe(1);
+
+    method(2);
+    expect(results.length).toBe(2);
+    expect(results[1]).toBe(2);
+  });
+
+  it('runs with an observable', () => {
+    const results: string[] = [];
+    const method = TestBed.runInInjectionContext(() =>
+      rxMethod<string>(pipe(tap((value) => results.push(value))))
+    );
+    const subject$ = new Subject<string>();
+
+    method(subject$);
+    expect(results.length).toBe(0);
+
+    subject$.next('ngrx');
+    expect(results[0]).toBe('ngrx');
+
+    subject$.next('rocks');
+    expect(results[1]).toBe('rocks');
+  });
+
+  it(
+    'runs with a signal',
+    testEffects((tick) => {
+      const results: number[] = [];
+      const method = rxMethod<number>(
+        pipe(tap((value) => results.push(value)))
+      );
+      const sig = signal(1);
+
+      method(sig);
+      expect(results.length).toBe(0);
+
+      tick();
+      expect(results[0]).toBe(1);
+
+      sig.set(10);
+      expect(results.length).toBe(1);
+
+      tick();
+      expect(results[1]).toBe(10);
+    })
+  );
+
+  it('runs with void input', () => {
+    const results: number[] = [];
+    const subject$ = new Subject<void>();
+    const method = TestBed.runInInjectionContext(() =>
+      rxMethod<void>(pipe(tap(() => results.push(1))))
+    );
+
+    method();
+    expect(results.length).toBe(1);
+
+    method(subject$);
+    expect(results.length).toBe(1);
+
+    subject$.next();
+    expect(results.length).toBe(2);
+  });
+
+  it(
+    'manually unsubscribes from method instance',
+    testEffects((tick) => {
+      const results: number[] = [];
+      const method = rxMethod<number>(
+        pipe(tap((value) => results.push(value)))
+      );
+      const subject$ = new Subject<number>();
+      const sig = signal(0);
+
+      const sub1 = method(subject$);
+      const sub2 = method(sig);
+      expect(results).toEqual([]);
+
+      subject$.next(1);
+      sig.set(1);
+      tick();
+      expect(results).toEqual([1, 1]);
+
+      sub1.unsubscribe();
+      subject$.next(2);
+      sig.set(2);
+      tick();
+      expect(results).toEqual([1, 1, 2]);
+
+      sub2.unsubscribe();
+      sig.set(3);
+      tick();
+      expect(results).toEqual([1, 1, 2]);
+    })
+  );
+
+  it('manually unsubscribes from method and all instances', () => {
+    const results: number[] = [];
+    let destroyed = false;
+    const method = TestBed.runInInjectionContext(() =>
+      rxMethod<number>(
+        pipe(
+          tap({
+            next: (value) => results.push(value),
+            finalize: () => (destroyed = true),
+          })
+        )
+      )
+    );
+    const subject1$ = new BehaviorSubject(1);
+    const subject2$ = new BehaviorSubject(1);
+
+    method(subject1$);
+    method(subject2$);
+    method(1);
+    expect(results).toEqual([1, 1, 1]);
+
+    method.unsubscribe();
+    expect(destroyed).toBe(true);
+
+    subject1$.next(2);
+    subject2$.next(2);
+    method(2);
+    expect(results).toEqual([1, 1, 1]);
+  });
+
+  it('unsubscribes from method and all instances on destroy', () => {
+    const results: number[] = [];
+    let destroyed = false;
+    const subject$ = new BehaviorSubject(1);
+    const sig = signal(1);
+
+    @Injectable()
+    class TestService {
+      method = rxMethod<number>(
+        pipe(
+          tap({
+            next: (value) => results.push(value),
+            finalize: () => (destroyed = true),
+          })
+        )
+      );
+    }
+
+    const { service, tick, destroy } = createLocalService(TestService);
+
+    service.method(subject$);
+    service.method(sig);
+    service.method(1);
+    tick();
+    expect(results).toEqual([1, 1, 1]);
+
+    destroy();
+    expect(destroyed).toBe(true);
+
+    subject$.next(2);
+    sig.set(2);
+    service.method(2);
+    tick();
+    expect(results).toEqual([1, 1, 1]);
+  });
+});

modules/signals/rxjs-interop/src/index.ts

@@ -0,0 +1 @@
+export { rxMethod } from './rx-method';

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

@@ -0,0 +1,51 @@
+import {
+  assertInInjectionContext,
+  DestroyRef,
+  inject,
+  Injector,
+  isSignal,
+  Signal,
+} from '@angular/core';
+import { toObservable } from '@angular/core/rxjs-interop';
+import { isObservable, Observable, of, Subject, Unsubscribable } from 'rxjs';
+
+type RxMethodInput<Input> = Input | Observable<Input> | Signal<Input>;
+
+type RxMethod<Input> = ((input: RxMethodInput<Input>) => Unsubscribable) &
+  Unsubscribable;
+
+export function rxMethod<Input>(
+  generator: (source$: Observable<Input>) => Observable<unknown>,
+  config?: { injector?: Injector }
+): RxMethod<Input> {
+  if (!config?.injector) {
+    assertInInjectionContext(rxMethod);
+  }
+
+  const injector = config?.injector ?? inject(Injector);
+  const destroyRef = injector.get(DestroyRef);
+  const source$ = new Subject<Input>();
+
+  const sourceSub = generator(source$).subscribe();
+  destroyRef.onDestroy(() => sourceSub.unsubscribe());
+
+  const rxMethodFn = (input: RxMethodInput<Input>) => {
+    let input$: Observable<Input>;
+
+    if (isSignal(input)) {
+      input$ = toObservable(input, { injector });
+    } else if (isObservable(input)) {
+      input$ = input;
+    } else {
+      input$ = of(input);
+    }
+
+    const instanceSub = input$.subscribe((value) => source$.next(value));
+    sourceSub.add(instanceSub);
+
+    return instanceSub;
+  };
+  rxMethodFn.unsubscribe = sourceSub.unsubscribe.bind(sourceSub);
+
+  return rxMethodFn;
+}

modules/signals/spec/helpers.ts

@@ -14,27 +14,29 @@ export function testEffects(testFn: (tick: () => void) => void): () => void {
   };
 }
 
-export function createLocalStore<Store extends Type<unknown>>(
-  storeToken: Store
+export function createLocalService<Service extends Type<unknown>>(
+  serviceToken: Service
 ): {
-  store: InstanceType<Store>;
+  service: InstanceType<Service>;
+  tick: () => void;
   destroy: () => void;
 } {
   @Component({
     standalone: true,
     template: '',
-    providers: [storeToken],
+    providers: [serviceToken],
   })
   class TestComponent {
-    store = inject(storeToken);
+    service = inject(serviceToken);
   }
 
   const fixture = TestBed.configureTestingModule({
     imports: [TestComponent],
   }).createComponent(TestComponent);
 
   return {
-    store: fixture.componentInstance.store,
+    service: fixture.componentInstance.service,
+    tick: () => fixture.detectChanges(),
     destroy: () => fixture.destroy(),
   };
 }

modules/signals/spec/signal-store.spec.ts

@@ -8,7 +8,7 @@ import {
   withState,
 } from '../src';
 import { STATE_SIGNAL } from '../src/signal-state';
-import { createLocalStore } from './helpers';
+import { createLocalService } from './helpers';
 
 describe('signalStore', () => {
   describe('creation', () => {
@@ -190,7 +190,7 @@ describe('signalStore', () => {
         })
       );
 
-      createLocalStore(Store).destroy();
+      createLocalService(Store).destroy();
 
       expect(message).toBe('onDestroy');
     });
@@ -233,7 +233,7 @@ describe('signalStore', () => {
         })
       );
 
-      createLocalStore(Store).destroy();
+      createLocalService(Store).destroy();
 
       expect(message).toBe('onDestroy');
     });
@@ -256,7 +256,7 @@ describe('signalStore', () => {
           },
         })
       );
-      const { destroy } = createLocalStore(Store);
+      const { destroy } = createLocalService(Store);
 
       expect(messages).toEqual(['onInit']);
 

modules/signals/tsconfig.build.json

@@ -22,7 +22,7 @@
     "noImplicitReturns": true,
     "noFallthroughCasesInSwitch": true
   },
-  "files": ["public_api.ts"],
+  "files": ["index.ts"],
   "include": ["**/*.ts"],
   "exclude": ["**/*.spec.ts"],
   "angularCompilerOptions": {

tsconfig.json

@@ -43,6 +43,7 @@
         "./modules/schematics/schematics-core"
       ],
       "@ngrx/signals": ["./modules/signals"],
+      "@ngrx/signals/rxjs-interop": ["./modules/signals/rxjs-interop"],
       "@ngrx/signals/schematics-core": ["./modules/signals/schematics-core"],
       "@ngrx/store": ["./modules/store"],
       "@ngrx/store-devtools": ["./modules/store-devtools"],