SSE আর streaming — polling, long polling আর AI-র উত্তর stream করা

১ · খবর পৌঁছানোর চারটি পথ — বাজারঘরের অর্ডার পাতা

বাজারঘরে একজন ক্রেতা অর্ডার ৪২ দিয়েছেন। তিনি “আমার অর্ডার” পাতাটি খুলে রেখেছেন। অবস্থা বদলায় এভাবে: placed, confirmed, packed, shipped, delivered. প্রশ্ন হলো, গুদামে প্যাক শেষ হলে খবরটি তাঁর পাতায় যাবে কীভাবে? HTTP-তে নিয়ম হলো, browser জিজ্ঞেস করে আর server উত্তর দেয়। Server নিজে থেকে browser-কে ডাকতে পারে না। এই সীমার চারপাশে চারটি কৌশল তৈরি হয়েছে। এদের তত্ত্ব আর বড় পরিসরের নকশা System Design-এর real-time communication পর্বে আছে। এখানে আমরা application developer হিসেবে Node.js, Express 5 আর Next.js-এ এগুলো বানাব আর মাপব।

চারটিকে একটি ছবিতে দেখুন। সময় বাঁ থেকে ডানে ৪০ সেকেন্ড। লাল দাগে অর্ডারের অবস্থা বদলায়: ১৩ সেকেন্ডে confirmed, ৩১ সেকেন্ডে packed.

ছবিটি ধারণা। এবার আসল মাপ। একটি সত্যিকারের ল্যাবে Express 5.3.0 server (Node 24.15.0) একই অর্ডার ৪২-এর জন্য চারটি endpoint দিয়েছে। চারটি client একসাথে ২০ সেকেন্ড পাতাটি দেখেছে। অবস্থা বদলেছে ৩.০, ৯.৫ আর ১৬.২ সেকেন্ডে। Short polling প্রতি ২ সেকেন্ডে জিজ্ঞেস করেছে, long polling-এর timeout ৮ সেকেন্ড। WebSocket-এর জন্য ws package ব্যবহার হয়েছে; এর বিস্তারিত পর্ব ৬০-এ (WebSocket — Socket.IO, room আর real-time chat). সবকিছু চলেছে শুধু 127.0.0.1-এ।

$ node measure.js        # সত্যিকারের run, একবারের local মাপ
window 20 s, changes at 3.0 s, 9.5 s, 16.2 s
poll  requests/connections = 10  lag ms = 1020, 553, 1886
long  requests/connections =  4  lag ms = 2, 1, 1
sse   requests/connections =  1  lag ms = 3, 2, 2
ws    requests/connections =  1  lag ms = 3, 2, 2

এটি একটি মেশিনে একবারের মাপ, সাধারণ benchmark নয়। Local-এ network-এর দেরি প্রায় শূন্য, তাই long polling, SSE আর WebSocket সমান দেখায়। আসল internet-এ প্রতিটিতে যোগ হবে এক দিকের যাত্রার সময়, আর long polling-এ প্রতিটি উত্তরের পরে নতুন অনুরোধ পাঠানোর সময়। কিন্তু মূল শিক্ষা বদলায় না: polling-এর দেরি নির্ভর করে বিরতির ওপর, আর তার বেশির ভাগ অনুরোধ খালি হাতে ফেরে।

২ · Short polling — TanStack Query-র refetchInterval আর তার খরচ

Short polling সবচেয়ে সরল। আলাদা server লাগে না, proxy-র ঝামেলা নেই, সাধারণ GET আর cache-এর নিয়ম খাটে। পর্ব ১৯-এ আপনি TanStack Query দিয়ে server state এনেছেন। সেখানে polling মানে একটি option:

import { useQuery } from '@tanstack/react-query';

function OrderStatus({ id }: { id: number }) {
  const { data } = useQuery({
    queryKey: ['order', id],
    queryFn: () => fetch(`/api/orders/${id}`).then((r) => r.json()),
    // delivered হলে আর জিজ্ঞেস নয়; না হলে প্রতি ৫ সেকেন্ডে
    refetchInterval: (query) =>
      query.state.data?.status === 'delivered' ? false : 5000,
    // refetchIntervalInBackground: false  ← ডিফল্ট: লুকানো tab-এ থামে
  });
  return <p>অবস্থা: {data?.status ?? '…'}</p>;
}

এখানে দুটি জিনিস লক্ষ করুন। TanStack Query v5-এর docs অনুযায়ী refetchInterval একটি সংখ্যা, false, অথবা একটি ফাংশন নিতে পারে, যা query দেখে বিরতি ঠিক করে। তাই অর্ডার পৌঁছে গেলে polling নিজেই থামে। আর refetchIntervalInBackground-এর ডিফল্ট false: tab লুকানো থাকলে polling বন্ধ। এই দুটি ছোট সিদ্ধান্ত server-এর বোঝা অনেক কমায়। (উদাহরণটি এই পর্বের ল্যাবে চালানো হয়নি; option-গুলো লেখার সময়ের TanStack Query 5.105-এর docs থেকে মেলানো।)

খরচের হিসাব

Polling-এর খরচ সহজ সূত্রে বাঁধা যায়। ধরুন Nটি খোলা পাতা, প্রতিটি T সেকেন্ড পরপর জিজ্ঞেস করে।

• Server-এ অনুরোধের হার: N / T অনুরোধ প্রতি সেকেন্ডে।
• গড় দেরি: বদল যেকোনো মুহূর্তে হতে পারে, তাই গড়ে T / 2, সবচেয়ে খারাপ ক্ষেত্রে প্রায় T (সাথে network-এর সময়)।
• খালি অনুরোধের ভাগ: এক ঘণ্টায় kটি বদল হলে 3600 / Tটি অনুরোধের মধ্যে বড়জোর kটি কাজের।

