Lesson 4

RxJS Interop: The Bridge

The Core Idea

You already have services that return observables, stores that expose selectors as observables, and HTTP calls that return observables. You don't need to rewrite any of that. The interop layer lets you consume those observables as signals in your templates, and feed signal-based state back into observable pipelines when needed.

Greenfield or Migration?

A fair question: is this interop layer just a way to avoid throwing away legacy code? No. Even in a brand-new application, you'd use both systems - and you'd still need the bridge between them.

The reason is that signals and observables solve fundamentally different problems:

• Signals own synchronous, UI-bound state. Component state, form state, derived/computed values, store slices - anything the template reads.
• Observables own async coordination. HTTP calls, WebSocket streams, complex event orchestration (debounce, retry, race, combineLatest across time), route param streams - anything where time and event sequencing matter.

In a greenfield app, your HttpClient still returns observables. Your WebSocket service still returns observables. A route guard that chains multiple async checks is still cleaner as an observable pipeline. You use toSignal() at the boundary where async data enters your component/store state, and toObservable() where a signal value needs to feed into a pipeline (e.g., a search term signal driving a debounced HTTP lookup).

What wouldn't happen in a greenfield app is wrapping everything in toSignal() just because you can. You wouldn't create a BehaviorSubject for local component state when a signal() does the job directly. The interop layer isn't legacy glue - it's the designed interface between two complementary systems.

Example: A Greenfield Search Feature

Here's a complete policy search component you might write from scratch today. Each piece uses the right primitive for its job:

import { rxResource } from '@angular/core/rxjs-interop';
import { signal, computed, inject } from '@angular/core';
import { retry } from 'rxjs';

@Component({
  template: `
    <input
      [value]="query()"
      (input)="query.set($event.target.value)"
      placeholder="Search policies..."
    />

    @if (results.isLoading()) {
      <app-spinner />
    }

    @for (policy of results.value(); track policy.id) {
      <app-policy-card
        [policy]="policy"
        [selected]="policy.id === selectedId()"
        (click)="selectedId.set(policy.id)"
      />
    }
  `
})
export class PolicySearchPage {
  private readonly policyService = inject(PolicyService);

  // --- SIGNALS: synchronous UI state ---
  readonly query = signal('');           // user input - local, writable
  readonly selectedId = signal<string | null>(null);  // selection state

  // --- OBSERVABLE PIPELINE: async orchestration ---
  // rxResource bridges an Observable-returning loader into signal-land,
  // giving you .value(), .isLoading(), and .error() as signals.
  // Covered in depth in Lesson 8 - for now, note the shape.
  readonly results = rxResource({
    request: () => this.query(),  // reactive signal dependency
    loader: ({ request: query }) => this.policyService.search(query).pipe(
      retry({ count: 2, delay: 1000 })
    )
  });

  // --- DERIVED SIGNAL: synchronous computation ---
  readonly selectedPolicy = computed(() => {
    const id = this.selectedId();
    return this.results.value()?.find(p => p.id === id) ?? null;
  });
}

rxResource is a higher-level abstraction over the manual toObservable() -> pipe -> toSignal() sandwich - you declare a reactive dependency and an Observable-returning loader, and Angular gives you a complete async state machine as signals. We'll cover it properly in the rxResource section below and in full depth in Lesson 8.

Notice the separation:

• Signals (query, selectedId, selectedPolicy) - synchronous state the template reads and the user writes. No async involved.
• Observable/Resource (rxResource wrapping policyService.search()) - async coordination with retry logic. The service returns an observable; the resource bridges it into signal-land with loading/error state included.

No legacy code in sight. This is what the architecture looks like when both systems are used for what they're good at.

toSignal: Observable to Signal

toSignal() subscribes to an observable and exposes its latest emission as a signal. It's the replacement for the async pipe - but usable anywhere, not just templates:

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

@Component({
  template: `
    @if (policies(); as policies) {
      <app-policy-list [policies]="policies" />
    } @else {
      <app-loading />
    }
  `
})
export class PolicyPage {
  private readonly http = inject(HttpClient);

  // Observable from HTTP - unchanged from your existing code
  private readonly policies$ = this.http.get<Policy[]>('/api/policies');

