SSE wrapper — custom Observable с teardown

Паттерн

EventSource (Server-Sent Events), SDK с callback-API, legacy-библиотеки — это не Observable. Чтобы использовать их в RxJS-pipeline, оборачиваем через new Observable(observer => {...}) с явным teardown.

Какую проблему решаем

Если оборачивать «снаружи» — добавлять listener в ngOnInit, удалять в ngOnDestroy — легко забыть один из шагов. Особенно если код переезжает между компонентами. Открытое соединение остаётся жить после ухода со страницы.

Операторы и зачем они нужны

• new Observable(observer => {...}) — конструктор. Внутри открываем resource (EventSource, listener) и вызываем observer.next(...) при событиях.
• Return из конструктора — teardown function. Вызывается при unsubscribe. Здесь закрываем resource.
• observer.next / observer.error / observer.complete — стандартные методы для эмита.

Подводные камни

• Забыли вернуть teardown — соединение останется открытым после unsubscribe. Утечка.
• Не вызывайте observer.next после unsubscribe. Очищайте таймеры в teardown.
• SSE/WebSocket — это не HTTP one-shot. shareReplay и refCount требуют осторожной настройки, чтобы не держать соединение открытым вечно.