তবু polling খারাপ নয়। যেখানে কয়েক সেকেন্ডের দেরি চলে (পেমেন্টের অবস্থা, রিপোর্ট তৈরি হলো কি না), ব্যবহারকারী কম, বা পরিবেশে দীর্ঘ সংযোগ রাখা যায় না, সেখানে polling-ই সবচেয়ে কম ঝুঁকির পথ। Short polling-এর উত্তরে ETag দিলে অপরিবর্তিত উত্তর 304 হয়ে ছোট হয়, কিন্তু অনুরোধের সংখ্যা কমে না।

৩ · Long polling ঠিকভাবে — timeout, cursor আর আবার জিজ্ঞেস করা

Long polling-এ client একই রকম GET পাঠায়, কিন্তু server নতুন কিছু না থাকলে উত্তর দেয় না, অনুরোধটি ধরে রাখে। বদল হলেই উত্তর যায়, আর client সাথে সাথে পরের অনুরোধ পাঠায়। ঠিকভাবে লিখতে চারটি নিয়ম মানতে হয়:

1. Cursor: client বলে দেয় সে কোন version পর্যন্ত জানে (?since=3). দুই অনুরোধের ফাঁকে বদল হলে server সাথে সাথে উত্তর দেয়, কিছু হারায় না।
2. Timeout: server নিজেই একটি সীমা পরে খালি উত্তর (204) দেয়। সীমাটি পথের সব reverse proxy-এর timeout-এর চেয়ে কম হতে হবে। nginx-এর proxy_read_timeout-এর ডিফল্ট ৬০ সেকেন্ড, তাই ২৫–৩০ সেকেন্ড প্রচলিত পছন্দ।
3. পরিষ্কার করা: client চলে গেলে (req.on('close')) timer আর listener সরাতে হবে, না হলে memory জমে।
4. ভুলের পরে ধীরে ফেরা: network ভুলে সাথে সাথে আবার চেষ্টা নয়; ক্রমশ বাড়ানো অপেক্ষা (backoff) আর একটু এলোমেলো সময় (jitter), যাতে server ফিরলে সবাই একসাথে না আসে।

// Express 5: long polling, ল্যাবের server.js-এর ধারণা পরিষ্কার নামে
app.get('/api/orders/:id/wait', async (req, res) => {
  const since = Number(req.query.since ?? 0);
  const now = await orders.current(req.params.id, req.user.id);
  if (!now) return res.sendStatus(404);
  if (now.version > since) return res.json(now);       // আগেই নতুন কিছু আছে
  const off = bus.once(`order:${now.id}`, (o) => { clearTimeout(timer); res.json(o); });
  const timer = setTimeout(() => { off(); res.status(204).end(); }, 25_000);
  req.on('close', () => { clearTimeout(timer); off(); });  // client চলে গেছে
});

// Browser: long polling-এর loop, backoff আর jitter সহ
async function watch(id, since, render) {
  for (let fails = 0; ; ) {
    try {
      const r = await fetch(`/api/orders/${id}/wait?since=${since}`);
      if (r.status === 200) { const o = await r.json(); since = o.version; render(o); }
      fails = 0;                                  // 204 হলে শুধু আবার জিজ্ঞেস
    } catch {
      fails++;
      const wait = Math.min(30_000, 1000 * 2 ** fails) * (0.5 + Math.random() / 2);
      await new Promise((ok) => setTimeout(ok, wait));
    }
  }
}

দুটি ব্লকই উদাহরণ; ল্যাবে চলেছে একই নিয়মের সরল রূপ, যার মাপ ওপরের টেবিলে (৪টি অনুরোধ, ১–২ ms দেরি)। orders.current আর bus এখানে কাল্পনিক helper. একাধিক server process থাকলে bus হতে হবে সবার ভাগ করা কিছু, যেমন পর্ব ৪৪-এর Redis-এর pub/sub; এর নকশা System Design-এর pub/sub পর্বে।

এবার short আর long polling পাশাপাশি, একটি ছোট সময়রেখায়: short polling প্রতি ১০ সেকেন্ডে, long polling-এর timeout ২০ সেকেন্ড, বদল ১৩ আর ৩১ সেকেন্ডে।

1t=0   দুই client প্রথম অনুরোধ পাঠায়2t=10  short: অনুরোধ #23t=13  বদল: placed → confirmed4t=20  short: অনুরোধ #35t=30  short: অনুরোধ #46t=31  বদল: confirmed → packed7t=40  short: অনুরোধ #5

এক ঘণ্টার হিসাব হাতে না করে একটি নকল ঘড়িতে চালান। কোনো সত্যিকারের সময় বা এলোমেলো সংখ্যা নেই, তাই আউটপুট প্রতিবার এক।

// Fake clock in seconds: one order page open for 1 hour, 4 status changes.
const changes = [127, 913, 2411, 3304];
const HOUR = 3600;

function shortPolling(every) {
  let requests = 0, lagSum = 0, seen = 0;
  for (let t = 0; t < HOUR; t += every) {
    requests++;
    while (seen < changes.length && changes[seen] <= t) { lagSum += t - changes[seen]; seen++; }
  }
  return { requests, avgLag: lagSum / seen };
}

function longPolling(timeout) {
  let requests = 0, t = 0, next = 0;
  while (t < HOUR) {
    requests++;                                   // one held request
    const change = changes[next];
    if (change !== undefined && change <= t + timeout) { t = change; next++; }  // answered at once, lag ~0
    else t += timeout;                            // empty timeout, ask again
  }
  return { requests, avgLag: 0 };
}

for (const s of [2, 5, 30]) {
  const r = shortPolling(s);
  console.log(`short poll every ${String(s).padStart(2)} s: ${String(r.requests).padStart(4)} requests, avg lag ${r.avgLag.toFixed(1)} s`);
}
const lp = longPolling(30);
console.log(`long poll, 30 s timeout: ${String(lp.requests).padStart(4)} requests, avg lag ~0 s`);
console.log(`SSE or WebSocket:         1 connection (+ heartbeats), avg lag ~0 s`);
const pages = 2000, r5 = shortPolling(5).requests;
console.log(`2000 open pages, 5 s polling: ${(pages * r5 / HOUR).toFixed(0)} requests/s at the server`);
console.log(`2000 open pages, long poll:   ${(pages * lp.requests / HOUR).toFixed(1)} requests/s, but 2000 held connections`);

