Server-Sent Events NestJS'da: @Sse, Observable va MessageEvent

Qisqacha: Nest'da SSE endpoint — @Sse("oqim") dekoratorli metod, u Observable<MessageEvent> qaytaradi. Har next({ type, data }) — mijozga bitta hodisa: Nest uni event:, id:, data: qatorlariga o'zi aylantiradi va text/event-stream sarlavhalarini qo'yadi. Mijoz uzilsa, Nest obunani bekor qiladi va Observable'ning teardown funksiyasi ishlaydi — tinglovchini shu yerda olib tashlaysiz. Bilib qo'ying: id ni bermasangiz, Nest har ulanish uchun 1 dan sanaydi va Last-Event-ID bilan qayta ulanishni o'zi qo'llamaydi.

Bu darsda

• @Sse() va Observable bilan SSE endpoint yozasiz va Nest yuboradigan xom matnni o'qiysiz.
• MessageEvent ning besh maydonini (data, type, id, retry, comment) ishlatasiz.
• Mijoz uzilganda teardown va @SseSignal() qanday ishlashini ko'rasiz.
• Oqim o'rtasidagi xato, sekin mijoz va avtomatik id tuzoqlarini bilasiz.
• vazifalar-api ga /v1/vazifalar/oqim qo'shib, loyiha bosqichini commit qilasiz.

Oldin bilishingiz kerak: Server-Sent Events, Interceptor'lar va RxJS, Hodisalar: @nestjs/event-emitter, Health check'lar.

Oldingi uch darsda rejali tozalash, domen hodisalari va salomatlik tekshiruvi qo'shildi. Bu dars bosqichni yakunlaydi: xizmat chiqarayotgan vazifa.* hodisalarini brauzerga jonli oqim qilib yuboramiz va to'rt darsning ishini bitta commit qilamiz.

1. Nega bu kerak?

Vazifalar ilovasi telefonda va kompyuterda ochiq. Telefonda vazifani "bajarildi" qildingiz. Kompyuterdagi ro'yxat esa eski holatda — sahifani yangilamaguningizcha. 29-qismda shu muammoni SSE bilan hal qilgandik (Server-Sent Events): server ochiq HTTP javobga hodisalarni yozib boradi, brauzerdagi EventSource ularni o'qiydi.

O'sha darsda node:http va Express'da SSE'ni qo'lda yozdik: sarlavhalar, id:/data: qatorlari, res.write, uzilishni req.on("close") bilan kuzatish. Nest'da bu ishning katta qismini freymvork oladi. Siz faqat hodisalar oqimini berasiz.

Oqim nima bilan beriladi? Nest javobni Observable sifatida kutadi — RxJS'ning vaqt bo'ylab qiymat chiqaradigan obyekti. Interceptor'lar va RxJS darsida Observable'ni ko'rgandik: subscribe qilinadi, next bilan qiymat chiqaradi va obuna bekor bo'lganda teardown funksiyasi ishlaydi. SSE'ga aynan shu kerak.

O'xshatish: radio. @Sse endpoint — to'lqin, Observable — diktor. Tinglovchi radioni yoqadi (ulanadi), diktor yangilik o'qiydi (next), tinglovchi o'chirsa (uziladi) — diktorning shu tinglovchi uchun ishi tugaydi.

2. Eng kichik @Sse endpoint

2.1 Kod va xom javob

Biz demo loyihada (NestJS 12.2.0, rxjs 7.8.2) to'rt hodisa chiqarib tugaydigan endpoint yozdik:

import { Controller, Sse } from "@nestjs/common";
import type { MessageEvent } from "@nestjs/common";
import { Observable, of } from "rxjs";

@Controller()
class OqimController {
  @Sse("qisqa")
  qisqa(): Observable<MessageEvent> {
    return of(
      {
        type: "vazifa.yaratildi", id: "1",
        data: { id: 1, matn: "Non olish" },
      },
      { data: "birinchi qator\nikkinchi qator" },
      { comment: "ping" },
      { retry: 5000, data: "oddiy satr" },
    );
  }
}

of(...) — berilgan qiymatlarni ketma-ket chiqarib, tugaydigan Observable. Javobni fetch bilan to'liq o'qidik va JSON.stringify bilan chiqardik — \n lar ko'rinsin (biz ishga tushirdik):

1) /qisqa 200
   content-type: text/event-stream
   cache-control: private, no-cache, no-store, must-revalidate, max-age=0, no-transform
   connection: keep-alive
   x-accel-buffering: no
"\nevent: vazifa.yaratildi\nid: 1\ndata: {\"id\":1,\"matn\":\"Non olish\"}\n\nid: 1\ndata: birinchi qator\ndata: ikkinchi qator\n\n: ping\n\nid: 2\nretry: 5000\ndata: oddiy satr\n\n"