  // Expose as a signal for the template - no async pipe needed
  readonly policies = toSignal(this.policies$);
}

Key behaviours:

• Subscribes immediately - not lazily like templates with async. The subscription starts the moment toSignal() is called.
• Auto-unsubscribes - when the component/service that created it is destroyed, the subscription is cleaned up. No takeUntilDestroyed() needed.
• Injection context required - must be called in a constructor, field initialiser, or with an explicit injector option (same rule as effect()).

The Initial Value Problem

Observables can be async - they might not emit immediately. But signals always have a current value. This creates a type mismatch that toSignal() resolves in three ways:

// 1. No initial value - signal type includes undefined
const policies = toSignal(this.policies$);
// Type: Signal<Policy[] | undefined>

// 2. Explicit initial value - no undefined in the type
const policies = toSignal(this.policies$, { initialValue: [] });
// Type: Signal<Policy[]>

// 3. requireSync - for observables that emit synchronously (BehaviorSubject)
const count = toSignal(this.count$, { requireSync: true });
// Type: Signal<number> (no undefined, no initial value needed)

requireSync in Depth

The problem requireSync solves is a type-level one. When you call toSignal(obs$), TypeScript has no way to know whether the observable will emit immediately or later. So Angular defaults to the safe assumption: the signal might not have a value yet, giving you Signal<T | undefined>.

But some observables always emit synchronously on subscribe - they have a current value ready before the subscription function returns. The classic examples:

• BehaviorSubject - holds a current value, emits it immediately to every new subscriber
• NgRx selectors - backed by BehaviorSubject internally, store.select(...) always emits the current slice on subscribe
• ReplaySubject(1) - replays the last value synchronously if one exists
• Observables piped through startWith() - the startWith value emits synchronously

For these, requireSync: true tells Angular: "I guarantee this observable will have emitted a value by the time toSignal() returns." Angular trusts you - it gives back Signal<T> (no undefined). But it also verifies at runtime: if the observable hasn't emitted by the time the constructor completes, Angular throws NG0601.

// This works - BehaviorSubject emits 'initial' immediately
const source$ = new BehaviorSubject('initial');
const value = toSignal(source$, { requireSync: true });
// Type: Signal<string>
// value() === 'initial' immediately, no undefined possible

// This CRASHES at runtime - HTTP is async, nothing emits on subscribe
const data = toSignal(this.http.get('/api/data'), { requireSync: true });
// Runtime error: NG0601

Why not just use initialValue instead? You could - toSignal(store.select(selectPolicies), { initialValue: [] }) compiles fine. But it's semantically wrong: the signal will never be that initial value because the store emits immediately. requireSync communicates intent more clearly and avoids inventing a phantom state that never occurs.

Error Handling

If the source observable errors, the error is thrown when the signal is read. This means it'll trigger Angular's error handling (ErrorHandler) or crash the component. Handle errors in the observable pipeline before converting:

// BAD - error crashes the component when signal is read
const policies = toSignal(this.http.get<Policy[]>('/api/policies'));

// GOOD - handle errors in the observable pipeline
const policies = toSignal(
  this.http.get<Policy[]>('/api/policies').pipe(
    catchError(() => of([]))  // fallback to empty array
  ),
  { initialValue: [] }
);

Custom Equality

Like all signal primitives (see Custom Equality Functions in Lesson 1), toSignal() accepts an equal option to prevent unnecessary downstream updates:

// Only update signal (and re-render) when the policy ID changes
const activePolicy = toSignal(this.activePolicy$, {
  initialValue: null,
  equal: (a, b) => a?.id === b?.id,
});

toObservable: Signal to Observable

toObservable() goes the other direction - it creates an observable that emits whenever the signal's value changes. Use it when you need to feed signal state into an existing RxJS pipeline:

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

@Component({
  template: `
    <input [value]="searchQuery()" (input)="searchQuery.set($event.target.value)" />
    <app-results [results]="results()" />
  `
})
export class PolicySearch {
  readonly searchQuery = signal('');

  // Convert signal to observable so we can use RxJS operators
  private readonly results$ = toObservable(this.searchQuery).pipe(
    debounceTime(300),
    switchMap(query => this.policyService.search(query))
  );

