Real‑Time NgRx for Angular 20+: WebSocket Streams, Optimistic Updates, and Typed Actions for Telemetry Dashboards

A familiar production scene

At a telecom, ad events spiked during primetime and our untyped stream flooded change detection. We stabilized it with typed actions/effects, entity adapters, and signal-driven selectors. Optimistic UI kept operators fast; correlationIds made it safe.

• Firehose stream spikes to 50k events/min

• Toggle buttons show success before ACK

• Filters miss frames and jitter

Why NgRx + Signals for real-time

NgRx gives us deterministic state and time-travelable reducers; Signals give components stable, push-driven inputs. Together, they tame WebSocket noise and make dashboards measurable with Angular DevTools.

• Effects isolate IO and concurrency

• Reducers stay pure and replayable

• selectSignal minimizes renders

Typed protocols prevent chaos

If the wire format isn’t modeled, your store becomes a dumping ground. I use TypeScript discriminated unions and NgRx createActionGroup so every event is explicit and testable.

• ActionGroup per channel

• No any in reducers

• Runtime guards optional

Optimistic UI that doesn’t lie

Operators shouldn’t wait for round‑trip latency to flip a switch. We apply the change locally, tag it with a correlationId, and reconcile on server response—logging time‑to‑ack in GA4 or Firebase for SLOs.

• CorrelationId per command

• ACK commits, NACK rollbacks

• Telemetry on latency and errors

Signals keep renders low

Signals convert selectors into stable, synchronous sources. On a busy telematics board we cut re-renders ~60% by switching high-churn selectors to selectSignal with view-level computed derivatives.

• Store.selectSignal for views

• computed() for projections

• auditTime for heavy lists

1) Define a typed event model

Treat the socket as a protocol. Group actions by direction and event type.

Code: action groups and schema

// telemetry.actions.ts
import { createActionGroup, props, emptyProps } from '@ngrx/store';

export type TelemetryEvent =
  | { type: 'device.update'; id: string; status: 'online'|'offline'; ts: number }
  | { type: 'metric.append'; id: string; metric: string; value: number; ts: number }
  | { type: 'ack'; correlationId: string; ts: number }
  | { type: 'nack'; correlationId: string; error: string; ts: number };

export const TelemetryActions = createActionGroup({
  source: 'Telemetry',
  events: {
    'Connect': emptyProps(),
    'Connected': emptyProps(),
    'Disconnected': props<{ reason?: string }>(),

    'Server Event': props<{ event: TelemetryEvent }>(),

    'Cmd Toggle Device': props<{ id: string; next: 'online'|'offline'; correlationId: string }>(),
    'Cmd Sent': props<{ correlationId: string }>(),

    'Apply Device Update': props<{ id: string; status: 'online'|'offline'; optimistic?: boolean }>(),
    'Append Metric': props<{ id: string; metric: string; value: number; ts: number }>(),

    'Ack Received': props<{ correlationId: string }>(),
    'Nack Received': props<{ correlationId: string; error: string }>(),
  }
});

2) A resilient WebSocket service

// telemetry.socket.ts
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { retryBackoff } from 'backoff-rxjs';
import { filter, map, shareReplay, tap } from 'rxjs/operators';
import { Observable, timer } from 'rxjs';

export interface Wire<T> { type: string; payload: T; }

export class TelemetrySocket {
  private socket$!: WebSocketSubject<unknown>;

  connect(url: string) {
    this.socket$ = webSocket({ url, deserializer: e => JSON.parse(e.data) });
    const heartbeats$ = timer(0, 15000).pipe(tap(() => this.socket$.next({ type: 'ping' })));

    const stream$ = this.socket$.pipe(
      retryBackoff({ initialInterval: 1000, maxInterval: 15000, jitter: 0.3 }),
      shareReplay({ bufferSize: 1, refCount: true })
    );

    return { stream$, heartbeats$ } as const;
  }

  send<T>(message: Wire<T>) { this.socket$.next(message); }
}

• Backoff with jitter

• Heartbeat/keepalive

• ShareReplay for late subscribers

3) Effects: connect, map, and commit

