Історично в середовищі Node.js розвивалася власна архітектура потоків (node:stream), заснована на системному класі EventEmitter, колбеках та бінарних структурах Buffer. Паралельно консорціум WHATWG стандартизував Web Streams API для браузерів та Edge-середовищ. Сучасний веб-стандарт базується на контрактах Promise, типізованих масивах Uint8Array, ексклюзивних блокуваннях зчитування (Reader Locks) та функціональних конвеєрах pipeThrough.
У сучасному fullstack-стеку (Node.js 18-24+, Next.js 15+, React 19, Cloudflare Workers, Vercel Serverless) Web Streams API став універсальним фундаментом обміну даними. Глобальний метод fetch(), стандартні об'єкти Request і Response, потоковий рендеринг React Server Components (RSC Flight Protocol) та обробники Next.js Route Handlers використовують стандарт WHATWG Streams за замовчуванням.
Розуміння принципів роботи Web Streams та вміння безшовно інтегрувати їх із класичними потоками Node.js (node:stream) є ключовою навичкою для розробки високонавантажених бекендів та сучасних SSR-застосунків.
Специфікація Web Streams API визначає три ключові сутності:
ReadableStream: джерело даних. Керується через внутрішній контролер:
ReadableStreamDefaultController: універсальний контролер, що передає чанки довільних типів.ReadableByteStreamController (BYOB — Bring Your Own Buffer): оптимізований контролер для низькорівневого двійкового введення-виведення. Дозволяє споживачу передавати власний попередньо виділений буфер, виключаючи зайве копіювання пам'яті в V8.WritableStream: абстракція цільового приймача даних, керована контролером WritableStreamDefaultController.TransformStream: пара зі зв'язаного WritableStream (stream.writable) та ReadableStream (stream.readable), призначена для модифікації чи перекодування чанків на льоту.На відміну від потоків Node.js, де на подію 'data' можна зареєструвати необмежену кількість слухачів, Web Streams забезпечують сувору безпеку через блокування:
const reader = readableStream.getReader();
console.log(readableStream.locked); // trueЯк тільки викликано метод getReader(), потік переходить у заблокований стан (locked === true). Жоден інший компонент застосунку не може підключитися до потоку або викликати pipeTo(), доки поточний рідер явно не викличе reader.releaseLock() або не завершить читання. Щоб передати один потік двом незалежним споживачам, його необхідно явно розгалузити методом readableStream.tee().
Керування зворотним тиском у Web Streams API визначається стратегіями черг (Queuing Strategies):
CountQueuingStrategy({ highWaterMark: N }): вимірює наповненість буфера кількістю дискретних елементів.ByteLengthQueuingStrategy({ highWaterMark: N }): розраховує сумарний розмір даних у байтах на основі властивості chunk.byteLength.Коли сумарний розмір елементів у черзі досягає highWaterMark, властивість controller.desiredSize стає меншою або рівною нулю (desiredSize <= 0). Це сигналізує джерелу про необхідність призупинити генерацію нових чанків.
Для усунення бар'єрів між середовищами модуль node:stream містить двосторонні адаптери без накладних витрат на копіювання пам'яті:
Readable.toWeb(nodeReadable): трансформує потік Node.js у стандартний ReadableStream.Readable.fromWeb(webReadableStream): перетворює WHATWG ReadableStream на звичний Readable з node:stream.Writable.toWeb(nodeWritable) та Writable.fromWeb(webWritableStream): аналогічні мости для потоків запису.Конвеєр Web Streams є функціональним ланцюжком із суворим контролем блокувань та наскрізною сигналізацією зворотного тиску:
[ ReadableStream ] (locked: true)
|
| (pipeThrough)
v
+-----------------------+
| TransformStream |
| [writable] -> [readable]
+-----------------------+
|
| (pipeThrough)
v
+------------------------------------+
| CompressionStream('gzip') |
| [writable: raw] -> [readable: gzip]|
+------------------------------------+
|
| (pipeTo)
v
[ WritableStream / HTTP Response ]new Response(readableStream).<Suspense>.CompressionStream('gzip'), що однаково працює в Node.js, Deno та Edge Workers.Генератор потоку даних враховує зворотний тиск через властивість desiredSize:
export function createIncrementalStream(maxItems: number): ReadableStream<Uint8Array> {
let current = 0;
const encoder = new TextEncoder();
return new ReadableStream<Uint8Array>({
start(controller) {
console.log('Потік ініціалізовано');
},
// pull викликається автоматично, коли буфер готовий прийняти нові дані
async pull(controller) {
while ((controller.desiredSize ?? 0) > 0 && current < maxItems) {
current++;
const chunk = encoder.encode(`data: {"id": ${current}, "timestamp": ${Date.now()}}\n\n`);
controller.enqueue(chunk);
if (current % 100 === 0) {
await new Promise((resolve) => setTimeout(resolve, 10));
}
}
if (current >= maxItems) {
controller.close();
}
},
cancel(reason) {
console.warn('Потік скасовано клієнтом:', reason);
},
}, new ByteLengthQueuingStrategy({ highWaterMark: 64 * 1024 }));
}Драйвери баз даних та хмарні SDK часто повертають потоки node:stream. Для їх використання в Route Handlers Next.js застосовується адаптер Readable.toWeb():
import { Readable, Writable } from 'node:stream';
import { createReadStream } from 'node:fs';
// 1. Конвертація потоку читання Node.js у WHATWG ReadableStream
const nodeReadable = createReadStream('large-dataset.csv');
const webReadableStream: ReadableStream<Uint8Array> = Readable.toWeb(nodeReadable);
// 2. Використання Web Stream у циклі читання з ексклюзивним рідером
async function consumeWebStream(stream: ReadableStream<Uint8Array>): Promise<number> {
const reader = stream.getReader();
let totalBytes = 0;
try {
while (true) {
const { value, done } = await reader.read();
if (done) break;
totalBytes += value.byteLength;
}
} finally {
// Обов'язкове звільнення блокування в блоці finally
reader.releaseLock();
}
return totalBytes;
}
// 3. Зворотна конвертація: WHATWG ReadableStream -> Node.js Readable
const convertedBackToNode: Readable = Readable.fromWeb(webReadableStream);Платформенний клас CompressionStream дозволяє створювати високопродуктивні конвеєри стиснення без сторонніх npm-бібліотек:
import { NextRequest, NextResponse } from 'next/server';
import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';
import { join } from 'node:path';
export async function GET(req: NextRequest): Promise<Response> {
const filePath = join(process.cwd(), 'data', 'audit-records.json');
// 1. Отримуємо нативний потік файлової системи Node.js
const fileReadStream = createReadStream(filePath);
// 2. Перетворюємо на WHATWG ReadableStream
const sourceWebStream = Readable.toWeb(fileReadStream);
// 3. Пропускаємо через конвеєр стиснення GZIP
const compressedStream = sourceWebStream.pipeThrough(
new CompressionStream('gzip')
);
// 4. Повертаємо потокову HTTP-відповідь клієнту
return new Response(compressedStream, {
status: 200,
headers: {
'Content-Type': 'application/json',
'Content-Encoding': 'gzip',
'Transfer-Encoding': 'chunked',
'Cache-Control': 'no-cache, no-store',
},
});
}Потік читання файлу fs.createReadStream з розміром чанка 64 КБ.
Огортання потоку Node.js у WHATWG ReadableStream без копіювання пам'яті.
Стиснення CompressionStream('gzip') на льоту по мірі надходження чанків.
Створення веб-стандартного об'єкта Response з потоковими заголовками.
Клієнт декомпресує та обробляє чанки на льоту через протокол HTTP/2.
Ізоморфний модуль для побудови надійних конвеєрів Web Streams з підтримкою Server-Sent Events, компресії та обробки скасування:
export interface StreamPipelineOptions {
compression?: 'gzip' | 'deflate' | 'none';
signal?: AbortSignal;
highWaterMarkBytes?: number;
onChunkSent?: (bytes: number) => void;
}
export class WebStreamPipelineHelper {
private static readonly DEFAULT_HIGH_WATER_MARK = 64 * 1024; // 64 КБ
/**
* Огортає асинхронний генератор у оптимізований ReadableStream.
*/
public static fromAsyncGenerator<T>(
generator: AsyncGenerator<T, void, unknown>,
serializer: (item: T) => Uint8Array,
options: StreamPipelineOptions = {}
): ReadableStream<Uint8Array> {
const highWaterMark = options.highWaterMarkBytes ?? this.DEFAULT_HIGH_WATER_MARK;
const baseStream = new ReadableStream<Uint8Array>(
{
async pull(controller) {
try {
const { value, done } = await generator.next();
if (done) {
controller.close();
return;
}
const binaryChunk = serializer(value);
controller.enqueue(binaryChunk);
if (options.onChunkSent) {
options.onChunkSent(binaryChunk.byteLength);
}
} catch (err) {
controller.error(err);
}
},
cancel(reason) {
console.warn('Pipeline cancelled by consumer:', reason);
},
},
new ByteLengthQueuingStrategy({ highWaterMark })
);
return this.applyCompression(baseStream, options.compression);
}
/**
* Додає потокову компресію до конвеєра.
*/
public static applyCompression(
stream: ReadableStream<Uint8Array>,
compression: 'gzip' | 'deflate' | 'none' = 'none'
): ReadableStream<Uint8Array> {
if (compression === 'none') {
return stream;
}
const compressor = new CompressionStream(compression);
return stream.pipeThrough(compressor);
}
/**
* Створює заголовки HTTP для потокової відповіді.
*/
public static createStreamHeaders(
contentType: string = 'text/event-stream',
compression: 'gzip' | 'deflate' | 'none' = 'none'
): Headers {
const headers = new Headers({
'Content-Type': contentType,
'Cache-Control': 'no-cache, no-transform',
'Connection': 'keep-alive',
'X-Content-Type-Options': 'nosniff',
});
if (compression !== 'none') {
headers.set('Content-Encoding', compression);
}
return headers;
}
}Виклик stream.getReader() в одній частині коду та подальша спроба передати той самий stream у new Response(stream) або викликати stream.pipeThrough(). Середовище видає фатальну помилку TypeError: Cannot get a reader for a locked stream.
Завжди звільняйте блокування через reader.releaseLock() перед повторним використанням, або попередньо розгалужуйте потік методом const [branchA, branchB] = stream.tee().
Коли джерело вичерпується, розробник забуває викликати controller.close(). Клієнт нескінченно очікує завершальних байтів, з'єднання зависає до таймауту, а сервер утримує пам'ять.
Завжди викликайте controller.close() при закінченні даних та передавайте помилки через controller.error(err).
Передача fs.createReadStream() безпосередньо у конструктор new Response(nodeStream) у Next.js або Node.js. Конструктор очікує WHATWG ReadableStream або AsyncIterable. Це спричиняє помилку типізації або передачу пустої відповіді.
Завжди огортайте класичні потоки через Readable.toWeb(nodeStream).
Користувач закриває вкладку браузера, але сервер продовжує обчислювати чанки та виконувати запити до бази даних.
Завжди передавайте req.signal із NextRequest у генератори даних та підписуйтеся на подію abort.
Клієнт надсилає HTTP-запит до Route Handler із заголовком Accept-Encoding: gzip.
Створення ReadableStream із контролем наповненості буфера через desiredSize.
Прокачування байтів через CompressionStream('gzip') за допомогою pipeThrough().
Відправка стиснених чанків клієнту з заголовком Transfer-Encoding: chunked.
Виклик controller.close() після завершення даних із вивільненням системних ресурсів.