  // Convert back to signal for the template
  readonly results = toSignal(this.results$, { initialValue: [] });
}

This is the classic pattern: signal for user input (synchronous state), observable for async orchestration (debounce + switchMap), signal for template consumption. Each tool used where it's strongest.

Timing and Batching

toObservable() uses an effect() internally to track the signal. This means:

• Emissions are asynchronous - even if you call .set() multiple times synchronously, only the final value is emitted.
• No initial synchronous emission on subscribe - unlike BehaviorSubject, the first value may arrive async. The first value is emitted via a ReplaySubject(1) though, so late subscribers do get it.

const count = signal(0);
const count$ = toObservable(count);

count$.subscribe(v => console.log(v));

count.set(1);
count.set(2);
count.set(3);
// Only logs: 3 (the stabilised value after the microtask)

takeUntilDestroyed: The Cleanup Operator

takeUntilDestroyed() isn't a signal primitive - it's a lifecycle utility for observable subscriptions. It lives in the same @angular/core/rxjs-interop package because it solves a problem you'll hit constantly when working with observables alongside signals: cleaning up subscriptions that toSignal() doesn't manage for you (manual .subscribe() calls for side effects, WebSocket listeners, etc.).

It completes an observable when the enclosing context (component, directive, service) is destroyed, replacing the manual ngOnDestroy + Subject pattern:

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

@Component({ /* ... */ })
export class PolicyUpdates {
  private readonly destroyRef = inject(DestroyRef);

  constructor() {
    // Option 1: in constructor (injection context available)
    this.policyService.liveUpdates$.pipe(
      takeUntilDestroyed()
    ).subscribe(update => this.handleUpdate(update));
  }

  // Option 2: outside constructor (pass DestroyRef explicitly)
  startListening(): void {
    this.policyService.liveUpdates$.pipe(
      takeUntilDestroyed(this.destroyRef)
    ).subscribe(update => this.handleUpdate(update));
  }
}

The Old Way vs The New Way

// OLD: manual destroy$ subject
export class OldComponent implements OnDestroy {
  private destroy$ = new Subject<void>();

  ngOnInit() {
    this.data$.pipe(takeUntil(this.destroy$)).subscribe(/*...*/);
    this.other$.pipe(takeUntil(this.destroy$)).subscribe(/*...*/);
  }

  ngOnDestroy() {
    this.destroy$.next();
    this.destroy$.complete();
  }
}

// NEW: takeUntilDestroyed - no boilerplate
export class NewComponent {
  constructor() {
    this.data$.pipe(takeUntilDestroyed()).subscribe(/*...*/);
    this.other$.pipe(takeUntilDestroyed()).subscribe(/*...*/);
  }
  // No ngOnDestroy needed
}

Real-World Patterns

Pattern 1: Existing NgRx Store + Signal Components

You have a classic NgRx store with selectors. You want to use signals in your new components without rewriting the store:

@Component({
  template: `
    <h2>{{ selectedPolicy()?.holder }}</h2>
    <p>Premium: {{ selectedPolicy()?.premium | currency }}</p>
    @if (isLoading()) {
      <app-spinner />
    }
  `
})
export class PolicyDetail {
  private readonly store = inject(Store);

  // NgRx selectors return Observables - convert to signals
  readonly selectedPolicy = toSignal(
    this.store.select(selectSelectedPolicy),
    { requireSync: true }  // NgRx selectors always emit synchronously
  );

  readonly isLoading = toSignal(
    this.store.select(selectPoliciesLoading),
    { requireSync: true }
  );
}

No async pipe, no subscription management, full type safety. The store is unchanged - you just consume it differently at the component level.

Pattern 2: Signal Input + Async Pipeline

A component receives an ID as an input signal. You need to fetch data whenever it changes - that's an async operation that benefits from RxJS operators:

@Component({
  template: `
    @if (policyDetail(); as detail) {
      <app-policy-summary [policy]="detail" />
    } @else {
      <app-skeleton-loader />
    }
  `
})
export class PolicyDetailContainer {
  readonly policyId = input.required<string>();