৪ · SSE-এর তারের ভাষা — text/event-stream

Server-Sent Events (SSE) আসলে একটি সাধারণ HTTP GET, যার উত্তর শেষ হয় না। Server header-এ বলে Content-Type: text/event-stream, তারপর সময়ে সময়ে কয়েক লাইন লেখা পাঠায়। নিয়মটি WHATWG-এর HTML মানে লেখা, আর এত সরল যে curl দিয়ে চোখে পড়া যায়। ল্যাবের Express server থেকে সত্যিকারের run:

$ curl -N -i http://127.0.0.1:47969/api/orders/42/events    # সত্যিকারের run, Express 5.3.0
HTTP/1.1 200 OK
X-Powered-By: Express
Content-Type: text/event-stream; charset=utf-8
Cache-Control: no-cache
X-Accel-Buffering: no
Connection: keep-alive
Keep-Alive: timeout=5
Transfer-Encoding: chunked

retry: 1000

id: 1
event: status
data: {"order":42,"status":"placed"}

: ping

id: 2
event: status
data: {"order":42,"status":"confirmed"}

: ping

প্রতিটি event কয়েকটি নাম: মান লাইন, আর একটি ফাঁকা লাইনে event শেষ। চারটি নাম অর্থ বহন করে:

লক্ষ করুন, chunked transfer encoding-এ উত্তরের মোট দৈর্ঘ্য আগে বলা নেই, তাই HTTP/1.1-এ উত্তর খোলা রেখে টুকরো পাঠানো যায়। এর বাইরে SSE নতুন কোনো protocol নয়। নিচের ব্লকে node:http দিয়ে একটি ছোট SSE server আর একটি client, যে পুরো স্রোত পড়ে নিজে event-এ ভাঙে।

const http = require("node:http");
// SSE server: one stream for order 42
const server = http.createServer((req, res) => {
  res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache" });
  res.write("retry: 2000\n\n");                       // reconnect wait: 2 s
  res.write(": heartbeat, the browser ignores this\n\n");
  res.write("id: 7\nevent: status\ndata: {\"status\":\"packed\"}\n\n");
  res.write("data: line one\ndata: line two\n\n");    // no event name = "message"
  setTimeout(() => res.end("id: 8\nevent: status\ndata: {\"status\":\"shipped\"}\n\n"), 50);
});

server.listen(0, "127.0.0.1", async () => {
  const url = `http://127.0.0.1:${server.address().port}/orders/42/events`;
  const res = await fetch(url);
  console.log(res.status, res.headers.get("content-type"));
  const text = await res.text();                      // whole stream, then parse
  for (const block of text.split("\n\n")) {
    if (!block) continue;
    const ev = { event: "message", data: [] };
    for (const line of block.split("\n")) {
      if (line.startsWith(":")) { ev.comment = true; continue; }
      const i = line.indexOf(":");
      const field = line.slice(0, i), value = line.slice(i + 1).replace(/^ /, "");
      if (field === "data") ev.data.push(value); else ev[field] = value;
    }
    if (ev.comment) console.log("comment  (ignored)");
    else if (ev.retry) console.log("retry    ", ev.retry, "ms");
    else console.log(`${ev.event.padEnd(8)} id=${ev.id ?? "-"} data=${JSON.stringify(ev.data.join("\n"))}`);
  }
  server.close();
});

টুকরো যেখানে খুশি ভাঙে — একটি শক্ত parser

Network কখনো event-এর সীমা মেনে টুকরো পাঠায় না। একটি TCP টুকরো \r-এর পরে আর \n-এর আগে শেষ হতে পারে। বাংলা অক্ষর UTF-8-এ ৩ byte, টুকরো তার মাঝখানেও ভাঙতে পারে। নিচের parser দুটোই সামলায়: TextDecoder-কে stream: true দিয়ে অর্ধেক অক্ষর ধরে রাখে, আর শেষের অসম্পূর্ণ লাইন পরের টুকরোর জন্য রেখে দেয়।

// An SSE parser that survives chunks split anywhere (even inside a Bangla letter)
function createParser(onEvent) {
  const decoder = new TextDecoder();            // stream: true keeps half a letter
  let buf = "", ev = { data: [] }, lastId = "";
  return function feed(bytes) {
    buf += decoder.decode(bytes, { stream: true });
    const cr = buf.endsWith("\r");              // "\r" may be half of "\r\n"
    const lines = (cr ? buf.slice(0, -1) : buf).split(/\r\n|\r|\n/);
    buf = lines.pop() + (cr ? "\r" : "");       // last piece may be incomplete
    for (const line of lines) {
      if (line === "") {                        // blank line = dispatch
        if (ev.data.length) onEvent({ type: ev.event || "message", id: lastId, data: ev.data.join("\n") });
        ev = { data: [] }; continue;
      }
      if (line[0] === ":") continue;            // comment / heartbeat
      const i = line.indexOf(":");
      const f = i < 0 ? line : line.slice(0, i);
      const v = i < 0 ? "" : line.slice(i + 1).replace(/^ /, "");
      if (f === "data") ev.data.push(v);
      else if (f === "event") ev.event = v;
      else if (f === "id") lastId = v;
    }
  };
}

const wire = new TextEncoder().encode(
  ": ping\r\n\r\nid: 41\r\nevent: status\r\ndata: প্যাক হয়েছে\r\n\r\ndata: a\ndata: b\n\n");
const cuts = [3, 17, 47, 50, 70];   // 17 splits "\r|\n"; 47 and 50 split a letter
const feed = createParser((e) => console.log("  EVENT", JSON.stringify(e)));
let start = 0;
for (const end of [...cuts, wire.length]) {
  console.log(`chunk ${start}-${end}`);
  feed(wire.slice(start, end)); start = end;
}

