fix(derived-async): fix unsubscribing from old observable when new va…

…lue is not observable

libs/ngxtension/derived-async/src/derived-async.spec.ts

@@ -1,6 +1,6 @@
 import { Signal, signal } from '@angular/core';
-import { TestBed, fakeAsync, tick } from '@angular/core/testing';
-import { Observable, catchError, delay, map, of, startWith } from 'rxjs';
+import { fakeAsync, TestBed, tick } from '@angular/core/testing';
+import { catchError, delay, map, Observable, of, startWith, timer } from 'rxjs';
 import { tap } from 'rxjs/operators';
 import { derivedAsync } from './derived-async';
 
@@ -383,6 +383,40 @@ describe(derivedAsync.name, () => {
 		}));
 	});
 
+	describe('works with observables and raw values', () => {
+		it('waits for them to resolve', fakeAsync(() => {
+			TestBed.runInInjectionContext(() => {
+				const logs: number[] = [];
+				const value = signal(false);
+
+				const s = derivedAsync(() => {
+					return value() ? timer(0, 100) : null;
+				});
+
+				expect(s()).toEqual(undefined); // initial value
+				TestBed.flushEffects();
+				expect(s()).toEqual(null); // raw value
+
+				value.set(true);
+				TestBed.flushEffects();
+				expect(s()).toEqual(null); // still raw value
+
+				tick(); // wait for timer to resolve
+				expect(s()).toEqual(0);
+
+				tick(100); // wait for next timer tick
+				expect(s()).toEqual(1);
+
+				value.set(false);
+				TestBed.flushEffects();
+				expect(s()).toEqual(null); // raw value
+
+				tick(100); // still raw value
+				expect(s()).toEqual(null);
+			});
+		}));
+	});
+
 	describe('works with contextual observables + requireSync', () => {
 		it('and recovers from errors', fakeAsync(() => {
 			TestBed.runInInjectionContext(() => {

libs/ngxtension/derived-async/src/derived-async.ts

@@ -18,6 +18,7 @@ import {
 	exhaustAll,
 	isObservable,
 	mergeAll,
+	of,
 	switchAll,
 } from 'rxjs';
 
@@ -266,15 +267,14 @@ export function derivedAsync<T>(
 				return;
 			}
 
-			if (isObservable(newSource) || isPromise(newSource)) {
-				// we untrack the source$.next() so that we don't register other signals as dependencies
-				untracked(() => sourceEvent$.next(newSource));
-			} else {
-				// if the new source is not an observable or a promise, we set the value immediately
-				untracked(() =>
-					sourceValue.set({ kind: StateKind.Value, value: newSource as T }),
+			// we untrack the source$.next() so that we don't register other signals as dependencies
+			untracked(() => {
+				sourceEvent$.next(
+					isObservable(newSource) || isPromise(newSource)
+						? newSource
+						: of(newSource as T),
 				);
-			}
+			});
 		});
 
 		// we return a computed value that will return the current value