// telemetry.effects.ts
import { inject, Injectable } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { Store } from '@ngrx/store';
import { TelemetryActions } from './telemetry.actions';
import { TelemetrySocket } from './telemetry.socket';
import { map, tap, filter, merge, switchMap } from 'rxjs/operators';
import { of, merge as rxMerge } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class TelemetryEffects {
  private actions$ = inject(Actions);
  private socket = inject(TelemetrySocket);
  private store = inject(Store);

  connect$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.connect),
    switchMap(() => {
      const { stream$, heartbeats$ } = this.socket.connect('/ws/telemetry');
      return rxMerge(
        heartbeats$.pipe(map(() => ({ type: '[noop] hb' } as any))),
        stream$.pipe(map((wire: any) => TelemetryActions.serverEvent({ event: wire as any })))
      );
    })
  ));

  routeServerEvents$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.serverEvent),
    map(({ event }) => {
      switch (event.type) {
        case 'device.update':
          return TelemetryActions.applyDeviceUpdate({ id: event.id, status: event.status });
        case 'metric.append':
          return TelemetryActions.appendMetric({ id: event.id, metric: event.metric, value: event.value, ts: event.ts });
        case 'ack':
          return TelemetryActions.ackReceived({ correlationId: event.correlationId });
        case 'nack':
          return TelemetryActions.nackReceived({ correlationId: event.correlationId, error: event.error });
      }
    })
  ));

  sendToggle$ = createEffect(() => this.actions$.pipe(
    ofType(TelemetryActions.cmdToggleDevice),
    tap(({ id, next, correlationId }) => this.socket.send({ type: 'cmd.toggle', payload: { id, next, correlationId } })),
    map(({ correlationId, id, next }) => TelemetryActions.applyDeviceUpdate({ id, status: next, optimistic: true }))
  ));
}

• Edge-IO in effects

• Map server events to actions

• Emit ACK/NACK outcomes

4) Reducer and entity adapter

// telemetry.reducer.ts
import { createEntityAdapter, EntityState } from '@ngrx/entity';
import { createReducer, on } from '@ngrx/store';
import { TelemetryActions } from './telemetry.actions';

type Device = { id: string; status: 'online'|'offline'; pending?: string | null };

const adapter = createEntityAdapter<Device>();
export interface State extends EntityState<Device> { pending: Record<string, { id: string; prev: 'online'|'offline' }>; }
const initial: State = { ...adapter.getInitialState(), pending: {} };

export const reducer = createReducer(initial,
  on(TelemetryActions.applyDeviceUpdate, (state, { id, status, optimistic }) => {
    const prev = state.entities[id]?.status ?? 'offline';
    const change = adapter.upsertOne({ id, status }, state);
    return optimistic ? { ...change, pending: { ...change.pending, [id]: { id, prev } } } : change;
  }),
  on(TelemetryActions.nackReceived, (state, { correlationId }) => {
    // correlate to id as needed (store map correlationId → id)
    // simplified example: rollback all pending
    let next = state;
    for (const pid of Object.keys(state.pending)) {
      const prev = state.pending[pid].prev;
      next = adapter.updateOne({ id: pid, changes: { status: prev } }, next);
    }
    return { ...next, pending: {} };
  }),
  on(TelemetryActions.ackReceived, (state) => ({ ...state, pending: {} }))
);

export const { selectAll, selectEntities } = adapter.getSelectors();

5) Selectors with selectSignal

// telemetry.selectors.ts
import { createSelector, createFeatureSelector } from '@ngrx/store';
import { computed, inject, signal } from '@angular/core';
import { Store } from '@ngrx/store';
import { State } from './telemetry.reducer';

export const selectFeature = createFeatureSelector<State>('telemetry');
export const selectDevices = createSelector(selectFeature, s => Object.values(s.entities));

export function useTelemetrySignals() {
  const store = inject(Store);
  const devicesSig = store.selectSignal(selectDevices);
  const onlineCount = computed(() => devicesSig().filter(d => d?.status === 'online').length);
  return { devicesSig, onlineCount } as const;
}

6) Component wiring (Signals in the view)