৫ · Browser-এ EventSource — নিজে থেকে reconnect আর Last-Event-ID

Browser-এ নিজে parser লিখতে হয় না। EventSource object সংযোগ খোলে, event ভাঙে, আর সংযোগ কাটলে নিজেই আবার চেষ্টা করে। ল্যাবের পাতার কোড:

const es = new EventSource('/api/orders/42/events');
es.onopen = () => log('open, readyState =', es.readyState);
es.onerror = () => log('error, readyState =', es.readyState);
es.addEventListener('status', (e) => log('status', e.data, 'lastEventId =', e.lastEventId));
// পাতা ছেড়ে গেলে বা আর দরকার না হলে: es.close();

readyState-এর তিনটি মান: 0 সংযোগ হচ্ছে (CONNECTING), 1 খোলা (OPEN), 2 বন্ধ (CLOSED). Server ভুল status (যেমন 401) বা ভুল Content-Type দিলে browser আর চেষ্টা করে না, অবস্থা হয় 2. শুধু network-এর বিচ্ছেদে আবার চেষ্টা হয়।

এবার আসল পরীক্ষা। Microsoft Edge-কে Playwright দিয়ে চালিয়ে ল্যাবের পাতাটি খোলা হয়েছে। দুটি বদলের পরে server process-টিকে তার নিজের process ID দিয়ে বন্ধ করা হয়েছে। Server বন্ধ থাকার সময়ে database-এর জায়গায় রাখা একটি JSON ফাইলে আরেকটি বদল (v4 shipped) লেখা হয়েছে, যেন অন্য কোনো কর্মী তা করেছে। তারপর নতুন server চালু, আর আরেকটি বদল (delivered). পাতার log আর দুই server-এর log:

# সত্যিকারের run: Edge + Express 5.3.0, একবারের local মাপ
[    4ms] open, readyState = 1
[    5ms] status {"order":42,"status":"placed"} lastEventId = 1
[  609ms] status {"order":42,"status":"confirmed"} lastEventId = 2
[ 1011ms] status {"order":42,"status":"packed"} lastEventId = 3
--- server killed (by PID) ---
[ 1620ms] error, readyState = 0
[ 2631ms] error, readyState = 0
--- (server down) DB records v4 shipped ---
[ 3633ms] error, readyState = 0
--- server restarted ---
[ 4652ms] open, readyState = 1
[ 4652ms] status {"order":42,"status":"shipped"} lastEventId = 4
[ 6152ms] status {"order":42,"status":"delivered"} lastEventId = 5

== server A
SSE connect, Last-Event-ID = (none) cookie = sid=example-session
CHANGE -> confirmed v2
CHANGE -> packed v3
== server B
SSE connect, Last-Event-ID = 3 cookie = sid=example-session
CHANGE -> delivered v5

open, readyState = 1
status placed     lastEventId = 1
status confirmed  lastEventId = 2
status packed     lastEventId = 3

process ID দিয়ে বন্ধ করা হলো
error, readyState = 0   (CONNECTING)
browser মনে রেখেছে: lastEventId = 3

[1620ms] error, readyState = 0
[2631ms] error, readyState = 0
প্রায় ১ সেকেন্ড পরপর চেষ্টা, কারণ
server শুরুতে পাঠিয়েছিল retry: 1000

[{"id":1,"status":"placed"},
 {"id":2,"status":"confirmed"},
 {"id":3,"status":"packed"},
 {"id":4,"status":"shipped"}]
[3633ms] error, readyState = 0

GET /api/orders/42/events
Last-Event-ID: 3
Cookie: sid=example-session
→ history থেকে id > 3 পাঠানো হয়: id 4 shipped

[4652ms] open, readyState = 1
[4652ms] status shipped    lastEventId = 4
[6152ms] status delivered  lastEventId = 5
কোনো event হারায়নি, কোনোটি দুবার আসেনি

এই নকশার মূল দাবি: Last-Event-ID শুধু তখনই কাজের, যখন id-গুলো ক্রমিক বা তুলনা করা যায় এমন, আর ইতিহাস কোথাও টেকসইভাবে রাখা। শুধু memory-তে রাখলে server restart-এ ইতিহাস হারায়। ল্যাবে JSON ফাইল ছিল database-এর বদলি; বাস্তবে এটি database-এর একটি events টেবিল, Redis Streams, বা পর্ব ৪৩-এর outbox টেবিল হতে পারে। ইতিহাস কতদিন রাখবেন, সেটিও নীতি: খুব পুরোনো Last-Event-ID এলে পুরো বর্তমান অবস্থা একবার পাঠিয়ে দেওয়াই নিরাপদ।

Heartbeat কেন লাগে

অর্ডারের অবস্থা হয়তো ২০ মিনিট বদলাবে না। তখন সংযোগে কোনো byte চলবে না। পথে থাকা proxy, load balancer বা মোবাইল অপারেটরের NAT অনেক সময় নীরব সংযোগকে মৃত ধরে কেটে দেয়। আরেক দিকে server জানতেও পারে না যে client আসলে চলে গেছে। সমাধান হলো প্রতি ১৫–৩০ সেকেন্ডে একটি comment লাইন: : ping. Browser এটি উপেক্ষা করে, MDN-ও সংযোগ জাগিয়ে রাখতে এই পথের কথা বলে। বিরতি অবশ্যই পথের সবচেয়ে ছোট idle timeout-এর চেয়ে কম হবে।

৬ · Express 5 আর Next.js 16.4.4-এ SSE

Express 5

ল্যাবের server-এর SSE অংশ, পরিষ্কার নামে সাজানো। এখানে অনুমতি যাচাই, ইতিহাস থেকে ফাঁক পূরণ, heartbeat আর বিদায়ের পরে পরিষ্কার করা, চারটিই আছে।