  private readonly policyDetail$ = toObservable(this.policyId).pipe(
    switchMap(id => this.policyService.getById(id)),
    catchError(() => of(null))
  );

  readonly policyDetail = toSignal(this.policyDetail$);
}

The flow: signal input -> observable (for switchMap/retry/catchError) -> signal (for template). This is the signal-observable-signal sandwich pattern you'll use anywhere async orchestration meets signal-based templates.

Pattern 3: WebSocket Stream to Signal

@Component({
  template: `
    <span class="premium-badge" [class.stale]="!livePremium()">
      {{ livePremium()?.amount | currency }}
    </span>
  `
})
export class LivePremiumBadge {
  readonly policyId = input.required<string>();

  private readonly livePremium$ = toObservable(this.policyId).pipe(
    switchMap(id => this.wsService.subscribeToPremiumUpdates(id)),
    map(msg => ({ amount: msg.premium, timestamp: msg.ts }))
  );

  readonly livePremium = toSignal(this.livePremium$);
}

WebSocket streams are inherently observable territory - continuous async events. toSignal() gives the template a synchronous snapshot of the latest value. When the component is destroyed, toSignal unsubscribes, which triggers switchMap's teardown, which closes the WebSocket subscription. Clean.

When to Bridge vs When to Stay Pure

resource and rxResource: Declarative Async Data

The toObservable() -> pipe -> toSignal() sandwich works, but it only gives you the resolved value. In real applications you also need loading state, error state, cancellation, and reload. Angular's resource() and rxResource() provide all of this as a single declarative primitive.

They're the same concept - the only difference is the loader's return type:

• resource() - loader returns a Promise and receives an AbortSignal for cancellation. Lives in @angular/core.
• rxResource() - loader returns an Observable. No AbortSignal needed - cancellation is handled by unsubscribing (which RxJS does natively). Lives in @angular/core/rxjs-interop. Use this when your services already return observables (which, with HttpClient, they always do).

The Shape

resource (Promise-based)

import { resource } from '@angular/core';

readonly policy = resource({
  request: () => this.policyId(),
  loader: async ({ request: id, abortSignal }) => {
    const res = await fetch(`/api/policies/${id}`, {
      signal: abortSignal
    });
    return res.json();
  },
});

rxResource (Observable-based)

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

readonly policy = rxResource({
  request: () => this.policyId(),
  loader: ({ request: id }) =>
    this.policyService.getById(id),
});

// Usage - policy is a ResourceRef, not a signal itself.
// Access its signal properties:
policy.value()      // Policy | undefined
policy.isLoading()  // boolean
policy.error()      // unknown | undefined
policy.reload()     // force re-fetch

Both take the same two properties:

• request - a reactive computation (like computed). Whenever signals read inside it change, the resource re-fetches. If it returns undefined, the resource becomes idle and the loader doesn't run.
• loader - receives the request value. Returns a Promise (resource) or Observable (rxResource). If the request changes while a previous load is in flight, the previous operation is cancelled.

Complete Options

The full configuration object:

What You Get Back

Both return a ResourceRef - an object with signal properties and methods representing a complete async state machine:

Signals (read the current state):

Methods (control the resource):

This replaces the pattern where you'd previously manage separate data$, loading$, and error$ observables (or an NgRx CallState discriminated union).

Full Example

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

@Component({
  template: `
    @switch (policyResource.status()) {
      @case ('loading') {
        <app-spinner />
      }
      @case ('error') {
        <app-error
          [error]="policyResource.error()"
          (retry)="policyResource.reload()"
        />
      }
      @case ('resolved') {
        <app-policy-detail [policy]="policyResource.value()" />
      }
    }
  `
})
export class PolicyContainer {
  private readonly policyService = inject(PolicyService);
  readonly policyId = input.required<string>();

  readonly policyResource = rxResource({
    request: () => this.policyId(),
    loader: ({ request: id }) => this.policyService.getById(id),
  });
}

resource vs the Manual Sandwich

Manual: toObservable + pipe + toSignal

readonly policyId = input.required<string>();

private readonly policy$ = toObservable(this.policyId).pipe(
  switchMap(id => this.svc.getById(id)),
  catchError(() => of(null))
);

readonly policy = toSignal(this.policy$);
// No loading state, no error state - just the value