// devices.component.ts
import { Component, inject } from '@angular/core';
import { Store } from '@ngrx/store';
import { TelemetryActions } from '../state/telemetry.actions';
import { useTelemetrySignals } from '../state/telemetry.selectors';

@Component({ selector: 'ux-devices', templateUrl: './devices.component.html', standalone: true })
export class DevicesComponent {
  private store = inject(Store);
  sig = useTelemetrySignals();
  connect() { this.store.dispatch(TelemetryActions.connect()); }
  toggle(id: string, next: 'online'|'offline') {
    const correlationId = crypto.randomUUID();
    this.store.dispatch(TelemetryActions.cmdToggleDevice({ id, next, correlationId }));
  }
}

<!-- devices.component.html -->
<button pButton (click)="connect()" label="Connect" icon="pi pi-wifi"></button>
<p>Online: {{ sig.onlineCount() }}</p>
<ul>
  <li *ngFor="let d of sig.devicesSig()">
    <span [class.online]="d?.status==='online'">{{ d?.id }}</span>
    <button (click)="toggle(d!.id, d!.status==='online' ? 'offline' : 'online')">Toggle</button>
  </li>
</ul>

7) Backpressure and render control

For firehose metrics, buffer events in effects and reduce in batches. In views, virtualize with cdk-virtual-scroll or PrimeNG TurboTable. I’ve hit 60fps on 20k-row tables by combining Signals with virtualization and batched entity updates.

• Use auditTime(16) or rAF for 60fps

• Batch metric.append in reducers

• Virtualize long lists

What happens on click

This pattern kept airline kiosks responsive even when the network was spotty. We tracked time-to-ack per device and alerted if latency exceeded SLOs.

• Dispatch cmdToggleDevice

• Effect sends WS and applies optimistic state

• ACK commits, NACK rolls back

Telemetry and metrics

At AngularUX I pipe ack latency into GA4 (BigQuery export) to analyze outliers by tenant and region. On one insurance telematics board, this surfaced a 2.3% NACK spike tied to a bad firmware rollout.

• Angular DevTools render counts

• GA4 custom metrics for ack latency

• Firebase Logs for NACK reasons

CI guardrails (Nx + Cypress)

# .github/workflows/ci.yml
name: ci
on: [push]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-node@v4
        with: { node-version: 20 }
      - run: npm ci
      - run: npx nx run-many -t test --parallel=3
      - run: npx nx e2e dashboard-e2e --configuration=ci
  
# In Cypress, mock WS ACK/NACKs and spike traffic to catch jitter

Playbook I use on rescues

When I’m brought in to rescue a chaotic codebase, I map the event path and measure. We stabilize reducers, add typed actions, convert key selectors to selectSignal, and land feature flags to swap protocols without a deploy freeze.

• Trace WebSocket → effect → reducer → selector → Signals

• Instrument before/after renders and INP

• Add feature flags for protocol changes

Tech I reach for

I’ve delivered this stack across Fortune 100 teams—multi‑tenant role‑based portals, real‑time analytics, and offline‑tolerant kiosk flows with Docker hardware simulation. If you need to hire an Angular developer who has done this under pressure, let’s talk.

• Angular 20, NgRx, PrimeNG/Material

• D3/Highcharts for viz

• Firebase or Node/.NET backends

Practical guardrails

Measure with Angular DevTools, GA4/BigQuery, and CI. When we implemented this at a telecom, render counts dropped ~60% and operator INP stayed <100ms during peak traffic.

• Typed protocol + action groups

• Effects isolate IO; reducers stay pure

• selectSignal everywhere in views

• Optimistic commands with correlationId

• Batch and virtualize high-churn data

What to instrument next

Hook these into Firebase Analytics or your telemetry lake. Use feature flags to iterate safely.

• ACK latency percentiles per tenant

• Dropped frame rate on heavy lists

• Retry/jitter impact on connect times

Good moments to bring me in

I can audit your pipeline in a week and ship a hardened pattern in 2–4 weeks. If your team needs an Angular consultant with Fortune 100 real‑time experience, I’m available for remote engagements.

• You’re adding WebSocket streams and fear jitter

• You need NgRx/Signals expertise to stabilize prod

• You’re upgrading to Angular 20+ with live data