app.get('/api/orders/:id/events', requireSession, async (req, res) => {
  const order = await orders.findForUser(req.params.id, req.user.id);
  if (!order) return res.sendStatus(404);              // অন্যের অর্ডার নয়
  res.set({
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache, no-transform',
    'X-Accel-Buffering': 'no',
  });
  res.flushHeaders();                                  // header এখনই পাঠান
  res.write('retry: 3000\n\n');
  const last = Number(req.get('Last-Event-ID') ?? 0);
  for (const e of await events.since(order.id, last)) send(res, e);  // ফাঁক পূরণ
  const off = bus.on(`order:${order.id}`, (e) => send(res, e));
  const hb = setInterval(() => res.write(': ping\n\n'), 20_000);
  req.on('close', () => { clearInterval(hb); off(); });
});

function send(res, e) {
  res.write(`id: ${e.id}\nevent: status\ndata: ${JSON.stringify(e.data)}\n\n`);
}

এটি উদাহরণ; requireSession, orders, events, bus কাল্পনিক helper, যেগুলো পর্ব ৪৫ আর পর্ব ৪৮-এর ধাঁচে বানানো যায়। ল্যাবে চলেছে এর সরল রূপ, যার curl আউটপুট ৪ নম্বর অংশে। একটি জিনিস মনে রাখুন: পর্ব ৩১-এর graceful shutdown-এ server খোলা সংযোগ শেষ হওয়ার অপেক্ষা করে। SSE সংযোগ নিজে থেকে শেষ হয় না, তাই shutdown-এর সময় এগুলো নিজে বন্ধ করতে হবে; client তো Last-Event-ID নিয়ে অন্য instance-এ ফিরবেই।

Next.js 16.4.4-এর route handler

পর্ব ৫৭-এ দেখেছেন, route handler একটি Web Response ফেরত দেয়। Response-এর body হতে পারে একটি ReadableStream, যাতে আমরা সময়ে সময়ে byte ঢোকাই (enqueue). ল্যাবের আসল ফাইল:

const enc = new TextEncoder();
const steps = ['confirmed', 'packed', 'shipped'];

export async function GET(request: Request, ctx: RouteContext<'/api/orders/[id]/events'>) {
  const { id } = await ctx.params;
  const last = Number(request.headers.get('last-event-id') ?? 1);
  let timer: ReturnType<typeof setInterval> | undefined;
  const stream = new ReadableStream<Uint8Array>({
    start(controller) {
      controller.enqueue(enc.encode('retry: 3000\n\n'));
      let v = last;
      timer = setInterval(() => {
        v += 1;
        if (v > steps.length + 1) { clearInterval(timer); controller.close(); return; }
        const data = JSON.stringify({ order: Number(id), status: steps[v - 2] });
        controller.enqueue(enc.encode(`id: ${v}\nevent: status\ndata: ${data}\n\n`));
      }, 400);
    },
    cancel() { clearInterval(timer); console.log('client went away'); },
  });
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      'X-Accel-Buffering': 'no',
    },
  });
}

এটি শেখার জন্য: আসল অ্যাপে setInterval-এর বদলে event store আর pub/sub থাকবে। next build-এর route table আর next start-এর পরে প্রতিটি টুকরো কখন পৌঁছায়, তার সত্যিকারের মাপ (Next.js 16.4.4, React 19.3.1, Node 24.15.0):

$ next build                                  # সত্যিকারের run
Route (app)
┌ ○ /
├ ○ /_not-found
├ ƒ /api/chat
├ ƒ /api/orders/[id]/events
└ ƒ /api/plain

$ node client.mjs /api/orders/42/events      # next start, 127.0.0.1
/api/orders/42/events  -> 200 content-type=text/event-stream; charset=utf-8 content-encoding=- transfer-encoding=chunked
  +  25 ms  13 bytes  "retry: 3000\n\n"
  + 437 ms  61 bytes  "id: 2\nevent: status\ndata: {\"order\":42,\"status\":\"confirmed\"}\n
  + 846 ms  58 bytes  "id: 3\nevent: status\ndata: {\"order\":42,\"status\":\"packed\"}\n\n"
  +1258 ms  59 bytes  "id: 4\nevent: status\ndata: {\"order\":42,\"status\":\"shipped\"}\n\n
  end at +1669 ms

$ curl -s -N --max-time 0.6 http://127.0.0.1:47968/api/orders/42/events
retry: 3000

id: 2
event: status
data: {"order":42,"status":"confirmed"}
# server-এর log-এ:
client went away

তিনটি ফল স্পষ্ট। প্রথমত, route-টি ƒ (dynamic), প্রতি অনুরোধে চলে। দ্বিতীয়ত, buffer হয়নি: প্রতিটি event প্রায় ৪০০ ms পরপর নিজের টুকরোয় এসেছে, পুরো উত্তরের শেষে একসাথে নয়। তৃতীয়ত, curl ০.৬ সেকেন্ডে চলে যেতেই stream-এর cancel() ডাকা হয়েছে, log-এ client went away. এই hook না থাকলে timer চলতেই থাকত। আরেকটি পরীক্ষা: Accept-Encoding: gzip পাঠালে এই একই run-এ HTML পাতা (/) gzip হয়ে এসেছে, কিন্তু দুটি SSE route-এর কোনোটিই compress হয়নি, এমনকি no-transform ছাড়া route-টিও না। এটি এই version-এ একবারের পর্যবেক্ষণ; ভবিষ্যতের version বা আপনার সামনের proxy আলাদা আচরণ করতে পারে, তাই নিজের deploy-এ মেপে নেবেন।

৭ · পথের বাধা — সংযোগের সীমা, proxy, compression আর backpressure

SSE সরল, কিন্তু browser আর server-এর মাঝের প্রতিটি স্তর এটিকে ভাঙতে পারে। চারটি বাধা সত্যিকারের run-এ দেখুন।

ক. HTTP/1.1-এ ছয়টি সংযোগ

