import { Controller, Sse, Headers } from '@nestjs/common';
import { interval, Observable } from 'rxjs';
import { map } from 'rxjs/operators';
interface FeedPayload {
id: number;
message: string;
at: string;
}
@Controller('feed')
export class EventsController {
@Sse('stream')
stream(@Headers('last-event-id') lastEventId?: string): Observable<MessageEvent> {
let seq = lastEventId ? Number(lastEventId) : 0;
return interval(1000).pipe(
map(() => {
seq += 1;
const payload: FeedPayload = {
id: seq,
message: `tick #${seq}`,
at: new Date().toISOString(),
};
// `id` lets the browser send Last-Event-ID on reconnect
return {
id: String(seq),
data: payload,
} as MessageEvent;
}),
);
}
}
import { useEffect, useRef, useState } from 'react';
export type Status = 'connecting' | 'open' | 'error';
interface Options {
maxItems?: number;
}
export function useEventSource<T>(url: string, options: Options = {}) {
const { maxItems = 50 } = options;
const [events, setEvents] = useState<T[]>([]);
const [status, setStatus] = useState<Status>('connecting');
const sourceRef = useRef<EventSource | null>(null);
useEffect(() => {
const source = new EventSource(url);
sourceRef.current = source;
source.onopen = () => setStatus('open');
source.onerror = () => setStatus('error');
source.onmessage = (evt: MessageEvent) => {
try {
const parsed = JSON.parse(evt.data) as T;
setEvents((prev) => {
const next = [...prev, parsed];
return next.length > maxItems ? next.slice(-maxItems) : next;
});
} catch {
// ignore malformed frames rather than crash the stream
}
};
return () => {
source.close();
sourceRef.current = null;
};
}, [url, maxItems]);
return { events, status };
}
import React from 'react';
import { useEventSource } from './useEventSource';
interface FeedItem {
id: number;
message: string;
at: string;
}
export function LiveFeed() {
const { events, status } = useEventSource<FeedItem>('/feed/stream', {
maxItems: 100,
});
return (
<section className="live-feed">
<header>
<h2>Live activity</h2>
<span className={`dot dot--${status}`} aria-label={status} />
</header>
{events.length === 0 ? (
<p className="empty">Waiting for events\u2026</p>
) : (
<ul>
{events.map((item) => (
<li key={item.id}>
<time dateTime={item.at}>
{new Date(item.at).toLocaleTimeString()}
</time>
<span>{item.message}</span>
</li>
))}
</ul>
)}
</section>
);
}
This snippet shows the full path of a Server-Sent Events feature: a NestJS controller that streams events over a long-lived HTTP connection, a reusable React hook that consumes them via the browser EventSource API, and a small component that renders a live, capped list. SSE is a good fit here because the flow is one-directional server-to-client, works over plain HTTP/1.1, and gives automatic reconnection for free — unlike WebSockets, which add a second protocol and bidirectional complexity that a live feed does not need.
In events.controller.ts, the stream handler uses RxJS interval mapped into MessageEvent objects, which NestJS serializes into the data: frames the browser expects when a route is annotated with @Sse. Each payload carries an incrementing id and an ISO timestamp so the client can dedupe and order. The Last-Event-ID header is read on connect so a reconnecting client can resume from where it left off rather than replaying the whole stream — this is the backbone of SSE's built-in resilience.
In useEventSource.ts, the hook wraps the imperative EventSource in a declarative interface. It parses each message event's JSON, appends to state while capping the buffer with maxItems to avoid unbounded memory growth, and tracks a status so the UI can reflect connecting, open, or error. Crucially it stores the raw EventSource in a ref and closes it in the useEffect cleanup, preventing duplicate connections across re-renders or Strict Mode double-invocation. The browser handles retry timing itself, but the onerror handler surfaces the transient error state without tearing down the socket.
In LiveFeed.tsx, the component simply calls the hook and maps over events. Because the hook owns the buffer cap and connection lifecycle, the component stays purely presentational. A common pitfall the hook guards against is memory leaks from never-closed connections; another is stale closures over setEvents, avoided by using the functional updater form. Reach for this pattern for dashboards, notification feeds, log tails, and progress streams where updates flow one way and eventual reconnection matters more than sub-millisecond latency.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
import axios, { AxiosError } from 'axios'
import { v4 as uuidv4 } from 'uuid'
const api = axios.create({
baseURL: import.meta.env.VITE_API_URL || 'http://localhost:3000/api/v1',
timeout: 15000,Axios API client with interceptors
import { create } from 'zustand'
import { persist } from 'zustand/middleware'
interface FilterState {
search: string
category: string | nullZustand for lightweight state management
class Comment < ApplicationRecord
belongs_to :article
belongs_to :author, class_name: "User"
validates :body, presence: true, length: { maximum: 2_000 }Live comments with model broadcasts + turbo_stream_from
import React from "react";
type FallbackProps = {
error: Error;
reset: () => void;
};React Error Boundary + error reporting hook
Share this code
Here's the card — post it anywhere.
