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

Why Real-Time Angular Needs Typed NgRx in 2025

Architecture: WebSocket Effects, Entity Upserts, ACK Correlation, and Signals

// telemetry.models.ts
export type TelemetryEvent =
  | { type: 'metric'; id: string; ts: number; name: string; value: number; tenantId: string }
  | { type: 'status'; id: string; ts: number; state: 'online' | 'offline' | 'error'; tenantId: string }
  | { type: 'ack'; correlationId: string; ok: boolean; error?: string };

export interface Command {
  kind: 'restart' | 'calibrate' | 'throttle';
  targetId: string;
}

WebSocket Service with Exponential Backoff (Reset on Success)

// telemetry.socket.ts
import { Injectable, inject, PLATFORM_ID } from '@angular/core';
import { isPlatformBrowser } from '@angular/common';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { Observable, EMPTY, timer } from 'rxjs';
import { shareReplay, retry, tap } from 'rxjs/operators';
import { TelemetryEvent } from './telemetry.models';

@Injectable({ providedIn: 'root' })
export class TelemetrySocket {
  private socket$?: WebSocketSubject<TelemetryEvent>;
  private platformId = inject(PLATFORM_ID);

  connect(url: string): Observable<TelemetryEvent> {
    if (!isPlatformBrowser(this.platformId)) return EMPTY;

    this.socket$ = webSocket<TelemetryEvent>({
      url,
      deserializer: e => JSON.parse(e.data),
    });

    return this.socket$.pipe(
      // retry with exponential backoff; resets after any success
      retry({
        delay: (err, retryCount) => timer(Math.min(500 * 2 ** (retryCount - 1), 10_000)),
        resetOnSuccess: true,
      }),
      shareReplay({ bufferSize: 1, refCount: true })
    );
  }

  send(msg: unknown) { this.socket$?.next(msg as any); }
  close() { this.socket$?.complete(); }
}

Typed Actions and Effects for Telemetry (NgRx 17+)

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

export const TelemetryActions = createActionGroup({
  source: 'Telemetry',
  events: {
    'Connect': props<{ url: string }>(),
    'Connected': emptyProps(),
    'Disconnect': emptyProps(),
    'Message Received': props<{ event: TelemetryEvent }>(),
    'Send Command': props<{ command: Command; correlationId: string }>(),
    'Ack Received': props<{ correlationId: string; ok: boolean; error?: string }>(),
    'Connection Error': props<{ error: unknown }>(),
  },
});

// telemetry.effects.ts
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { Injectable, inject } from '@angular/core';
import { map, switchMap, takeUntil, filter, tap, mergeMap, timer, take } from 'rxjs';
import { TelemetryActions } from './telemetry.actions';
import { TelemetrySocket } from './telemetry.socket';

@Injectable()
export class TelemetryEffects {
  private actions$ = inject(Actions);
  private socket = inject(TelemetrySocket);

  connect$ = createEffect(() =>
    this.actions$.pipe(
      ofType(TelemetryActions.connect),
      switchMap(({ url }) =>
        this.socket.connect(url).pipe(
          tap(() => {/* connection is alive */}),
          map(event => TelemetryActions.messageReceived({ event })),
          takeUntil(this.actions$.pipe(ofType(TelemetryActions.disconnect)))
        )
      )
    )
  );

  routeInbound$ = createEffect(() =>
    this.actions$.pipe(
      ofType(TelemetryActions.messageReceived),
      map(({ event }) => {
        if (event.type === 'ack') {
          return TelemetryActions.ackReceived({ correlationId: event.correlationId, ok: event.ok, error: event.error });
        }
        // fall-through: handle metric/status via reducer upserts
        return { type: '[Telemetry] Upsert From Event', event } as const;
      })
    )
  );