rxResource

readonly policyId = input.required<string>();

readonly policy = rxResource({
  request: () => this.policyId(),
  loader: ({ request: id }) => this.svc.getById(id),
});
// .value(), .isLoading(), .error(), .status() included

The manual sandwich gives you raw value only - you'd need additional signals or observables to track loading/error state. rxResource gives you the full state machine for free.

Key Behaviours

• Auto-cancellation: If request changes while a previous load is in flight, the previous operation is cancelled. For resource(), this uses AbortSignal (passed to your loader). For rxResource(), the previous Observable subscription is unsubscribed. Same semantics as switchMap - no stale responses arriving after a newer request.
• Idle when undefined: If the request computation returns undefined, the loader doesn't run and status becomes 'idle'. Useful for conditional fetching:

// Only fetch when we have a valid ID
readonly policy = rxResource({
  request: () => this.policyId() || undefined,  // undefined = don't fetch
  loader: ({ request: id }) => this.svc.getById(id),
});

• Manual reload: Call .reload() to re-run the loader with the same request. Status becomes 'reloading' and the previous value remains accessible while loading.
• Local updates: Call .set(value) or .update(fn) to optimistically update the value without re-fetching. Status becomes 'local'.
• Default value: Pass defaultValue to avoid undefined in the type, similar to initialValue on toSignal():

readonly policies = rxResource({
  request: () => this.filters(),
  loader: ({ request: filters }) => this.svc.search(filters),
  defaultValue: [],  // Type: ResourceRef<Policy[]> (no undefined)
});

resource or rxResource?

Simple rule: if your service returns an Observable (which is almost always the case with HttpClient), use rxResource. If you're working with fetch() or any other Promise-based API, use resource. They're interchangeable otherwise - same state machine, same signals, same behaviours.

Lesson 8 covers httpResource (a higher-level wrapper around HttpClient that removes even more boilerplate), resource chaining, streaming resources, and advanced error recovery patterns.

Quiz

Try It Yourself

// Existing service - do NOT modify
@Injectable({ providedIn: 'root' })
export class ClaimsService {
  readonly claims$ = this.http.get<Claim[]>('/api/claims').pipe(
    shareReplay(1)
  );
  constructor(private http: HttpClient) {}
}

// TODO: Write the component that displays claims using toSignal()

@Component({
  template: `
    @for (claim of claims(); track claim.id) {
      <div class="claim-row">
        {{ claim.id }} - {{ claim.amount | currency }} - {{ claim.status }}
      </div>
    } @empty {
      <p>No claims found.</p>
    }
  `
})
export class ClaimsListComponent {
  private readonly claimsService = inject(ClaimsService);

  readonly claims = toSignal(this.claimsService.claims$, { initialValue: [] });
}

@Component({
  template: `
    <input (input)="query.set($event.target.value)" />
    <ul>
      @for (result of results(); track result.id) {
        <li>{{ result.name }}</li>
      }
    </ul>
  `
})
export class TypeAheadComponent {
  readonly query = signal('');

  // TODO: convert query to observable, debounce 300ms,
  // switchMap to search service, convert back to signal
}

@Component({
  template: `
    <input (input)="query.set($event.target.value)" />
    <ul>
      @for (result of results(); track result.id) {
        <li>{{ result.name }}</li>
      }
    </ul>
  `
})
export class TypeAheadComponent {
  private readonly searchService = inject(SearchService);

  readonly query = signal('');

  private readonly results$ = toObservable(this.query).pipe(
    debounceTime(300),
    filter(q => q.length >= 2),
    switchMap(q => this.searchService.search(q)),
    catchError(() => of([]))
  );

  readonly results = toSignal(this.results$, { initialValue: [] });
}

Your Tangible Win

You now know how to bridge signals and observables in both directions. The key insight: you don't have to choose one world or the other. Use signals for synchronous state and templates, observables for async orchestration and time-based logic, and toSignal()/toObservable() at the boundaries. Your existing services and stores work unchanged - you just consume them differently at the component level.

Next lesson, we'll start building with NgRx SignalStore - purpose-built state management that speaks signals natively, so you won't need these bridges for new state patterns.