2.2 Sarlavhalar

Hamma sarlavhani Nest o'zi qo'ydi. 29-qismda ularni qo'lda yozgandik, nima uchunligini eslaylik:

Oxirgisi eng ko'p unutiladigani edi: nginx sukut bo'yicha javobni yig'ib, keyin yuboradi va hodisalar bir necha soniya kechikib keladi. Nest buni ham qo'yadi.

2.3 MessageEvent maydonlari

Xom matnni to'rt hodisaga ajratib o'qing — har biri bo'sh qator (\n\n) bilan tugaydi:

1. type → event: vazifa.yaratildi. Brauzerda addEventListener("vazifa.yaratildi", …) bilan tinglanadi. type bo'lmasa, hodisa sukut nomi message bilan keladi.
2. data obyekt bo'lsa — Nest uni JSON.stringify qiladi. Satr bo'lsa — har qatorini alohida data: qatoriga bo'ladi (ikkinchi hodisa). Brauzer ularni qayta \n bilan birlashtiradi. Qo'lda yozganda bu qoidani unutib, ko'p qatorli matnni buzish oson edi.
3. comment → : ping. Ikki nuqta bilan boshlangan qator — izoh: brauzer uni e'tiborsiz qoldiradi. Ulanishni "tirik" ushlash uchun ideal.
4. retry → retry: 5000 — uzilsa, brauzer 5 soniyadan keyin qayta ulansin.

Va eng qiziq joy — id. Birinchi hodisaga id: "1" ni o'zimiz berdik. Ikkinchi va to'rtinchisiga bermadik, lekin javobda id: 1 va id: 2 bor. Nest id ni o'zi qo'ydi. Natijada oqimda ikkita id: 1 paydo bo'ldi. Bu tuzoqni «id tuzog'i» bo'limida batafsil ko'ramiz — u loyihaga ham tegishli.

Tekshirib ko'ring: next({ data: { matn: "a\nb" } }) qilsak, nechta data: qatori chiqadi?

3. Ulanish hayoti

3.1 Cheksiz oqim va teardown

Haqiqiy SSE tugamaydi — mijoz o'zi uziladi. Demoda har 30 ms da hodisa chiqaradigan Observable yozdik va har bosqichni jurnalga yozdik:

@Sse("cheksiz")
cheksiz(@SseSignal() signal: AbortSignal): Observable<MessageEvent> {
  signal.addEventListener("abort", () => {
    jurnal.push("signal: abort");
  });
  return new Observable<MessageEvent>((obunachi) => {
    jurnal.push("obuna boshlandi");
    let n = 0;
    const t = setInterval(() => {
      obunachi.next({ data: { n: ++n } });
    }, 30);
    return () => {
      clearInterval(t);
      jurnal.push(`teardown: ${n} ta hodisadan keyin`);
    };
  });
}

new Observable(fn) — fn obuna paytida ishlaydi va teardown funksiyasini qaytaradi. Mijoz tomonda uchta hodisa o'qib, AbortController bilan uzdik:

2) /cheksiz — 3 hodisa oldik va uzdik: ["\nid: 1\ndata: {\"n\":1}","id: 2\ndata: {\"n\":2}","id: 3\ndata: {\"n\":3}"]
   jurnal: ["obuna boshlandi","signal: abort","teardown: 3 ta hodisadan keyin"]

Tartib: obuna mijoz ulanganda boshlandi. Mijoz uzilishi bilan Nest ikki ishni qildi — @SseSignal() signalini bekor qildi va Observable obunasini bekor qildi. Teardown ishladi, clearInterval taymerni to'xtatdi. Teardown bo'lmasa, uzilgan mijoz uchun taymer abadiy aylanardi — xotira sizishi (memory leak).

sequenceDiagram
  participant B as Brauzer
  participant N as Nest
  participant O as Observable
  B->>N: GET /cheksiz
  N->>O: subscribe()
  O-->>N: next(hodisa)
  N-->>B: id/data qatorlari
  B--xN: uzildi
  N->>O: unsubscribe()
  O->>O: teardown: clearInterval

Diagrammani yuqoridan pastga o'qing: subscribe — ulanishda, unsubscribe — uzilishda. Ikkalasini Nest chaqiradi, siz faqat teardown'ni yozasiz.

3.2 @SseSignal()