  sendCommand$ = createEffect(() =>
    this.actions$.pipe(
      ofType(TelemetryActions.sendCommand),
      tap(({ command, correlationId }) => this.socket.send({ ...command, correlationId })),
      mergeMap(({ correlationId }) =>
        // timeout in 5s if no ACK; rollback path
        timer(5000).pipe(
          takeUntil(
            this.actions$.pipe(
              ofType(TelemetryActions.ackReceived),
              filter(a => a.correlationId === correlationId),
              take(1)
            )
          ),
          map(() => TelemetryActions.ackReceived({ correlationId, ok: false, error: 'timeout' }))
        )
      )
    )
  );
}

Reducers: Entity Upserts and Optimistic Pending Map

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

export interface DeviceMetric { id: string; name: string; value: number; tenantId: string; ts: number; }
export interface PendingCommand { command: Command; ts: number }

export interface TelemetryState extends EntityState<DeviceMetric> {
  pending: Record<string, PendingCommand>;
}

export const adapter = createEntityAdapter<DeviceMetric>({ selectId: m => m.id });
const initialState: TelemetryState = adapter.getInitialState({ pending: {} });

function eventToMetric(e: TelemetryEvent): DeviceMetric | null {
  if (e.type !== 'metric') return null;
  return { id: e.id, name: e.name, value: e.value, tenantId: e.tenantId, ts: e.ts };
}

export const telemetryReducer = createReducer(
  initialState,
  on({ type: '[Telemetry] Upsert From Event' } as any, (state, { event }) => {
    const metric = eventToMetric(event as TelemetryEvent);
    return metric ? adapter.upsertOne(metric, state) : state;
  }),
  on(TelemetryActions.sendCommand, (state, { correlationId, command }) => ({
    ...state,
    pending: { ...state.pending, [correlationId]: { command, ts: Date.now() } },
  })),
  on(TelemetryActions.ackReceived, (state, { correlationId, ok }) => {
    const { [correlationId]: _drop, ...pending } = state.pending;
    // Optionally apply server-confirmed state changes when ok===true
    return { ...state, pending };
  })
);

export const { selectAll: selectAllMetrics } = adapter.getSelectors();

Bridging NgRx to Signals and PrimeNG Virtual Tables

// telemetry.selectors.ts
import { createSelector } from '@ngrx/store';
import { selectAllMetrics } from './telemetry.reducer';

export const selectMetricsByTenant = (tenantId: string) =>
  createSelector(selectAllMetrics, list => list.filter(m => m.tenantId === tenantId));

// telemetry.component.ts
import { Component, computed, inject, signal } from '@angular/core';
import { Store } from '@ngrx/store';
import { selectMetricsByTenant } from './telemetry.selectors';

@Component({ selector: 'telemetry-table', templateUrl: './telemetry.component.html' })
export class TelemetryComponent {
  private store = inject(Store);
  tenantId = signal('acme');

  metrics = this.store.selectSignal(selectMetricsByTenant(this.tenantId()));
  count = computed(() => this.metrics().length);
}

<!-- telemetry.component.html -->
<p-table
  [value]="metrics()"
  [virtualScroll]="true"
  [rows]="50"
  [scrollHeight]="'60vh'"
  [trackBy]="(i, row) => row.id"
  styleClass="p-datatable-sm"
>
  <ng-template pTemplate="header">
    <tr><th>Device</th><th>Metric</th><th>Value</th><th>Time</th></tr>
  </ng-template>
  <ng-template pTemplate="body" let-row>
    <tr>
      <td>{{ row.id }}</td>
      <td>{{ row.name }}</td>
      <td>{{ row.value | number:'1.0-2' }}</td>
      <td>{{ row.ts | date:'shortTime' }}</td>
    </tr>
  </ng-template>
</p-table>

Multi‑Tenant Guards and Offline‑Tolerant UX

Testing and CI Guardrails for Streams

# .github/workflows/ci.yml
name: ci
on: [pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: pnpm/action-setup@v3
        with: { version: 9 }
      - run: pnpm install --frozen-lockfile
      - run: pnpm nx run-many -t lint,test --configuration=ci
      - run: pnpm nx e2e dashboard-e2e --configuration=ci

When to Hire an Angular Developer for Real‑Time Dashboards

Example Instrumentation and Performance Tuning