HTTP/1.1-এ একটি সংযোগে একসাথে একটিই উত্তর চলে। Browser তাই এক host-এ কয়েকটি সমান্তরাল সংযোগ খোলে, Chromium আর Firefox-এ ৬টি। প্রতিটি SSE একটি সংযোগ চিরদিনের জন্য দখল করে। MDN বলছে, এই সীমা browser আর domain মিলিয়ে, সব tab একসাথে, আর Chrome ও Firefox-এ এটি “Won't fix” হিসেবে চিহ্নিত। HTTP/2-তে একটি TCP সংযোগে বহু stream চলে (multiplexing); সংখ্যাটি দুই পক্ষ ঠিক করে, RFC 9113 অন্তত ১০০ রাখার পরামর্শ দেয়। একটি পাতা ৮টি EventSource খুলে তারপর একটি সাধারণ fetch করেছে, প্রথমে HTTP/1.1-এ, তারপর TLS-সহ HTTP/2-তে (node:http2, নিজের বানানো পরীক্ষার certificate):

# সত্যিকারের run: Microsoft Edge, ২.৫ সেকেন্ড দেখা
== HTTP/1.1
[    5ms] stream 1 open
[    6ms] stream 2 open
  …  (3, 4, 5)
[    6ms] stream 6 open
                       ← stream 7, 8 আর fetch: কোনো উত্তর নেই
== HTTP/2 over TLS
[    7ms] stream 1 open
  …  (2 থেকে 7)
[   10ms] stream 8 open
[  323ms] fetch /api/ping -> pong (2.0)

HTTP/1.1-এ সপ্তম আর অষ্টম stream খোলেইনি, আর fetch-ও আটকে গেছে, কারণ ছয়টি সংযোগই SSE-র দখলে। অর্থাৎ ভুলটি শুধু “কিছু notification আসছে না” নয়, পুরো সাইটের API থেমে যায়। HTTP/2-তে আটটিই খুলেছে, fetch-ও উত্তর পেয়েছে। বাস্তবে production-এ HTTP/2 সাধারণত সামনের proxy বা CDN দেয়; Node server নিজে HTTP/1.1 বললেও browser থেকে proxy পর্যন্ত পথটি HTTP/2 হতে পারে। তবু নকশার নিয়ম: এক পাতায় একটিই SSE সংযোগ, তার ভেতরে event নাম দিয়ে নানা খবর। একাধিক tab-এ সংযোগ ভাগ করতে BroadcastChannel বা SharedWorker ব্যবহারের কৌশলও প্রচলিত।

খ. Proxy-র buffering

nginx-এর মতো reverse proxy ডিফল্টে পেছনের server-এর উত্তর জমিয়ে (buffer করে) তারপর client-কে দেয়। সাধারণ পাতার জন্য এটি ভালো, SSE-র জন্য মারাত্মক: event-গুলো জমে থাকে, বড় দলা হয়ে দেরিতে পৌঁছায়। nginx docs অনুযায়ী proxy_buffering on ডিফল্ট, আর উত্তরে X-Accel-Buffering: no header দিলে সেই উত্তরের জন্য buffering বন্ধ হয়। তাই আমাদের দুই server-ই এই header পাঠায়। সাথে proxy_read_timeout-এর ডিফল্ট ৬০ সেকেন্ড, তাই heartbeat এর চেয়ে ঘন হওয়া চাই। এই পর্বের ল্যাবে nginx চালানো হয়নি; এই অংশ nginx-এর অফিসিয়াল docs থেকে, আর অন্য proxy বা CDN-এর নিয়ম আলাদা হতে পারে।

গ. Compression স্রোত আটকে দেয়

gzip ভালো সংকোচন করতে কিছু byte জমিয়ে তারপর লেখে। Express-এর জনপ্রিয় compression middleware (1.8.2) একই Express server-এ লাগিয়ে তিনটি অবস্থা মাপা হয়েছে। Heartbeat ছিল ৩০০ ms পরপর, একটি বদল ৫০০ ms-এ।

$ node compress.js        # সত্যিকারের run, একবারের local মাপ
== compression(), no flush: headers after 28 ms, content-encoding = gzip
no body bytes in 2 s
== compression() + res.flush(): headers after 5 ms, content-encoding = gzip
7 ms: 71 bytes
525 ms: 69 bytes
== no compression: headers after 3 ms, content-encoding = null
4 ms: 71 bytes
314 ms: 8 bytes
514 ms: 61 bytes
617 ms: 8 bytes
… 8 chunks in 2 s
== compression(), Cache-Control: no-transform, no flush: content-encoding = null
27 ms: 71 bytes
339 ms: 8 bytes
541 ms: 61 bytes
… 8 chunks in 2 s

প্রথম অবস্থায় দুই সেকেন্ডে একটি byte-ও আসেনি: সব gzip-এর পেটে। দ্বিতীয় অবস্থায় শুধু event লেখার পরে res.flush() ডাকা হয়েছিল, heartbeat-এর পরে নয়। তাই ৩০০ ms-এর heartbeat আটকে ছিল, আর ৫২৫ ms-এ event-এর সাথে একসাথে বেরিয়েছে: ৬১ + ৮ = ৬৯ byte. শিক্ষা: compression রাখলে প্রতিটি লেখার পরে flush. সহজ পথ হলো SSE-তে compression না করা। শেষ run দেখায়, compression package Cache-Control: no-transform পেলে নিজেই সংকোচন বাদ দেয়; এটি package-এর নিজের নিয়ম, অন্য middleware বা proxy-তে আলাদা হতে পারে।

ঘ. Backpressure — ধীর পাঠক

দ্রুত লেখক আর ধীর পাঠকের সমস্যা পর্ব ২৪-এর stream-এ দেখেছেন। SSE-তে পাঠক হলো দুর্বল মোবাইল network-এর একজন ক্রেতা। res.write() false ফেরত দিলে বোঝায়, ভেতরের buffer ভরে গেছে, 'drain' পর্যন্ত থামুন। একটি client যে পড়া বন্ধ করে দিয়েছে, তার দিকে দেড় সেকেন্ড ধরে প্রায় ১ KB-র event লেখা হয়েছে:

$ node backpressure.js    # সত্যিকারের run, একবারের local মাপ
naive       events=5650  held in Node memory=2.7 MB  drain-waits=0
drain-aware events=1750  held in Node memory=0.0 MB  drain-waits=103

সাধারণ লেখক থামেনি; ২.৭ MB জমে গেছে Node-এর memory-তে, আর সময় বাড়লে এটি বাড়তেই থাকত। এমন কয়েকশো client মানে server-এর memory শেষ। Backpressure মানা লেখক ১০৩ বার থেমেছে, Node-এ প্রায় কিছুই জমেনি; বাকি byte অপারেটিং সিস্টেমের socket buffer-এ। বাস্তব নিয়ম: buffer একটি সীমা পেরোলে সেই client-কে বিচ্ছিন্ন করুন; সে Last-Event-ID নিয়ে পরে ফিরবে। ReadableStream-এ একই ধারণা আসে pull() আর desiredSize দিয়ে: পাঠক চাইলে তবেই উৎপাদন। বড় পরিসরের overload নিয়ন্ত্রণ System Design-এর overload আর backpressure পর্বে।

৮ · AI-র উত্তর stream করা — token, React আর AbortController

বাজারঘরের সাহায্য পাতায় একটি AI সহকারী যোগ হলো: “আমার অর্ডার কবে আসবে?” একটি language model পুরো উত্তর একবারে বানায় না, একটি একটি token (শব্দ বা শব্দের অংশ) করে বানায়। পুরো উত্তরের জন্য ১০ সেকেন্ড অপেক্ষা করালে মনে হয় পাতা আটকে গেছে। তাই প্রায় সব AI chat প্রথম token আসতেই লেখা দেখাতে শুরু করে। অনেক AI API-র streaming উত্তর ঠিক এই পর্বের text/event-stream ছাঁচেই আসে, তবে নাম আর JSON-এর আকার provider ভেদে আলাদা।

দুটি কারণে এখানে EventSource চলে না। এটি শুধু GET পাঠায়, অথচ প্রশ্নটি body-তে POST করতে হয়। আর এটি নিজের header বসাতে দেয় না। তাই client fetch দিয়ে POST করে, আর response.body, একটি ReadableStream, reader দিয়ে টুকরো টুকরো পড়ে। Next.js-এর ল্যাবে একটি /api/chat route ছিল, যা ১৫০ ms পরপর একটি নকল token পাঠায়, আর client বাতিল করলে cancel()-এ উৎপাদন থামায়। Node-এর fetch দিয়ে (browser-এর সেই একই API) সত্যিকারের run, প্রথমে পুরো উত্তর, তারপর ৫২০ ms-এ abort():

$ node chat-client.mjs          # সত্যিকারের run, Next.js 16.4.4, নকল token
+  50 ms  [start] {"id":"msg_1"}
+ 192 ms  ডেলিভারি
+ 343 ms  ডেলিভারি সাধারণত
+ 494 ms  ডেলিভারি সাধারণত ঢাকার
  …
+1251 ms  ডেলিভারি সাধারণত ঢাকার ভেতরে ২–৩ দিনে হয়।
+1403 ms  [done] {"tokens":8}

$ node chat-client.mjs 520      # ৫২০ ms-এ AbortController.abort()
+  31 ms  [start] {"id":"msg_1"}
+ 186 ms  ডেলিভারি
+ 338 ms  ডেলিভারি সাধারণত
+ 490 ms  ডেলিভারি সাধারণত ঢাকার
+523 ms  AbortError: stopped, kept "ডেলিভারি সাধারণত ঢাকার"
# next start-এর log:
chat: client aborted after 3 tokens, generation stopped

শেষ লাইনটিই সবচেয়ে জরুরি। ব্যবহারকারী “থামান” চাপলে শুধু পর্দায় লেখা থামানো যথেষ্ট নয়। Server-কে জানতে হবে, আর server-কেও তার পেছনের AI provider-এর অনুরোধ বাতিল করতে হবে; নইলে কেউ পড়ছে না এমন token-এর জন্য সময় আর (অনেক provider-এ) টাকা খরচ হতেই থাকবে। Route handler-এ request.signal upstream-এর fetch-এ দিয়ে দিলে, বা upstream-এর body সরাসরি নিজের Response-এ দিলে, বাতিলের খবর পুরো পথে ছড়ায়। (এটি নকশার পরামর্শ; আসল provider-এর সাথে এই পর্বে পরীক্ষা হয়নি।)

এবার React-এ। নিচের live ব্লকে fakeAnswer() একটি স্থানীয় ReadableStream বানায়, কোনো network নেই। কম্পোনেন্ট reader দিয়ে পড়ে, প্রতিটি token-এ setText ডাকে, আর “থামান” বোতাম AbortController দিয়ে স্রোত ভাঙে।

<div id="root"></div>
<script src="../course-kit/vendor/react-lab.js"></script>
<script type="text/jsx">
// No real AI is called: fakeAnswer() makes a local stream of tokens.
const TOKENS = ["অর্ডার", " ৪২", " প্যাক", " হয়েছে,", " আজ", " রাতে", " রওনা", " দেবে।"];
function fakeAnswer(signal) {
  let i = 0, t;
  return new ReadableStream({
    start(c) { t = setInterval(() => i < TOKENS.length ? c.enqueue(TOKENS[i++]) : (clearInterval(t), c.close()), 60);
      signal.addEventListener("abort", () => { clearInterval(t); c.error(new DOMException("stopped", "AbortError")); }); },
  });
}
function Chat() {
  const [text, setText] = useState(""); const [busy, setBusy] = useState(false); const ctl = useRef(null);
  async function ask() {
    ctl.current = new AbortController(); setText(""); setBusy(true); let all = "";
    const reader = fakeAnswer(ctl.current.signal).getReader();
    try { for (;;) { const { value, done } = await reader.read(); if (done) break; all += value; setText(all); } console.log("done:", all); }
    catch (e) { console.log(e.name + ", kept:", all); }
    setBusy(false);
  }
  useEffect(() => { ask(); }, []);
  return <div><p style={{ minHeight: "2em" }}>{text}{busy && "▌"}</p>
    <button onClick={ask} disabled={busy}>আবার জিজ্ঞেস করুন</button>{" "}
    <button onClick={() => ctl.current.abort()} disabled={!busy}>থামান</button></div>;
}
createRoot(document.getElementById('root')).render(<Chat />);
</script>

