typescript 37 lines · 8 steps

Streaming job progress with SSE in NestJS

A NestJS controller turns a job's progress stream into Server-Sent Events, with a heartbeat to keep the connection alive.

Explained by highlit
NestJS Intermediate server-sent-events reactive-streams rxjs real-time observables
1import { Controller, Param, Sse, MessageEvent } from '@nestjs/common';
2import { Observable, interval, merge } from 'rxjs';
3import { filter, map, takeWhile } from 'rxjs/operators';
4import { JobService } from './job.service';
5 
6@Controller('jobs')
7export class JobProgressController {
8 constructor(private readonly jobs: JobService) {}
9 
10 @Sse(':id/progress')
11 streamProgress(@Param('id') id: string): Observable<MessageEvent> {
12 const updates$ = this.jobs.progressStream(id).pipe(
13 filter((update) => update.jobId === id),
14 map((update) => ({
15 id: `${update.jobId}:${update.step}`,
16 type: update.status === 'failed' ? 'error' : 'progress',
17 data: {
18 step: update.step,
19 total: update.total,
20 percent: Math.round((update.step / update.total) * 100),
21 status: update.status,
22 message: update.message,
23 },
24 })),
25 takeWhile(
26 (event) => event.data.status !== 'completed' && event.type !== 'error',
27 true,
28 ),
29 );
30 
31 const heartbeat$ = interval(15000).pipe(
32 map<number, MessageEvent>(() => ({ type: 'ping', data: { ts: Date.now() } })),
33 );
34 
35 return merge(updates$, heartbeat$);
36 }
37}
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1@Sse lets a controller return an Observable that NestJS pipes to the client as a live Server-Sent Event stream.
  2. 2takeWhile with its inclusive flag emits the terminal event before closing the stream, so clients see the final state.
  3. 3Merging a periodic heartbeat into a data stream keeps idle SSE connections from timing out.

Related explainers

typescript 
import { registerLocaleData } from '@angular/common';
import localeFr from '@angular/common/locales/fr';
import localeFrExtra from '@angular/common/locales/extra/fr';
import localeDe from '@angular/common/locales/de';

Locale-aware bootstrapping in Angular

i18n localization dependency-injection
Intermediate 8 steps
python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
typescript 
import { Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import * as Joi from 'joi';
 

Validating env config at boot in NestJS

configuration schema-validation environment-variables
Intermediate 8 steps
typescript 
import { Inject, Injectable, Logger } from '@nestjs/common';
import { CACHE_MANAGER } from '@nestjs/cache-manager';
import { Cache } from 'cache-manager';
import { InjectRepository } from '@nestjs/typeorm';

A cache-aside country lookup in NestJS

cache-aside dependency-injection batch-lookup
Intermediate 8 steps
rust 
use axum::{
    extract::{Path, State},
    response::sse::{Event, KeepAlive, Sse},
};

Streaming import progress with SSE in Axum

server-sent-events streams watch-channel
Advanced 7 steps
typescript 
import { Injectable, effect, signal, computed } from '@angular/core';
 
interface Preferences {
  theme: 'light' | 'dark';

A signal-based preferences store in Angular

signals state-management persistence
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

Streaming job progress with SSE in NestJS — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code