Blend RxJS Streams with Angular 20+ Signals Using Typed Adapters — Keep SSR and Tests Deterministic

Real-world pain I’ve seen

On an airline kiosk rollout, WebSocket device state leaked into SSR and blocked the page. On a telecom ads dashboard, a hot stream caused hydration mismatch and a 2-3 frame jitter. On an insurance telematics app, tests failed intermittently due to non-deterministic timers. All three were solved with a single pattern: typed adapters converting RxJS to Signals with a deterministic initial snapshot.

• SSR renders a different first frame than the client, causing hydration warnings and flicker.

• Hot WebSocket streams connect during SSR and stall the render.

• Marble tests are flaky because timers and Date.now leak into operators.

Root causes

SSR must finish quickly without waiting for live streams. If your Observable has no synchronous initial value, the server emits nothing, the client immediately emits something, and hydration disagrees. Similarly, unguarded browser APIs crash or hang SSR. The adapter solves this by enforcing a typed initial snapshot and replaying the latest value deterministically.

• Server vs. browser side-effects (WebSocket, Firestore, setInterval).

• Hot streams without a cached value (no shareReplay).

• Different initial values between SSR and client.

• Timing dependencies (Date.now) and async schedulers.

Goals

• Same initial value on server and client.

• Live updates only in the browser.

• Replayed, distinct values to minimize renders.

• Type-safe inputs/outputs with small API.

Adapter implementation

This factory wraps toSignal with SSR guards, startWith, distinctUntilChanged, and shareReplay(1). It’s small enough to copy into your libs/state package in an Nx monorepo.

Code

import { Injectable, Inject, Injector, Signal, PLATFORM_ID } from '@angular/core';
import { isPlatformServer } from '@angular/common';
import { Observable, of } from 'rxjs';
import { distinctUntilChanged, shareReplay, startWith } from 'rxjs/operators';
import { toSignal } from '@angular/core/rxjs-interop';

export interface RxSignal<T> {
  value: Signal<T>;
  as$(): Observable<T>;
}

function shallowEqual(a: any, b: any): boolean {
  if (a === b) return true;
  if (!a || !b || typeof a !== 'object' || typeof b !== 'object') return false;
  const ak = Object.keys(a), bk = Object.keys(b);
  if (ak.length !== bk.length) return false;
  for (const k of ak) if (a[k] !== b[k]) return false;
  return true;
}

@Injectable({ providedIn: 'root' })
export class RxSignalAdapterFactory {
  constructor(private injector: Injector, @Inject(PLATFORM_ID) private platformId: Object) {}

  create<T>(source$: Observable<T>, opts: { initial: T; compare?: (a: T, b: T) => boolean }): RxSignal<T> {
    const isServer = isPlatformServer(this.platformId);
    const compare = opts.compare ?? shallowEqual;
    // Only initial snapshot on the server; live stream in browser.
    const safe$ = (isServer ? of(opts.initial) : source$).pipe(
      startWith(opts.initial),
      distinctUntilChanged(compare),
      shareReplay({ bufferSize: 1, refCount: true })
    );

    const value = toSignal(safe$, { initialValue: opts.initial, injector: this.injector });
    return { value, as$: () => safe$ };
  }
}

Why types matter for determinism

For real-time pipelines I ship (telecom ads analytics, telematics dashboards), we enforce typed event contracts at the gateway. That lets the adapter compare values deterministically and keeps PrimeNG tables from re-rendering rows unnecessarily.

• Stable shapes enable deep/shallow equality and reduce spurious renders.

• Typed events ease logging, replay, and test fixtures.

Example event type

export type TelemetryType = 'temp' | 'speed';
export interface TelemetryEvent {
  ts: number;       // epoch ms
  deviceId: string;
  type: TelemetryType;
  value: number;
}

Pattern

import { Injectable, Inject, PLATFORM_ID } from '@angular/core';
import { isPlatformServer } from '@angular/common';
import { Observable, EMPTY } from 'rxjs';
import { webSocket } from 'rxjs/webSocket';
import { TelemetryEvent } from './types';

@Injectable({ providedIn: 'root' })
export class TelemetryStream {
  constructor(@Inject(PLATFORM_ID) private platformId: Object) {}

  connect(): Observable<TelemetryEvent> {
    if (isPlatformServer(this.platformId)) return EMPTY; // SSR safe
    return webSocket<TelemetryEvent>('wss://example.com/telemetry');
  }
}

• Don’t create WebSockets, Firebase listeners, or timers on the server.

• Expose EMPTY or initial snapshots on the server; real connections in the browser.

Service using the adapter

@Injectable({ providedIn: 'root' })
export class TelemetryService {
  readonly telemetry: Signal<TelemetryEvent[]>;

  constructor(private stream: TelemetryStream, private adapter: RxSignalAdapterFactory) {
    const list$ = this.stream.connect().pipe(
      // group into a small collection for the table
      scan((acc, e) => [e, ...acc].slice(0, 25), [] as TelemetryEvent[])
    );

    this.telemetry = this.adapter.create(list$, { initial: [], compare: shallowArray }).value;
  }
}