const ac = new AbortController();
fetch('/api/chat', { method: 'POST',
  body: '{"q":"delivery?"}', signal: ac.signal })

event: start
data: {"id":"msg_1"}

(একটি ফাঁকা লাইন = event শেষ)

event: delta
data: {"text":"ডেলিভারি"}

reader.read() → { value, done: false }
parser → delta "ডেলিভারি"

text = "ডেলিভারি সাধারণত ঢাকার"
setText(text) → re-render
পর্দায়: ডেলিভারি সাধারণত ঢাকার▌

ac.abort()
client: AbortError, kept "ডেলিভারি সাধারণত ঢাকার"
server: chat: client aborted after 3 tokens,
        generation stopped

Streaming-এর আরও কয়েকটি বাস্তব প্রশ্ন আছে। আংশিক Markdown (অর্ধেক ** বা অর্ধেক code block) দেখানোর সময় renderer যেন ভেঙে না যায়। AI-র লেখা কখনো সরাসরি HTML হিসেবে বসাবেন না; সাধারণ লেখা হিসেবে দেখান বা sanitize করুন (পর্ব ৫০)। Screen reader-এর জন্য প্রতিটি token ঘোষণা না করে উত্তর শেষে একবার ঘোষণা করাই ভালো। আর Next.js-এর পাতার HTML streaming (Suspense দিয়ে) আলাদা জিনিস, সেটি পর্ব ৫৩-এ; এখানে আমরা ডেটার স্রোত নিয়ে কথা বলছি।

৯ · SSE-তে auth, আর কখন কোন পথ

Cookie চলে, নিজের header চলে না

EventSource-এর constructor শুধু URL আর একটি option নেয়: withCredentials. Authorization: Bearer … header বসানোর কোনো উপায় নেই। ওপরের reconnect run-এ দেখেছেন, একই origin-এ cookie নিজে থেকেই গেছে (cookie = sid=example-session), পুনঃসংযোগেও। তাই session cookie দিয়ে login করা অ্যাপে (পর্ব ৪৫) SSE সহজ। অন্য পথগুলো:

• অন্য origin-এ API: new EventSource(url, { withCredentials: true }), আর server-এ Access-Control-Allow-Credentials: true সহ নির্দিষ্ট origin (* নয়)। CORS-এর নিয়ম পর্ব ৫০-এ।
• URL-এ token দেবেন না। ?token=… server log, proxy log আর browser history-তে থেকে যায়। একান্তই লাগলে খুব কম আয়ুর, একবার ব্যবহারযোগ্য “ticket” দিন, যা সংযোগের সময়ই বাতিল হয়।
• Header দরকার হলে fetch-ভিত্তিক SSE: AI অংশের মতো fetch আর reader দিয়ে নিজে parse করুন। তবে তখন reconnect আর Last-Event-ID নিজেকে লিখতে হবে।
• দীর্ঘ সংযোগে মেয়াদ: সংযোগ খোলার সময় session বৈধ ছিল, কিন্তু ঘণ্টা পরে logout বা অনুমতি বাতিল হতে পারে। মাঝে মাঝে session আবার যাচাই করুন, বা logout-এ সেই ব্যবহারকারীর stream বন্ধ করুন। প্রতিটি event পাঠানোর আগে অনুমতির নিয়ম মানুন, শুধু সংযোগের শুরুতে নয়।

Token-ভিত্তিক login-এর তত্ত্ব Cybersecurity-র sessions আর tokens পর্বে।

কোনটি কখন

এখানে সম্প্রদায়ে মতভেদ আছে। কেউ বলেন, সব কিছুতে WebSocket নিলে একটি প্রযুক্তিই শিখতে হয়। অন্যরা বলেন, বেশির ভাগ অ্যাপের real-time আসলে এক দিকের, আর SSE সাধারণ HTTP হওয়ায় auth, log, load balancer সবই পুরোনো নিয়মে চলে। দুই পক্ষেরই যুক্তি আছে; বাজারঘরের অর্ডার পাতার জন্য SSE যথেষ্ট, আর বিক্রেতা-ক্রেতার chat-এর জন্য WebSocket. লাখো সংযোগের capacity আর reconnect storm-এর নকশা System Design-এ।

১০ · AI-এর যুগে SSE আর streaming

AI-কে real-time order status বানাতে বললে সে সাধারণত কাজ চলার মতো কোড দেয়। সমস্যা লুকিয়ে থাকে এমন জায়গায়, যা local-এ ধরা পড়ে না: proxy, reconnect, memory, auth. নিচের ভুলগুলো খেয়াল রাখুন।

AI-এর লেখা streaming কোড মেলানোর চেকলিস্ট

• curl -N দিয়ে নিজে দেখেছি, প্রতিটি event সময়মতো আলাদা আসে, শেষে একসাথে নয়
• Server বন্ধ করে আবার চালিয়েছি; browser Last-Event-ID পাঠিয়েছে আর কোনো event হারায়নি
• সংযোগ বন্ধ করলে server-এর log-এ পরিষ্কারের লাইন আসে, timer আর listener থাকে না
• HTTP/1.1-এ এক পাতায় একটির বেশি SSE খোলা নেই, আর production-এ HTTP/2 আছে কি না দেখেছি
• AI stream “থামান”-এ server-এর কাজও থামে, আর provider-এর key কখনো browser-এ যায় না
• অন্য ব্যবহারকারীর অর্ডারের stream চেয়ে 404 বা 403 পেয়েছি