Традиционные HTTP API-эндпоинты работают по модели дискретного буферизованного ответа «Request-Response»: сервер собирает всю полезную нагрузку в оперативной памяти (heap Node.js), сериализует её в единый JSON-документ и единовременно отправляет клиенту с заголовком Content-Length. Для сценариев с динамическим обновлением данных (токены больших языковых моделей LLM, биржевые котировки, системная телеметрия, длительный экспорт отчетов) такой подход приводит к трем критическим дефектам:
В Next.js 16 и рантаймах Node.js 22+ / Edge Runtime стандартом потоковой передачи данных является спецификация WHATWG Web Streams API. Route Handlers (app/api/**/route.ts) поддерживают возврат экземпляра стандартного Response, телом которого выступает ReadableStream. В сочетании с легковесным протоколом Server-Sent Events (SSE) это позволяет организовывать однонаправленный асинхронный пуш структурированных событий поверх долговременного HTTP/2-соединения без накладных расходов на протокольное рукопожатие WebSocket.
В потоковой архитектуре соединение рассматривается не как статический снимок данных, а как открытый конвейер с контролируемой пропускной способностью:
+-----------------------------------------------------------------------------------+
| АРХИТЕКТУРА WEB STREAMS & SERVER-SENT EVENTS (SSE) |
+-----------------------------------------------------------------------------------+
| ИСТОЧНИК ДАННЫХ NEXT.JS 16 ROUTE HANDLER КЛИЕНТ (БРАУЗЕР) |
| |
| [Redis Pub/Sub / ] ---> [ ReadableStreamSource ] |
| [LLM Inference / ] | controller.enqueue(chunk) | |
| [Postgres Cursor ] | | |
| | [ TransformStream ] | |
| | (SSE Text Encoder) | |
| +--------------+--------------+ |
| | HTTP/2 Frame |
| Backpressure Signal | "event: metric\n" |
| <-----------------------------+ "data: {...}\n\n" |
| (desiredSize <= 0: пауза) | |
| v |
| [request.signal] <==================================== [ EventSource / fetch() ]|
| (abort event) Клиент закрыл вкладку (getReader().read()) |
| -> чистка таймеров/сокетов |
+-----------------------------------------------------------------------------------+Клиент отправляет GET-запрос. Сервер отвечает HTTP-заголовками 'Content-Type: text/event-stream', 'Cache-Control: no-cache', 'Connection: keep-alive', удерживая сокет открытым.
Next.js Route Handler инициализирует экземпляр ReadableStream. Метод start(controller) подписывается на шину событий или открывает курсор базы данных.
Перед вызовом controller.enqueue() проверяется свойство controller.desiredSize. Если очередь заполнена, источник приостанавливает чтение из базы.
При обрыве связи браузером срабатывает событие abort на request.signal. Сервер отписывается от очередей, очищает интервалы и закрывает стрим без утечек памяти.
Что произойдет, если в ReadableStream в методе pull() или start() безостановочно вызывать controller.enqueue(chunk), игнорируя показатель controller.desiredSize при медленном клиенте?
Контроллер ReadableStreamDefaultController предоставляет геттер desiredSize. Это целочисленное значение указывает, сколько чанков (или байтов в зависимости от стратегии очередей ByteLengthQueuingStrategy) требуется поместить в очередь для заполнения желаемого лимита (highWaterMark):
desiredSize > 0: очередь не заполнена, источник может генерировать новые данные.desiredSize <= 0: очередь заполнена или переполнена (противодавление), источник обязан приостановить генерацию.desiredSize === null: поток закрыт или находится в состоянии ошибки.Реализуем полнофункциональный Route Handler в app/api/telemetry/stream/route.ts с поддержкой SSE-форматирования, валидацией backpressure и корректной очисткой при обрыве соединения клиентом.
import { NextRequest } from 'next/server';
export const runtime = 'nodejs'; // Или 'edge' в зависимости от инфраструктуры
export const dynamic = 'force-dynamic';
interface TelemetryPayload {
cpuUsage: number;
memoryUsageMb: number;
activeRequests: number;
timestamp: number;
}
export async function GET(request: NextRequest) {
const encoder = new TextEncoder();
const abortSignal = request.signal;
// Форматирование сообщения по спецификации SSE
const formatSseEvent = (event: string, data: unknown, id?: string): string => {
let message = '';
if (id) message += `id: ${id}\n`;
message += `event: ${event}\n`;
message += `data: ${JSON.stringify(data)}\n\n`;
return message;
};
const stream = new ReadableStream({
start(controller) {
let messageId = 0;
let timer: NodeJS.Timeout | null = null;
// Обработка принудительного разрыва соединения клиентом
const cleanup = () => {
if (timer) {
clearInterval(timer);
timer = null;
}
try {
controller.close();
} catch {
// Игнорируем ошибку, если контроллер уже закрыт
}
};
if (abortSignal.aborted) {
cleanup();
return;
}
abortSignal.addEventListener('abort', cleanup, { once: true });
// Генератор телеметрии с контролем противодавления
timer = setInterval(() => {
// Проверяем сигнал отмены перед каждым тиком
if (abortSignal.aborted) {
cleanup();
return;
}
// BACKPRESSURE CHECK: Если внутренний буфер переполнен, пропускаем такт
if (controller.desiredSize !== null && controller.desiredSize <= 0) {
console.warn('[SSE] Backpressure detected: client queue is full. Throttling generation.');
return;
}
messageId++;
const payload: TelemetryPayload = {
cpuUsage: Math.round(Math.random() * 100),
memoryUsageMb: Math.round(process.memoryUsage().heapUsed / 1024 / 1024),
activeRequests: Math.floor(Math.random() * 500),
timestamp: Date.now(),
};
const sseFormattedChunk = formatSseEvent('telemetry_tick', payload, String(messageId));
try {
controller.enqueue(encoder.encode(sseFormattedChunk));
} catch (err) {
console.error('[SSE] Failed to enqueue chunk, closing stream:', err);
cleanup();
}
}, 1000);
},
cancel(reason) {
console.log('[SSE] Stream cancelled by consumer:', reason);
}
});
// Возврат стандартного ответа со специальными заголовками
return new Response(stream, {
headers: {
'Content-Type': 'text/event-stream; charset=utf-8',
'Cache-Control': 'no-cache, no-transform',
'Connection': 'keep-alive',
// Критично для Nginx / OpenResty: отключает внутреннюю буферизацию прокси!
'X-Accel-Buffering': 'no',
},
});
}В браузере подписка на поток может осуществляться как через классический EventSource, так и через fetch с ReadableStreamDefaultReader (позволяющий передавать произвольные заголовки авторизации Authorization: Bearer ...):
'use client';
import { useEffect, useState, useCallback } from 'react';
export interface TelemetryData {
cpuUsage: number;
memoryUsageMb: number;
activeRequests: number;
timestamp: number;
}
export function useServerTelemetry() {
const [data, setData] = useState<TelemetryData | null>(null);
const [isConnected, setIsConnected] = useState(false);
const [error, setError] = useState<string | null>(null);
const connectStream = useCallback(() => {
const controller = new AbortController();
async function startFetchStream() {
try {
setIsConnected(true);
setError(null);
const response = await fetch('/api/telemetry/stream', {
signal: controller.signal,
headers: {
Accept: 'text/event-stream',
},
});
if (!response.ok || !response.body) {
throw new Error(`HTTP Stream failed: ${response.status}`);
}
const reader = response.body.getReader();
const decoder = new TextDecoder('utf-8');
let buffer = '';
while (true) {
const { value, done } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const parts = buffer.split('\n\n');
buffer = parts.pop() ?? ''; // Сохраняем незавершенный кусок
for (const rawMessage of parts) {
const lines = rawMessage.split('\n');
let eventType = 'message';
let rawData = '';
for (const line of lines) {
if (line.startsWith('event:')) {
eventType = line.replace('event:', '').trim();
} else if (line.startsWith('data:')) {
rawData = line.replace('data:', '').trim();
}
}
if (eventType === 'telemetry_tick' && rawData) {
const parsed: TelemetryData = JSON.parse(rawData);
setData(parsed);
}
}
}
} catch (err: unknown) {
if ((err as Error).name !== 'AbortError') {
console.error('[SSE Client Error]:', err);
setError((err as Error).message);
setIsConnected(false);
}
}
}
startFetchStream();
return () => {
// При размонтировании компонента прерываем чтение, посылая сигнал abort на сервер
controller.abort();
setIsConnected(false);
};
}, []);
useEffect(() => {
const cleanup = connectStream();
return cleanup;
}, [connectStream]);
return { data, isConnected, error };
}Для очистки и модификации потока на лету (например, фильтрации персональных данных или сжатия) используется связка пайплайнов ReadableStream.pipeThrough(new TransformStream()):
Асинхронный генератор читает записи из таблицы событий БД Postgres (Logical Replication / Listen-Notify).
TransformStream нормализует схемы, удаляет конфиденциальные атрибуты и проверяет права доступа текущей сессии.
Преобразование структурированных объектов в UTF-8 байты вида 'event: update\ndata: {...}\n\n'.
Передача чанков клиенту с контролем размера буфера окна TCP (Flow Control).
Почему в потоковом Route Handler критически важно передавать request.signal в асинхронные источники и подписываться на событие 'abort'?
Если ваше Next.js приложение развернуто в Docker за обратным прокси Nginx, Nginx по умолчанию буферизует входящие HTTP-ответы вплоть до достижения 64-128 Кб, прежде чем отправить их клиенту. В результате SSE-поток перестает быть потоковым: пользователь не видит никаких обновлений несколько минут, а затем получает огромную пачку сообщений разом. Всегда выставляйте X-Accel-Buffering: no в заголовках ответа Route Handler.
ReadableStream ожидает передачу бинарных чанков типа Uint8Array. Если вызвать controller.enqueue("event: ping\n\n") напрямую со строкой без энкодинга (new TextEncoder().encode(...)), в рантайме Node.js возникнет исключение TypeError [ERR_INVALID_ARG_TYPE]: The "chunk" argument must be of type string or an instance of Buffer or Uint8Array.
Создание таймера внутри new ReadableStream({ start(controller) { setInterval(...) } }) без очистки в request.signal.addEventListener('abort', ...) и в коллбэке cancel() оставляет активный таймер в Event Loop навсегда. На высоконагруженном сервисе 10 000 открытых и закрытых вкладок приведут к 10 000 "зомби-таймеров", потребляющих 100% CPU.
Если забыть явно объявить export const dynamic = 'force-dynamic' в потоковом Route Handler, компилятор Next.js во время сборки (next build) может попытаться статически предварительно отрендерить маршрут или кэшировать его результат, что приведет к фатальной ошибке сборки или полной остановке динамического потока.