Signals ↔ RxJS · Angular Atlas

Signals ersetzen RxJS nicht — sie ergänzen es. Signals eignen sich hervorragend für Zustand (was jetzt angezeigt wird), RxJS für Events über die Zeit (was wann passiert). Das Paket @angular/core/rxjs-interop enthält die Brücken zwischen beiden Welten. Neu bei Observables? Fang mit Observables von null an an.

Signals	RxJS / Observable
Haben immer einen aktuellen Wert	Emittieren Werte über die Zeit (womöglich noch nichts emittiert)
Synchron: `count()`	Asynchron: `subscribe()`
Kein Unsubscribe nötig	Brauchen Unsubscribe (oder einen Operator, der abschließt)
Ableitung: `computed()`	Ableitung: `map`, `combineLatest`, `switchMap`…
Ideal für Templates und UI-Zustand	Ideal für Websockets, Drag, Retry, komplexes Debounce

toSignal(): Observable → Signal

import { toSignal } from '@angular/core/rxjs-interop';

export class Clock {
  // Automatisches Subscribe + automatisches Unsubscribe beim Zerstören der Komponente
  readonly seconds = toSignal(interval(1000), { initialValue: 0 });

  // Ohne initialValue → der Typ ist T | undefined
  readonly user = toSignal(this.auth.user$);            // Signal<User | undefined>

  // requireSync: das Observable emittiert synchron (BehaviorSubject, startWith) → kein undefined
  readonly theme = toSignal(this.theme$, { requireSync: true }); // Signal<Theme>

  // Route-Parameter als Signal (falls du withComponentInputBinding nicht nutzt)
  private readonly route = inject(ActivatedRoute);
  readonly id = toSignal(this.route.paramMap.pipe(map((p) => Number(p.get('id')))), {
    initialValue: 0,
  });
}

toObservable(): Signal → Observable

import { toObservable } from '@angular/core/rxjs-interop';

export class ProductSearch {
  readonly query = signal('');

  // Signal → Observable → RxJS-Operatoren → zurück zu einem Signal
  readonly results = toSignal(
    toObservable(this.query).pipe(
      debounceTime(300),
      distinctUntilChanged(),
      filter((q) => q.length >= 2),
      switchMap((q) => this.api.search(q).pipe(catchError(() => of([])))),
    ),
    { initialValue: [] },
  );
}
// toObservable emittiert ASYNCHRON (über einen Effect) und nur den Endwert,
// wenn das Signal im selben Tick mehrmals gesetzt wird.

LivetoSignal und toObservable

toSignal(interval(1000))0 s

toSignal(fromEvent(window, 'resize'))1024 px

computed über beideDesktop

toObservable(count).pipe(pairwise()) → Button drücken

Wo RxJS besser bleibt

Ein mousemove-Stream mit Throttle und Akkumulation (scan) ist in RxJS trivial und mit Signals umständlich. Das Endergebnis legst du für das Template in ein Signal.

LiveEvent-Streams: throttleTime + scan

Beweg die Maus hierher

Position: 0, 0 · verarbeitete Events (Throttle 40 ms): 0

takeUntilDestroyed(): Abschied vom Subject destroy$

Jedes manuelle subscribe() in einer Komponente muss beim Zerstören beendet werden. takeUntilDestroyed() erledigt das automatisch mithilfe von DestroyRef. (Bei toSignal brauchst du es nicht — es meldet sich selbst ab.)

import { DestroyRef, inject } from '@angular/core';
import { takeUntilDestroyed } from '@angular/core/rxjs-interop';

export class Notifications {
  private readonly destroyRef = inject(DestroyRef);

  constructor() {
    // ✓ im Injection-Kontext: kein Argument
    this.socket.messages$.pipe(takeUntilDestroyed()).subscribe((m) => this.show(m));
  }

  connectLater() {
    // ✓ außerhalb des Konstruktors: das DestroyRef explizit übergeben
    this.poll$.pipe(takeUntilDestroyed(this.destroyRef)).subscribe();
  }
}

// Früher: Subject + ngOnDestroy
// private destroy$ = new Subject<void>();
// ngOnDestroy() { this.destroy$.next(); this.destroy$.complete(); }
// obs$.pipe(takeUntil(this.destroy$))

LiveMemory Leak vs. korrektes Cleanup

Erzeuge eine Kindkomponente, zerstöre sie dann und beobachte das Log.

Outputs aus und zu Observables

import { outputFromObservable, outputToObservable } from '@angular/core/rxjs-interop';

export class Search {
  private readonly query$ = new Subject<string>();

  // Ein Output, der aus einem Observable gebaut wird
  readonly queryChange = outputFromObservable(this.query$.pipe(debounceTime(300)));
}

// Umgekehrt: Du behandelst den Output einer Komponente als Observable (z. B. in einer Direktive)
const changes$ = outputToObservable(childComponentRef.instance.queryChange);

Architekturempfehlung

// Die Architekturempfehlung:
//   RxJS an den Rändern (I/O, Events),  Signals für den angezeigten Zustand.

@Service()
export class ChatStore {
  private readonly socket = inject(ChatSocket); // stellt Observable<Message> bereit

  // ein Event-Stream → mit RxJS akkumuliert → als Signal bereitgestellt
  readonly messages = toSignal(
    this.socket.messages$.pipe(scan((all, m) => [...all, m].slice(-100), [] as Message[])),
    { initialValue: [] },
  );

  readonly unread = computed(() => this.messages().filter((m) => !m.read).length);
}

Wann ja, wann nein

Bleib bei RxJS, wenn

Du Events über die Zeit hast: Websocket, SSE, Drag & Drop, Infinite Scroll.
Du Zeit-/Nebenläufigkeits-Operatoren brauchst: switchMap, exhaustMap, retry mit Backoff, bufferTime.
Du viele asynchrone Quellen mit komplexen Regeln kombinierst.
Du mit APIs arbeitest, die Observables bereitstellen (Router-Events, Reactive Forms valueChanges).

Wechsle zu Signals, wenn

Du UI-Zustand hältst (Filter, Auswahl, Toggles).
Du Werte für das Template ableitest — computed ist einfacher als combineLatest.
Du GETs ausführst, die vom Zustand abhängen — httpResource.
Du ein BehaviorSubject nur als „beobachtbare Variable“ nutzt — genau das ist ein signal.

Häufige Stolperfallen

toSignal außerhalb des Injection-Kontexts → ein Fehler. Erzeuge es als Feld oder übergib { injector }.
toSignal auf einem „kalten“ HTTP-Observable in einem Getter → bei jedem Aufruf ein neuer Request. Erzeuge es nur einmal.
Fehler: Wenn das Observable einen Fehler wirft, wirft das Lesen des Signals ebenfalls. Fang den Fehler vorher mit catchError ab.
toObservable emittiert nicht synchron: Verlass dich nicht darauf, den Wert direkt nach set() zu erhalten.

Spickzettel

toSignal(obs$, { initialValue })        // Observable → Signal (auto-unsubscribe)
toSignal(obs$, { requireSync: true })   // kein undefined, wenn es synchron emittiert
toObservable(sig)                       // Signal → Observable (emittiert asynchron)
takeUntilDestroyed()                    // im Injection-Kontext
takeUntilDestroyed(destroyRef)          // überall
outputFromObservable(obs$)              // Observable → output()
outputToObservable(output)              // output() → Observable
rxResource({ params, stream })          // asynchrone Daten über ein Observable → Resource
pendingUntilEvent()                     // hält die Anwendung „instabil“ (SSR) bis zur ersten Emission