@SseSignal() — parametr dekoratori, AbortSignal beradi (AbortController). U SSE javobi tugaganda — uzilish, tugash yoki xato — bekor bo'ladi. Teardown bor ekan, nega kerak? Teardown faqat Observable yaratilgandan keyin ishlaydi. Agar metod async bo'lib, Observable'dan oldin og'ir tayyorgarlik qilsa (bazadan tarix o'qish, tashqi xizmatga ulanish), mijoz shu orada uzilishi mumkin. Signal shu "oraliq" vaqtni ham qamraydi: if (signal.aborted) return EMPTY;.

4. Xatolar

4.1 Oqim o'rtasida xato

@Sse("xatoli")
xatoli(): Observable<MessageEvent> {
  return concat(
    of({ data: "birinchi" }),
    throwError(() => new Error("ombor uzildi")),
  );
}

ERROR [RouterResponseController] Error: ombor uzildi
3) /xatoli 200 "\nid: 1\ndata: birinchi\n\nevent: error\nid: 2\ndata: Internal server error\n\n"

Status 200 — u birinchi hodisa bilan allaqachon yuborilgan, uni o'zgartirib bo'lmaydi. Shuning uchun Nest xatoni hodisa qilib yubordi: event: error, data: Internal server error, keyin oqimni yopdi. Xato matni ("ombor uzildi") mijozga ketmadi, faqat serverdagi logga — bu xavfsiz tanlov. HttpException bo'lganda esa uning xabari yuboriladi.

Brauzer tomonida ehtiyot bo'ling: EventSource da error nomi band — u ulanish xatolari uchun ham ishlaydi. addEventListener("error", …) ikkala holatni ham oladi.

Ikkinchi holat — metod Observable emas, oddiy obyekt qaytardi:

ERROR [ExceptionsHandler] ReferenceError: You must return an Observable stream to use Server-Sent Events (SSE).
4) /notogri 500 {"statusCode":500,"message":"Internal server error"}

Tarjimasi: "Server-Sent Events ishlatish uchun Observable oqim qaytarishingiz kerak". Bu yerda hali hech narsa yuborilmagan — shuning uchun oddiy 500 va xato filtri ishladi.

4.2 Sekin mijoz

Mijoz hodisalarni o'qimayapti — internet sekin, yoki tab "uxlab" qolgan. Server esa yozishda davom etadi. Hodisalar qayerda yig'iladi? Nest xotirasida. 29-qismda buni "sekin mijoz" tuzog'i deb o'rgangandik.

Nest 12.2 da himoya bor — maxPendingDataLength: mijoz o'qimagan data va comment belgilari shu sondan oshsa, ulanish uziladi. Sukut — 4 194 304 belgi (4M). Demoda limitni 200 000 qilib, 2 KB lik 2 000 ta hodisa chiqardik. Mijoz sifatida xom TCP ulanish ochdik va undan hech narsa o'qimadik:

@Sse("katta", { maxPendingDataLength: 200_000 })
katta(): Observable<MessageEvent> {
  return interval(1).pipe(
    take(2000),
    map((i) => ({ id: String(i), data: "x".repeat(2048) })),
    finalize(() => jurnal.push("katta: obuna tugadi")),
  );
}

WARN [RouterResponseController] The client is not reading its events: 200704 character(s) of event data are queued for it, which exceeds the maximum allowed 200000
5) o'qimaydigan mijoz, limit 200 000 belgi: ["katta: obuna tugadi"] mijoz close: false

Tarjimasi: "Mijoz hodisalarini o'qimayapti: unga 200 704 belgi navbatda turibdi, bu ruxsat etilgan 200 000 dan ko'p". Nest ogohlantirish yozdi, ulanishni uzdi va obunani tugatdi (finalize ishladi). mijoz close: false — bizning "mijoz" o'qishni to'xtatgani uchun uzilishni ham sezmadi; server tomoni esa tozalandi, muhimi shu.

4M belgi — bitta mijoz uchun bir necha megabayt xotira. 1 000 ta sekin mijoz — gigabaytlar. Hodisalari kichik loyihada limitni kamaytirish mumkin: @Sse("oqim", { maxPendingDataLength: 256 * 1024 }). 29-qismdagi hub ham 256 KB da uzardi.

5. id tuzog'i va qayta ulanish

SSE'ning eng kuchli joyi — qayta ulanish. Brauzer uzilsa, o'zi qayta ulanadi va oxirgi olgan id ni Last-Event-ID sarlavhasida yuboradi. Server o'tkazib yuborilgan hodisalarni qayta beradi (Server-Sent Events).

@Sse buni o'zi qilmaydi. Biz ko'rgan uch fakt:

• id bermasangiz, Nest uni har ulanish uchun 1 dan sanaydi. Qayta ulangan mijoz yana id: 1 dan boshlaydi.
• Qo'lda va avtomatik id aralashsa, takror chiqadi (/qisqa dagi ikkita id: 1).
• Last-Event-ID sarlavhasini Nest o'qimaydi — uni @Headers("last-event-id") bilan o'zingiz olib, tarixni o'zingiz qayta yuborishingiz kerak.

Demak @Sse + in-process hodisalar — "eng yaxshi urinish" oqimi: ulangan paytda kelgan hodisalarni ko'rasiz, uzilish paytidagilari yo'qoladi. Frontend uchun amaliy qoida: qayta ulangandan keyin ro'yxatni bir marta GET /v1/vazifalar bilan yangilash.

To'liq kafolat uchun hodisalar tarixi kerak: 29-qismdagi SseHub Redis Stream'dan Last-Event-ID dan keyingi hodisalarni o'qib berardi. 29-qism integratsiyalari Nest'ga ko'chganda o'sha hub ham /v1/oqim sifatida ko'chadi va ikkala yondashuvni tezlik va xotira bo'yicha WebSocket gateway'lar darsida o'lchab solishtiramiz.

Tekshirib ko'ring: Loyihadagi oqimda vazifa hodisasi id: 3 bilan keldi. Bu vazifaning id'si 3 degani-mi?

6. JS modelida Nest formatlovchisi

Nest MessageEvent ni qanday matnga aylantiradi? Nest manbasidagi qoidani paketsiz takrorlaymiz — blok brauzerda ishlaydi va demodagi /qisqa chiqishini aynan beradi:

const toza = (v) => String(v).replace(/[\r\n]/g, "");
const qatorlar = (matn, old) =>
  matn.split(/\r\n|\r|\n/).map((q) => `${old}${q}\n`).join("");

function formatla(x) {
  let s = x.type ? `event: ${toza(x.type)}\n` : "";
  if (x.id != null) s += `id: ${toza(x.id)}\n`;
  if (x.retry != null) s += `retry: ${toza(x.retry)}\n`;
  if (x.comment != null) s += qatorlar(x.comment, ": ");
  if (x.data != null) {
    const d = typeof x.data === "object"
      ? JSON.stringify(x.data) : x.data;
    s += qatorlar(d, "data: ");
  }
  return s + "\n";
}

// har ulanishga o'z hisoblagichi — Nest'dagidek
function oqimYoz() {
  let oxirgiId = 0;
  return (x) => {
    const faqatIzoh = x.comment != null && x.data === undefined
      && x.type === undefined && x.retry === undefined;
    if (x.id == null && !faqatIzoh) {
      oxirgiId++;
      x = { ...x, id: String(oxirgiId) };
    }
    return formatla(x);
  };
}

const yoz = oqimYoz();
const tana = [
  {
    type: "vazifa.yaratildi", id: "1",
    data: { id: 1, matn: "Non olish" },
  },
  { data: "birinchi qator\nikkinchi qator" },
  { comment: "ping" },
  { retry: 5000, data: "oddiy satr" },
].map(yoz).join("");
console.log(JSON.stringify("\n" + tana));

Konsolda:

"\nevent: vazifa.yaratildi\nid: 1\ndata: {\"id\":1,\"matn\":\"Non olish\"}\n\nid: 1\ndata: birinchi qator\ndata: ikkinchi qator\n\n: ping\n\nid: 2\nretry: 5000\ndata: oddiy satr\n\n"