function shallowArray(a: any[], b: any[]) {
  return a.length === b.length && a.every((v, i) => v === b[i]);
}

Component and template

@Component({
  selector: 'telemetry-table',
  template: `
    <p-table [value]="rows()" [rows]="25" [paginator]="false">
      <ng-template pTemplate="header">
        <tr><th>Device</th><th>Type</th><th>Value</th><th>Time</th></tr>
      </ng-template>
      <ng-template pTemplate="body" let-row>
        <tr>
          <td>{{ row.deviceId }}</td>
          <td>{{ row.type }}</td>
          <td>{{ row.value }}</td>
          <td>{{ row.ts | date:'mediumTime' }}</td>
        </tr>
      </ng-template>
    </p-table>
  `
})
export class TelemetryTableComponent {
  rows = inject(TelemetryService).telemetry; // Signal<TelemetryEvent[]>
}

What you get

• SSR renders an empty table ([]) without hanging.

• Hydration picks up latest items immediately without mismatch.

• Angular DevTools shows minimal re-renders due to distinctUntilChanged.

SignalStore patching from adapters

import { SignalStore, withState, patchState } from '@ngrx/signals';

interface TelemetryState { rows: TelemetryEvent[]; connected: boolean; }

@Injectable({ providedIn: 'root' })
export class TelemetryStore extends SignalStore(withState<TelemetryState>({ rows: [], connected: false })) {
  constructor(stream: TelemetryStream, adapter: RxSignalAdapterFactory) {
    super();
    const connected$ = stream.connect().pipe(map(() => true), startWith(false));

    const connected = adapter.create(connected$, { initial: false }).value;
    const rows = adapter.create(stream.connect().pipe(
      scan((acc, e) => [e, ...acc].slice(0, 25), [] as TelemetryEvent[])
    ), { initial: [] }).value;

    effect(() => {
      patchState(this, { connected: connected(), rows: rows() });
    });
  }
}

Firebase note

I’ve shipped SSR on Firebase Hosting with Angular Universal. The same adapter works for Firestore collection snapshots: startWith initial, and connect the listener only in the browser. See my SSR guardrails article for hydration metrics and CI budgets.

• Use serverTimestamp and startWith a placeholder snapshot for SSR.

• Guard Firestore listeners on the server; render from cached SSR data or initial state.

Marble test

import { TestBed } from '@angular/core/testing';
import { TestScheduler } from 'rxjs/testing';

it('replays deterministically with initial snapshot', () => {
  const scheduler = new TestScheduler((a, b) => expect(a).toEqual(b));
  scheduler.run(({ cold, expectObservable }) => {
    const src$ = cold(' -a-b-|', { a: 1, b: 2 });
    const factory = TestBed.inject(RxSignalAdapterFactory);
    const { as$ } = factory.create(src$, { initial: 0 });
    expectObservable(as$()).toBe('(0a)b-|', { 0: 0, a: 1, b: 2 });
  });
});

SSR test

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

it('emits only initial on server', () => {
  TestBed.overrideProvider(PLATFORM_ID, { useValue: 'server' });
  const factory = TestBed.inject(RxSignalAdapterFactory);
  const src$ = new Subject<number>();
  const { value } = factory.create(src$.asObservable(), { initial: 42 });
  expect(value()).toBe(42); // no subscription occurred
});

My playbook on enterprise teams

This pattern has stabilized a global entertainment employee-tracking app, a telecom ads analytics dashboard, and an airport kiosk fleet—all Angular 20+ with Signals/SignalStore. If you want help implementing it, hire an Angular developer with Fortune 100 experience—this is what I do.

• Audit streams and define typed contracts at the edge.

• Wrap all live streams with the adapter; forbid direct subscriptions in components.

• Instrument render counts with Angular DevTools and log adapter emissions.

• Add SSR smoke tests and hydration diff checks in CI (Nx target + GitHub Action).

• Roll out feature by feature behind flags to avoid a freeze.

Signals that you need help now

I specialize in code rescues and upgrades without a delivery freeze. If your streams and Signals are fighting, let’s stabilize it. See how I "rescue chaotic code" and "stabilize your Angular codebase" via gitPlumbers—then we can discuss your Angular project.

• SSR hydration warnings in prod or Lighthouse drops after a Signals migration.

• Flaky tests around real-time features and timers.

• Hot streams causing memory leaks or excess re-renders.

• RxJS 7→8 upgrade created subscription churn or type regressions.

Quick wins to implement this week

Next, wire adapter emissions into SignalStore and use PrimeNG or Angular Material to render from Signals only. Watch Core Web Vitals improve by eliminating hydration mismatch and unnecessary renders.

• Introduce RxSignalAdapterFactory and migrate one hot stream.

• Add SSR guards to WebSocket/Firebase sources; startWith initial snapshots.

• Measure re-render counts before/after with Angular DevTools.

• Write a single marble test for your most critical stream.