Nest chiqishi bilan belgima-belgi bir xil. Uchta qoidaga qarang: toza — type va id dan yangi qatorni olib tashlaydi (aks holda hujumchi type orqali soxta qator qo'sha olardi); qatorlar — ko'p qatorli satrni bo'ladi; oqimYoz — har ulanishning o'z oxirgiId si, faqat izohli hodisaga id berilmaydi. Boshidagi "\n" — Nest sarlavhalardan keyin bitta bo'sh qator yozadi.

7. Express'da qanday edi → Nest'da qanday

Nima yutdik: SSE formatining hamma mayda qoidasi (ko'p qatorli data, izoh, id tozalash, sarlavhalar) freymvork ichida — xato qilish joyi kamaydi. Nima yo'qotdik: oqim RxJS orqali o'tadi — yana bir abstraksiya qatlami; boshqaruv biroz kamaydi (masalan, sukut limiti 4M, qo'lda yozgan hubimizda 256 KB edi). Tezlik va xotira bo'yicha o'lchovni yuqorida aytilgan darsda qilamiz.

8. Hujumchi nigohi

• Ma'lumot sizishi. Hozir oqim hamma vazifa hodisasini istalgan ulanishga yuboradi. Loyihada hali auth yo'q — bitta egasi. Auth qo'shilgach (30-qismning auth darslari), har ulanish faqat o'z vazifalarini ko'rishi shart. Kanonda auth ko'chirilganda aynan shu topildi va tuzatildi: SSE oqimi egasiga cheklandi. Bu xato oddiy GET'dagidan xavfliroq: oqim jim sizadi, logda faqat bitta uzun so'rov ko'rinadi.
• Ulanishlarni tugatish (DoS). Har SSE ulanishi soket, tinglovchi va xotira egallaydi. Hujumchi minglab ulanish ochib, serverni to'ldirishi mumkin. Himoya: ulanish soniga IP yoki foydalanuvchi bo'yicha limit, sekin mijozni uzish (maxPendingDataLength), proxy darajasida timeout.
• Hodisa nomi orqali injection. type foydalanuvchi matnidan yasalsa (masalan, teg nomi), ichidagi \n orqali soxta data: qatori qo'shish mumkin bo'lardi. Nest type va id dan yangi qatorlarni olib tashlaydi (modeldagi toza). Qo'lda formatlaganda buni unutmang.

9. Ko'p uchraydigan xatolar

Ikkinchi qator Controller'lar darsidagi marshrut tartibi tuzog'i — SSE uchun ham amal qiladi.

10. Mashqlar

Mashqlar kurs/mashqlar/30/42-sse/ da.

1-mashq (oson): bo'shliqlarni to'ldiring

• SSE endpoint dekoratori —
• Metod qaytaradigan tur — Observable<>
• Brauzer e'tiborsiz qoldiradigan, ikki nuqta bilan boshlanadigan qator —
• Nest sukut bo'yicha o'qimaydigan mijozga nechta belgigacha kutadi? (M bilan) —
• Uzilgan mijozning tinglovchisi qayerda olib tashlanadi? —

2-mashq (o'rta): heartbeat'ni izohga o'tkazish

Loyihadagi ping { type: "ping", data: "" } — oddiy hodisa: u id oladi va brauzerda ping tinglovchisini uyg'otadi. Uni { comment: "ping" } ga almashtirsak nima o'zgaradi? Darsdagi oqimYoz ning soddalashtirilgan nusxasini yozing va ikki variantda [hodisa, ping, hodisa] ketma-ketligi uchun id larni chiqaring. Ishora: faqat izohli hodisa id olmaydi; match(/^id: \d+$/m) (RegExp) har satrdan id qatorini topadi.

function oqimYoz() {
  let oxirgiId = 0;
  return (x) => {
    const faqatIzoh = x.comment != null && x.data === undefined;
    let s = x.type ? `event: ${x.type}\n` : "";
    if (!faqatIzoh) s += `id: ${++oxirgiId}\n`;
    if (x.comment != null) s += `: ${x.comment}\n`;
    if (x.data != null) s += `data: ${JSON.stringify(x.data)}\n`;
    return s;
  };
}

const hodisa = (id) => ({ type: "vazifa.yaratildi", data: { id } });
const pinglar = [{ type: "ping", data: "" }, { comment: "ping" }];
for (const ping of pinglar) {
  const yoz = oqimYoz();
  const idlar = [hodisa(1), ping, hodisa(2)]
    .map(yoz)
    .flatMap((s) => s.match(/^id: \d+$/m) ?? ["(id yo'q)"]);
  console.log(JSON.stringify(ping), "→", idlar.join(", "));
}

{"type":"ping","data":""} → id: 1, id: 2, id: 3
{"comment":"ping"} → id: 1, (id yo'q), id: 2

3-mashq (qiyin): sekin mijoz limiti

maxPendingDataLength modelini yozing. oqim(limit) obyekt qaytarsin: yoz(data) — mijoz uchun navbatga data.length belgi qo'shadi; navbat limit dan oshsa — bir marta "uzildi" xabarini chiqarib, keyingi yoz larni e'tiborsiz qoldiradi; oqi(n) — mijoz n belgi o'qidi; uzildi — getter. Ikki mijozni 10 ta 1000 belgilik hodisa bilan sinang: tezi har hodisadan keyin 1000 o'qiydi, sekini — 300. Limit 5000. Ishora: getter get uzildi() {}; Math.max(0, …) manfiyga tushmaslik uchun.

// Sekin mijoz: hodisalar navbatda yig'iladi, limitdan oshsa — uzamiz
function oqim(limit) {
  let kutmoqda = 0; // mijoz hali o'qimagan belgilar
  let uzildi = false;
  return {
    yoz(data) {
      if (uzildi) return;
      kutmoqda += data.length;
      if (kutmoqda > limit) {
        uzildi = true;
        console.log(`uzildi: ${kutmoqda} belgi (limit ${limit})`);
      }
    },
    oqi(n) {
      kutmoqda = Math.max(0, kutmoqda - n); // mijoz n belgi o'qidi
    },
    get uzildi() {
      return uzildi;
    },
  };
}

const tez = oqim(5000);
const sekin = oqim(5000);
for (let i = 1; i <= 10; i++) {
  const data = "x".repeat(1000);
  tez.yoz(data);
  tez.oqi(1000); // har hodisani darhol o'qiydi
  sekin.yoz(data);
  sekin.oqi(300); // 1000 dan 300 tasini o'qishga ulguradi
}
console.log(`tez: ${tez.uzildi}, sekin: ${sekin.uzildi}`);

uzildi: 5200 belgi (limit 5000)
tez: false, sekin: true

11. Loyiha qadami: /v1/vazifalar/oqim va commit

11.1 Controller

src/vazifalar/vazifalar.controller.ts — importlar. @nestjs/common importiga Sse qo'shiladi (kanon qatori uzun edi, bo'ldik), yana uchta import:

import {
  All, Body, Controller, Delete, Get, Header, HttpCode, Inject, Param,
  Patch, Post, Put, Query, Req, Res, Sse, UseInterceptors,
} from "@nestjs/common";
import type { MessageEvent } from "@nestjs/common";
import { EventEmitter2 } from "@nestjs/event-emitter";
import { interval, map, merge, Observable } from "rxjs";

Constructor'ga uchinchi parametr:

  constructor(
    private readonly xizmat: VazifalarService,
    @Inject(ILOVA_SOZLAMALARI)
    private readonly sozlamalar: IlovaSozlamalari,
    private readonly hodisalar: EventEmitter2,
  ) {}

Va @Get(":id") dan oldin yangi metod:

  /**
   * Server-Sent Events — vazifa hodisalarining jonli oqimi (frontend
   * boshqa tabda o'zgarganini darhol ko'radi). /:id dan OLDIN.
   * "vazifa.*" (wildcard) — xizmat chiqaradigan hamma hodisa; har
   * 25 s "ping" — proxy (nginx) bo'sh ulanishni uzmasin. Mijoz
   * uzilsa Nest obunani bekor qiladi → teardown tinglovchini olib
   * tashlaydi (xotira oqmaydi). Auth qo'shilgach — faqat o'z
   * vazifalari
   */
  @Sse("oqim")
  oqim(): Observable<MessageEvent> {
    const hodisalar = new Observable<MessageEvent>((obunachi) => {
      const tinglovchi = function (
        this: { event: string | string[] },
        data: unknown,
      ): void {
        obunachi.next({
          type: String(this.event), data: data as object,
        });
      };
      this.hodisalar.on("vazifa.*", tinglovchi);
      return () => {
        this.hodisalar.off("vazifa.*", tinglovchi);
      };
    });
    const ping = interval(25_000).pipe(
      map((): MessageEvent => ({ type: "ping", data: "" })),
    );
    return merge(hodisalar, ping);
  }

Metodni bo'lakma-bo'lak o'qiymiz:

• this.hodisalar.on("vazifa.*", tinglovchi) — oldingi darsdagi EventEmitter2, lekin dekorator emas, qo'lda obuna. Sabab: tinglovchi har ulanish uchun alohida va uzilganda o'chishi kerak. @OnEvent esa ilova umri davomida turadi.
• function (this: { event }) — strelkali emas, oddiy funksiya. eventemitter2 tinglovchini chaqirganda this.event ga hodisa nomini qo'yadi (vazifa.bajarildi). Strelkali funksiyada this tashqi klass bo'lib qolardi va nom yo'qolardi. this: parametri — TypeScript'ga this turini aytish usuli.
• type: String(this.event) — hodisa nomi SSE'da event: bo'ladi. Brauzer addEventListener("vazifa.bajarildi", …) bilan aniq hodisani tinglay oladi.
• Teardown — off aynan o'sha tinglovchi havolasi bilan (oldingi darsdagi off mashqining qoidasi).
• merge(hodisalar, ping) — ikki oqimni bittaga qo'shadi. interval(25_000) har 25 soniyada ping chiqaradi. Ko'p proxy'lar (nginx sukut — 60 s) jim ulanishni uzadi; ping uni "tirik" qiladi. Obuna bekor bo'lganda merge ikkalasini ham bekor qiladi — interval ham to'xtaydi.

11.2 Testlar

test/hodisa.test.ts — importlarga import { jsonYubor } from "./yordamchi.ts";, fayl boshidagi izohga "domen hodisalari → SSE oqimi", va birinchi test sifatida:

test("SSE /v1/vazifalar/oqim: yaratish va bajarish hodisalari "
  + "oqimda (text/event-stream)", async () => {
  const boshqaruv = new AbortController();
  const r = await fetch(`${i.manzil}/v1/vazifalar/oqim`, {
    signal: boshqaruv.signal,
  });
  expect(r.headers.get("content-type")).toMatch(/text\/event-stream/);
  const oquvchi = r.body?.getReader();
  if (oquvchi === undefined) throw new Error("tana yo'q");
  const yangi = (await (await fetch(`${i.manzil}/v1/vazifalar`,
    jsonYubor("POST", { matn: "SSE sinovi #oqim" }))).json()) as {
    id: number;
  };
  await fetch(`${i.manzil}/v1/vazifalar/${String(yangi.id)}`,
    jsonYubor("PATCH", { bajarildi: true }));
  let matn = "";
  const dekoder = new TextDecoder();
  while (!matn.includes("event: vazifa.bajarildi")) {
    const { value, done } = await oquvchi.read();
    if (done) break;
    matn += dekoder.decode(value);
  }
  boshqaruv.abort();
  expect(matn).toMatch(
    /event: vazifa\.yaratildi\nid: \d+\ndata: \{"id":1,"matn":"SSE sinovi #oqim"/,
  );
  expect(matn).toContain("event: vazifa.yangilandi");
  expect(matn).toContain("event: vazifa.bajarildi");
});

Test haqiqiy HTTP orqali ishlaydi: avval oqimga ulanadi, keyin boshqa so'rovlar bilan vazifa yaratib, bajaradi va oqimda uch hodisani kutadi. Bu oldingi ikki darsning ishini ham tekshiradi: hodisalar xizmatda chiqyapti, vazifa.* naqshi ishlayapti. id: \d+ — istalgan raqam: test Nest'ning hisoblagichiga tayanmaydi. Regex qatori 70 belgidan uzun — uni bo'lsak o'qish qiyinlashadi, kanonda ham shunday.

Yana ikki test faylga bittadan qator. test/marshrutlar.test.ts — "/vazifalar/:id": … qatoridan keyin:

    "/vazifalar/oqim": "GET, HEAD",

test/hujjat.test.ts — OpenAPI yo'llari ro'yxatiga "/v1/vazifalar/oqim" ("/v1/vazifalar" va "/v1/vazifalar/toplam" orasiga, alifbo tartibida). Swagger @Sse ni oddiy GET deb hujjatlaydi; /health/* esa versiyasiz bo'lgani uchun hujjatga kirmaydi.

11.3 Jonli tekshiruv

Ikki terminalda: birida oqim, ikkinchisida o'zgarishlar.

npm run build
SAQLASH=xotira PORT=4596 npm start
# 1-terminal
curl -N http://127.0.0.1:4596/v1/vazifalar/oqim
# 2-terminal
curl -X POST -H "Content-Type: application/json" \
  -d '{"matn":"Non olish #uy"}' http://127.0.0.1:4596/v1/vazifalar
curl -X PATCH -H "Content-Type: application/json" \
  -d '{"bajarildi":true}' http://127.0.0.1:4596/v1/vazifalar/1
curl -X DELETE http://127.0.0.1:4596/v1/vazifalar/1

Birinchi terminalda (biz ishga tushirdik; helmet va CORS sarlavhalarini olib tashladik):

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: private, no-cache, no-store, must-revalidate, max-age=0, no-transform
Pragma: no-cache
Expires: 0
X-Accel-Buffering: no
Transfer-Encoding: chunked


event: vazifa.yaratildi
id: 1
data: {"id":1,"matn":"Non olish #uy","bajarildi":false,"teglar":["uy"],"yaratildi":"2026-11-02T06:29:16.441Z","yangilandi":"2026-11-02T06:29:16.441Z","tartib":1,"versiya":1}

event: vazifa.yangilandi
id: 2
data: {"id":1,"matn":"Non olish #uy","bajarildi":true,"teglar":["uy"],"yaratildi":"2026-11-02T06:29:16.441Z","yangilandi":"2026-11-02T06:29:16.538Z","tartib":1,"versiya":2}

event: vazifa.bajarildi
id: 3
data: {"id":1,"matn":"Non olish #uy","bajarildi":true,"teglar":["uy"],"yaratildi":"2026-11-02T06:29:16.441Z","yangilandi":"2026-11-02T06:29:16.538Z","tartib":1,"versiya":2}

event: vazifa.ochirildi
id: 4
data: {"id":1}

Uchta so'rov — to'rtta hodisa. PATCH ikki hodisa berdi: yangilandi (har o'zgarishda) va bajarildi (faqat false → true o'tishida). id: 1…4 — ulanish hisoblagichi, vazifa id'si emas («id tuzog'i» bo'limi). Server logida esa vazifa o'chirildi qatori DELETE so'rovining sorovId si bilan yozildi — sinxron emit shu so'rov kontekstida ishladi.

Health ham ishlaydi: curl http://127.0.0.1:4596/health/ready → {"status":"ok","info":{"ombor":{"ms":1,"status":"up"},…}.

11.4 Tekshiruv va commit

npm run tip
npm run lint
npm test
npm run build

Biz kanon nusxasida (Node 24.21.0, bazasiz):

> tsc --noEmit
> eslint . --max-warnings=0
 Test Files  20 passed | 5 skipped (25)
      Tests  80 passed | 6 skipped (86)
> nest build

To'rt darsda testlar 77 tadan 80 taga chiqdi: @Interval, salomatlik, SSE. Kanon tekshiruvi esa Postgres va Redis bilan (Docker'da, PostgreSQL 18.6): tip 0, lint 0, Vitest 123 o'tdi, 1 skip, build, Testcontainers 10/10, jonli API 9 so'rov × 3 saqlash rejimi. Prod bog'liqliklar 19 tadan 22 taga chiqdi: @nestjs/schedule 12.0.2, @nestjs/event-emitter 12.0.1, @nestjs/terminus 12.1.0.

Commit (kanon xabari, 70 belgidan uzun bo'lgani uchun bo'lib yig'amiz):

git add -A
M="feat: domen hodisalari (event-emitter) — webhook tinglovchiga"
M+=" ko'chdi, SSE /v1/vazifalar/oqim, /health/live va /health/ready"
M+=" (terminus), eksport tozalash @Interval"
git commit -m "$M"

12. Real ishda

• Qayerda uchraydi: bildirishnomalar soni, buyurtma holati ("kuryer yo'lda"), AI javobini so'zma-so'z ko'rsatish (ChatGPT kabi interfeyslar aynan SSE bilan oqim qiladi), admin paneldagi jonli statistika.
• AI streaming: LLM javobi bo'laklarini @Sse + Observable bilan yuborish — Nest'da eng qisqa yo'l. 31-qismdagi AI integratsiyasida shu naqsh kerak bo'ladi.
• Masshtab: ikki va undan ko'p server nusxasida in-process hodisa yetmaydi — Redis Pub/Sub yoki Streams kerak (Realtime'ni masshtablash).
• Intervyu savollari: "SSE va WebSocket farqi?", "Nest'da SSE endpoint qanday yoziladi?", "Mijoz uzilganda resurslar qanday tozalanadi?", "SSE'da qayta ulanish qanday ishlaydi?".

Xulosa

• @Sse("yo'l") + Observable<MessageEvent> — Nest SSE; sarlavhalar va format freymvork ishi.
• MessageEvent: data (obyekt → JSON, satr → har qatori alohida), type, id, retry, comment.
• Uzilishda Nest obunani bekor qiladi — teardown'da tinglovchini olib tashlang; @SseSignal() — tayyorgarlik bosqichi uchun.
• Oqim o'rtasidagi xato event: error bo'lib ketadi; sekin mijoz maxPendingDataLength (4M) da uziladi.
• id ni bermasangiz — har ulanishda 1 dan; Last-Event-ID ni Nest o'zi qo'llamaydi.
• Loyihada /v1/vazifalar/oqim — vazifa.* hodisalari + 25 s ping; to'rt darslik bosqich commit qilindi.

Keyingi dars: GraphQL code-first — TypeScript klasslaridan GraphQL sxemasini yasaymiz: @ObjectType, @Resolver, @Query, @Mutation va Apollo drayveri.

Manbalar

• NestJS hujjati — Server-Sent Events: https://docs.nestjs.com/techniques/server-sent-events
• HTML Living Standard — Server-sent events (format, Last-Event-ID): https://html.spec.whatwg.org/multipage/server-sent-events.html
• RxJS — Observable, merge, interval: https://rxjs.dev/guide/observable
• @nestjs/core manba kodi — sse-stream.js, router-response-controller.js (12.2.0): https://github.com/nestjs/nest
• Kanon (IlmHamroh, 2026-10-28): /v1/vazifalar/oqim, hodisa.test (SSE, salomatlik, @Interval), tekshiruv — tip/lint 0, Vitest 123/124, build, Testcontainers 10/10. Demo (/qisqa, /cheksiz, /xatoli, /notogri, sekin mijoz), loyiha nusxasidagi tekshiruv va jonli curl oqimi — biz ishga tushirdik (2026-11-02, Node 24.21.0, NestJS 12.2.0, rxjs 7.8.2).