Настоящие потоки: fs, http, Web Streams и async-итерация

Здесь будет встроен видео-разбор темы (5–7 минут, YouTube / RuTube / VK Видео) — тем, кто лучше воспринимает на слух, текст ниже не обязателен.

Свой класс потока в Node писать почти не придётся: готовые раздают библиотеки. createReadStream и createWriteStream из node:fs дают поток файла, node:http — поток запроса, node:net — сокет, у процесса есть process.stdin и process.stdout, а node:zlib и node:crypto отдают потоки, которые одновременно читают и пишут: сжатие и хеш. Все они — те самые Readable и Writable, что собирала статья про читающий и пишущий поток: те же события data, end, error, close, finish и drain, тот же порог, тот же ответ write. Новое — не устройство, а содержимое.

В куске настоящего потока лежат байты: Buffer, а не строка. Порог у байтовых потоков считается в байтах — по умолчанию 64 КиБ и у файла, и у сокета (в объектном режиме — 16 штук). Отсюда главное следствие, на котором спотыкается каждый второй разбор файла: границу между кусками кладёт порог, и ни строка, ни даже отдельный символ его не интересуют. Русская буква в UTF-8 занимает два байта, и кусок кончается ровно там, где насчитал байты, — хоть посреди буквы.

Прогон читает один файл в три строки при пороге в 16 байт и собирает текст обратно четырьмя способами: склейкой байтов, склейкой строк по куску, потоком с собственным декодером и через readline. Прогноз: сколько будет кусков, совпадёт ли склейка с файлом в первых двух случаях и сколько строк отдаст readline?

прогон node 26.5.0 · 16 сентября 2026

Один файл при пороге в 16 байт: куски и четыре способа собрать из них текст

import { createReadStream } from "node:fs";
import { readFile } from "node:fs/promises";
import { createInterface } from "node:readline";

const FILE = "дневник.txt";
const text = await readFile(FILE, "utf8");
console.log(`файл: ${Buffer.byteLength(text)} байт, ${text.length} символов, ${text.trimEnd().split("\n").length} строки`);

// порог 16 байт вместо 64 КиБ по умолчанию: так куски видны на файле в три строки
const chunks = [];
for await (const chunk of createReadStream(FILE, { highWaterMark: 16 })) chunks.push(chunk);
console.log(`кусков: ${chunks.length}, тип куска: ${chunks[0].constructor.name}, длины: ${chunks.map((c) => c.length).join(" ")}`);
console.log(`первый кусок: «${chunks[0].toString()}»`);
console.log(`склеены как байты — текст цел: ${Buffer.concat(chunks).toString() === text}`);

// каждый кусок в строку по отдельности
const naive = chunks.map((c) => c.toString()).join("");
console.log(`каждый кусок в строку — текст цел: ${naive === text}`);
console.log(`наивная первая строка: «${naive.split("\n")[0]}»`);

// у потока с encoding декодер свой: хвост недочитанного символа ждёт следующего куска
const parts = [];
for await (const part of createReadStream(FILE, { highWaterMark: 16, encoding: "utf8" })) parts.push(part);
console.log(`encoding utf8: частей ${parts.length}, длины ${parts.map((p) => p.length).join(" ")}, текст цел: ${parts.join("") === text}`);

const lines = [];
for await (const line of createInterface({ input: createReadStream(FILE, { highWaterMark: 16 }), crlfDelay: Infinity })) {
  lines.push(line);
}
console.log(`readline: строк ${lines.length}`);
for (const line of lines) console.log(`  ${line}`);

const whole = [];
for await (const chunk of createReadStream(FILE)) whole.push(chunk);
console.log(`без highWaterMark: кусков ${whole.length}`);

node read.mjs

Сначала предположите, что выведет команда

вывод команды · код выхода 0

файл: 253 байт, 152 символов, 3 строки
кусков: 16, тип куска: Buffer, длины: 16 16 16 16 16 16 16 16 16 16 16 16 16 16 16 13
первый кусок: «14 сентяб�»
склеены как байты — текст цел: true
каждый кусок в строку — текст цел: false
наивная первая строка: «14 сентяб��я: поста��ил Node 26 и запусти�� первый ��ервер»
encoding utf8: частей 16, длины 9 9 13 8 9 10 9 9 12 9 12 9 8 9 10 7, текст цел: true
readline: строк 3
  14 сентября: поставил Node 26 и запустил первый сервер
  15 сентября: понял, зачем write отвечает false
  16 сентября: файл приходит кусками, а не строками
без highWaterMark: кусков 1

Кусков шестнадцать, строк три — числа не связаны никак. Первый кусок оборвался на «14 сентяб» и первом байте буквы «р»: toString на нём одном показывает ромбик с вопросом — замещающий символ, которым декодер сообщает, что байты не складываются в букву. Склейка Buffer.concat цела: она соединяет байты и декодирует один раз в конце. Наивная склейка — нет: каждый кусок декодируется отдельно, и половинка буквы пропадает на каждой границе.

Лечится это не руками. Опция encoding (или setEncoding("utf8") позже) включает у потока свой декодер: недочитанный хвост символа он придерживает до следующего куска и отдаёт уже строки — потому длины в символах и вышли неровными. createInterface из node:readline идёт дальше и собирает из кусков строки — тот самый приём, которым первая программа на Node читала stdin. А без явного порога файл в 253 байта приехал одним куском: 64 КиБ хватило на всё.

Собрать такой декодер самому придётся ровно один раз — когда куски приходят не из файла, а из сокета, и склеивать их надо своими руками. Этим займётся статья про тело запроса: там же выяснится, почему длину тела нельзя мерить длиной строки.

У потока есть второй режим — объектный: new Readable({ objectMode: true }). В нём кусок — любое значение: строка, объект, запись из базы. Порог тогда считается не в байтах, а в штуках, и по умолчанию их 16. В объектном режиме работают разборщики: readline отдаёт строки, парсер CSV — записи, а Readable.from делает поток из массива или генератора. Наша миниатюра к этому режиму близка, но не равна ему: кусками в ней были строки, как в объектном, а порог она считала не в штуках, а в символах — по chunk.length.

253 байта в 152 символах: русский текст в UTF-8 весит примерно вдвое больше, чем кажется по количеству букв.

Слушатель data ничего не знает про промисы. Он функция: его позвали, он вернул значение — поток пошёл дальше. Если внутри стоит await, слушатель возвращает промис, поток его выбрасывает и отдаёт следующий кусок. Получается код, который выглядит правильным и молча теряет работу: stream.on("data", async (chunk) => { await save(chunk); }) вычитывает весь файл в память и запускает столько сохранений сразу, сколько кусков было.

Замена — тот самый for await, которым первая программа на Node перебирала строки stdin: следующий кусок он ждёт, как await ждёт промис. Прогон сравнивает оба способа на одном файле в десять кусков, а потом показывает три особенности цикла. Прогноз: сколько кусков успеет сохраниться к событию end у слушателя и что станет с потоком, из цикла по которому вышли через break?

прогон node 26.5.0 · 16 сентября 2026

Один файл в десять кусков — слушателем и циклом, а потом ошибка, break и destroyOnReturn

import { once } from "node:events";
import { createReadStream } from "node:fs";
import { rm, writeFile } from "node:fs/promises";

const FILE = "данные.bin";
// 1000 байт при пороге 100: ровно десять кусков, считать нечего
await writeFile(FILE, "x".repeat(1000));
const open = () => createReadStream(FILE, { highWaterMark: 100 });
const wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// обработчики держит не таймер, а обещание: иначе счёт зависел бы от часов машины
let release;
const gate = new Promise((resolve) => { release = resolve; });
let arrived = 0;
let handled = 0;
const listened = open();
listened.on("data", async () => {
  arrived += 1;
  await gate;
  handled += 1;
});
await once(listened, "end");
console.log(`data: к концу потока приехало кусков ${arrived}, обработано ${handled}`);
release();

let seen = 0;
let handledInLoop = 0;
const steps = [];
for await (const chunk of open()) {
  seen += 1;
  if (steps.length < 3) steps.push(`${seen}/${handledInLoop}`);
  await wait(5);
  handledInLoop += 1;
}
console.log(`for await: прочитано/обработано на первых кусках — ${steps.join(" ")}, в конце ${seen}/${handledInLoop}`);

try {
  for await (const chunk of createReadStream("нет-такого-файла")) console.log(chunk.length);
} catch (err) {
  console.log(`ошибка потока пришла в catch: ${err.code}`);
}

const stopped = open();
for await (const chunk of stopped) break;
console.log(`после break: destroyed ${stopped.destroyed}, дочитан до конца ${stopped.readableEnded}`);
try {
  for await (const chunk of stopped) console.log(chunk.length);
} catch (err) {
  console.log(`второй проход по тому же потоку: ${err.code}`);
}

const kept = open();
for await (const chunk of kept.iterator({ destroyOnReturn: false })) break;
console.log(`iterator({ destroyOnReturn: false }): destroyed ${kept.destroyed}`);
let rest = 0;
for await (const chunk of kept) rest += 1;
console.log(`и дочитали ещё кусков: ${rest}`);

await rm(FILE);

node loop.mjs

Сначала предположите, что выведет команда

вывод команды · код выхода 0

data: к концу потока приехало кусков 10, обработано 0
for await: прочитано/обработано на первых кусках — 1/0 2/1 3/2, в конце 10/10
ошибка потока пришла в catch: ENOENT
после break: destroyed true, дочитан до конца false
второй проход по тому же потоку: ABORT_ERR
iterator({ destroyOnReturn: false }): destroyed false
и дочитали ещё кусков: 9

У слушателя к концу потока не сохранено ничего: десять кусков приехали, десять промисов повисли в воздухе. У цикла на каждом шаге прочитано ровно на один кусок больше, чем обработано, — следующий читается только после тела. Ошибка потока пришла исключением в catch, а не событием error: это вторая причина предпочитать цикл. Третья — уборка: break уничтожил поток, и второй проход по нему упал с ABORT_ERR. Если поток нужен дальше, цикл берут у метода iterator({ destroyOnReturn: false }) — тогда после выхода поток жив и дочитывается.

Магии в цикле нет: for await умеет работать с любым объектом, у которого есть метод под системным символом Symbol.asyncIterator, возвращающий объект с методом next, а тот отдаёт промис с { value, done }. Значит, наша миниатюра тоже это умеет — нужна очередь кусков и обещание, которое ждёт следующего. Заодно станет видно, откуда у цикла берётся обратное давление: забрали кусок — сняли паузу, получили следующий — снова поставили. Прогноз: в каком порядке встанут строки диска и обработки и что напечатает последняя строка?

js

// Emitter, потоки, pipe и pipeline — в подложке
function chunks(stream) {
  const queue = [];
  let ended = false;
  let waiting = null;
  // отдать кусок тому, кто его ждёт, — или сказать, что кусков больше не будет
  const answer = () => {
    if (!waiting) return;
    if (queue.length > 0) waiting({ value: queue.shift(), done: false });
    else if (ended) waiting({ value: undefined, done: true });
    else return;
    waiting = null;
  };
  stream.on("data", (chunk) => {
    queue.push(chunk);
    // кусок уже у нас: пока его не забрали, поток может помолчать
    stream.pause();
    answer();
  });
  stream.on("end", () => {
    ended = true;
    answer();
  });
  return {
    [Symbol.asyncIterator]() {
      return this;
    },
    next() {
      if (queue.length === 0 && !ended) stream.resume();
      return new Promise((resolve) => {
        waiting = resolve;
        answer();
      });
    },
  };
}

let read = 0;
const file = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    console.log(`диск: читаю кусок ${read}`);
    file.push(read < 5 ? `[кусок ${read}]` : null);
  },
});
file.on("close", () => console.log("поток закрыт"));

async function main() {
  for await (const chunk of chunks(file)) {
    console.log(`обработка: ${chunk}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
    if (chunk === "[кусок 3]") break;
  }
  console.log(`вышли из цикла: уничтожен ${file.destroyed}, в буфере кусков ${file.buffer.length}`);
}
main();

// Emitter, потоки, pipe и pipeline — в подложке
function chunks(stream) {
  const queue = [];
  let ended = false;
  let waiting = null;
  // отдать кусок тому, кто его ждёт, — или сказать, что кусков больше не будет
  const answer = () => {
    if (!waiting) return;
    if (queue.length > 0) waiting({ value: queue.shift(), done: false });
    else if (ended) waiting({ value: undefined, done: true });
    else return;
    waiting = null;
  };
  stream.on("data", (chunk) => {
    queue.push(chunk);
    // кусок уже у нас: пока его не забрали, поток может помолчать
    stream.pause();
    answer();
  });
  stream.on("end", () => {
    ended = true;
    answer();
  });
  return {
    [Symbol.asyncIterator]() {
      return this;
    },
    next() {
      if (queue.length === 0 && !ended) stream.resume();
      return new Promise((resolve) => {
        waiting = resolve;
        answer();
      });
    },
  };
}

async function main() {
  for await (const chunk of chunks(file)) {
    console.log(`обработка: ${chunk}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
    if (chunk === "[кусок 3]") break;
  }
  console.log(`вышли из цикла: уничтожен ${file.destroyed}, в буфере кусков ${file.buffer.length}`);
}
main();

страница примера

Сначала предположите, что выведет код

Строки чередуются: диск читает кусок только тогда, когда предыдущий забрали. Это и есть то самое чтение по спросу, только записанное не через drain, а через паузу, — и притом даром: следующий next не позовут, пока тело цикла не кончится. Хвост же выдаёт недоделку. Диск успел прочитать кусок 4 — поток читает на один вперёд, это нормально. А вот после break поток остался жив: «уничтожен false», прочитанный кусок лежит в буфере, файл не закрыт. Цикл кончился, а ресурс — нет.

Выход из цикла — не только break, но и return, и исключение — зовёт у итератора метод return, если он есть. Там и место уборке. Прогноз: чем вывод теперь будет отличаться от прошлого?

js

// Emitter, потоки, pipe и pipeline — в подложке
function chunks(stream) {
  const queue = [];
  let ended = false;
  let waiting = null;
  // отдать кусок тому, кто его ждёт, — или сказать, что кусков больше не будет
  const answer = () => {
    if (!waiting) return;
    if (queue.length > 0) waiting({ value: queue.shift(), done: false });
    else if (ended) waiting({ value: undefined, done: true });
    else return;
    waiting = null;
  };
  stream.on("data", (chunk) => {
    queue.push(chunk);
    // кусок уже у нас: пока его не забрали, поток может помолчать
    stream.pause();
    answer();
  });
  stream.on("end", () => {
    ended = true;
    answer();
  });
  return {
    [Symbol.asyncIterator]() {
      return this;
    },
    next() {
      if (queue.length === 0 && !ended) stream.resume();
      return new Promise((resolve) => {
        waiting = resolve;
        answer();
      });
    },
    // выход из цикла: break, return и исключение зовут именно этот метод
    return() {
      stream.destroy();
      return Promise.resolve({ value: undefined, done: true });
    },
  };
}

let read = 0;
const file = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    console.log(`диск: читаю кусок ${read}`);
    file.push(read < 5 ? `[кусок ${read}]` : null);
  },
});
file.on("close", () => console.log("поток закрыт"));

async function main() {
  for await (const chunk of chunks(file)) {
    console.log(`обработка: ${chunk}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
    if (chunk === "[кусок 3]") break;
  }
  console.log(`вышли из цикла: уничтожен ${file.destroyed}, в буфере кусков ${file.buffer.length}`);
}
main();

// Emitter, потоки, pipe и pipeline — в подложке
function chunks(stream) {
  const queue = [];
  let ended = false;
  let waiting = null;
  // отдать кусок тому, кто его ждёт, — или сказать, что кусков больше не будет
  const answer = () => {
    if (!waiting) return;
    if (queue.length > 0) waiting({ value: queue.shift(), done: false });
    else if (ended) waiting({ value: undefined, done: true });
    else return;
    waiting = null;
  };
  stream.on("data", (chunk) => {
    queue.push(chunk);
    // кусок уже у нас: пока его не забрали, поток может помолчать
    stream.pause();
    answer();
  });
  stream.on("end", () => {
    ended = true;
    answer();
  });
  return {
    [Symbol.asyncIterator]() {
      return this;
    },
    next() {
      if (queue.length === 0 && !ended) stream.resume();
      return new Promise((resolve) => {
        waiting = resolve;
        answer();
      });
    },
    // выход из цикла: break, return и исключение зовут именно этот метод
    return() {
      stream.destroy();
      return Promise.resolve({ value: undefined, done: true });
    },
  };
}

async function main() {
  for await (const chunk of chunks(file)) {
    console.log(`обработка: ${chunk}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
    if (chunk === "[кусок 3]") break;
  }
  console.log(`вышли из цикла: уничтожен ${file.destroyed}, в буфере кусков ${file.buffer.length}`);
}
main();

страница примера

Сначала предположите, что выведет код

Теперь на выходе из цикла приходит close, буфер пуст, ресурс отпущен. В Node ровно это и включено по умолчанию — опция destroyOnReturn у итератора потока, которую прогон выключал руками. Отсюда и правило: for await по потоку либо дочитывают до конца, либо выходят из него сознательно, понимая, что поток после этого мёртв.

Слушатель data с async внутри — не редкая экзотика, а самая частая ошибка в чужом коде на потоках: она не падает, а тихо съедает память.

Вторая библиотека потоков пришла с другой стороны. Когда браузеру понадобилось отдавать тело ответа по частям, комитет WHATWG описал свой стандарт — Web Streams, и тело ответа fetch, которое читается один раз, устроено именно так. Потом эти же классы появились в Node (глобально, без импорта), в Deno, в Bun и в воркерах на краю сети. Слова там другие, механизм тот же: push стал enqueue, push(null) — close, read — pull, destroy — cancel. Событий нет вовсе: всё, что в Node сообщают события, здесь возвращают промисы.

Пример — тот же, что был на миниатюре: диск отдаёт пять кусков, потребитель уходит на третьем. Прогноз: сколько раз позовут pull до первой обработки, что напечатается при выходе из цикла и сколько кусков достанется второму проходу?

js

let read = 0;
const file = new ReadableStream(
  {
    start() {
      console.log("start: поток открыт");
    },
    // pull зовут, когда в очереди есть место, — это и есть «чтение по спросу»
    pull(controller) {
      read += 1;
      console.log(`диск: читаю кусок ${read}`);
      if (read < 5) controller.enqueue(`[кусок ${read}]`);
      else controller.close();
    },
    cancel(reason) {
      console.log(`cancel: ${reason}`);
    },
  },
  { highWaterMark: 1 },
);

async function main() {
  for await (const chunk of file) {
    console.log(`обработка: ${chunk}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
    if (chunk === "[кусок 3]") break;
  }
  const again = [];
  for await (const chunk of file) again.push(chunk);
  console.log(`второй проход дал кусков: ${again.length}`);
}
main();

Сначала предположите, что выведет код

start зовут один раз при создании, pull — каждый раз, когда в очереди появляется место. Порог здесь считается в штуках, и по умолчанию это одна штука: поток держит один кусок про запас, поэтому до первой обработки pull позвали дважды — один кусок в очередь, второй в ответ на ожидающее чтение. Дальше видно то же чередование, что и на миниатюре. break дошёл до источника сам: cancel напечатался без всякого метода return — в Web-потоках выход из цикла отменяет источник по стандарту, а причину передают аргументом (её не передали — отсюда undefined).

Последняя строка — важное отличие от Node. Отменённый Web-поток не бросается ошибками: второй проход просто не дал ни одного куска, тихо и сразу. Node в той же ситуации кидает ABORT_ERR. Кому что нравится, но знать разницу нужно: молчание легко принять за пустой ответ сервера.

Дальше словарь пригодится целиком.

	Миниатюра и `node:stream`	Web Streams
источник отдаёт кусок	`push(chunk)`, ответ `false` — «пока хватит»	`controller.enqueue(chunk)`, место — `controller.desiredSize`
источник читает по спросу	метод `read` в опциях зовёт сам поток	`pull(controller)`, можно вернуть промис — тогда следующего `pull` не будет до его выполнения
конец	`push(null)` и событие `end`	`controller.close()`
ошибка	событие `error`	`controller.error(err)` — промис читателя отклоняется
потребитель читает	слушатель `data` или `for await`	`reader.read()` с ответом `{ value, done }` или `for await`
потребитель уходит	`destroy()` и событие `close`	`reader.cancel()` и колбэк `cancel`; после отмены чтение молча даёт `done`
запись с оглядкой	`write()` вернул `false` — ждать `drain`	`await writer.ready` перед записью, место — `writer.desiredSize`
порог	байты, по умолчанию 64 КиБ (в объектном режиме 16 штук)	штуки, по умолчанию одна; байты — стратегией `ByteLengthQueuingStrategy`
соединить потоки	`pipeline(источник, приёмник, колбэк)`	`источник.pipeThrough(преобразование).pipeTo(приёмник)`: `pipeThrough` отдаёт поток для следующего звена, `pipeTo` — промис конца
преобразование посередине	`Transform`, например `createGzip()`	`TransformStream`, готовые `TextDecoderStream` и `CompressionStream`

Одного в этой таблице нет — записи в файл. Web Streams умеют читать откуда угодно, но источников и приёмников в браузере им даёт платформа: тело ответа, тело запроса, ответ Service Worker. Открыть файл на диске они не могут: это дело node:fs, а значит — node:stream.

Оба мира встречаются в одном запросе. На сервере node:http даёт запрос и ответ, на клиенте fetch даёт Response — и ни один из них не спрашивает, какая библиотека потоков вам нравится. Прогон поднимает сервер на свободном порту, шлёт ему запрос из того же процесса и печатает, чем на самом деле оказались req и res, а потом переводит поток из мира в мир. Прогноз: наследуется ли res от Writable, что лежит в куске тела ответа у клиента и сколько кусков он получит?

прогон node 26.5.0 · 16 сентября 2026

Один сервер и два запроса: тело запроса циклом, тело ответа Web-потоком у клиента и мост в обе стороны

import { createReadStream } from "node:fs";
import { createServer } from "node:http";
import { Readable, Writable } from "node:stream";

const chain = (o) => {
  const names = [];
  while ((o = Object.getPrototypeOf(o))) names.push(o.constructor.name);
  return names.join(" < ");
};

let first = true;
const server = createServer(async (req, res) => {
  if (first) {
    first = false;
    console.log(`сервер: req — ${chain(req)}`);
    console.log(`сервер: res — ${chain(res)}, это Writable: ${res instanceof Writable}`);
  }
  let body = "";
  for await (const chunk of req) body += chunk;
  console.log(`сервер: тело запроса «${body}»`);
  res.writeHead(200, { "content-type": "text/plain; charset=utf-8" });
  res.write("ответ идёт ");
  // пауза, чтобы вторая половина ушла отдельным куском: так виден поток, а не готовая строка
  setTimeout(() => res.end("двумя кусками"), 20);
});

server.listen(0, "127.0.0.1", async () => {
  const base = `http://127.0.0.1:${server.address().port}`;
  console.log(`клиент: POST ${base}/`);

  const answer = await fetch(`${base}/`, { method: "POST", body: "привет" });
  console.log(`клиент: ${answer.status}, тело — ${answer.body.constructor.name}, слушатели: ${typeof answer.body.on}`);
  const pieces = [];
  for await (const chunk of answer.body) pieces.push(`${chunk.constructor.name}(${chunk.length})`);
  console.log(`клиент: куски ответа — ${pieces.join(", ")}`);

  const second = await fetch(`${base}/`, { method: "POST", body: "ещё раз" });
  const asNode = Readable.fromWeb(second.body);
  console.log(`fromWeb: ${chain(asNode)}, это Readable: ${asNode instanceof Readable}`);
  let text = "";
  asNode.on("data", (chunk) => { text += chunk; });
  await new Promise((resolve) => asNode.on("end", resolve));
  console.log(`fromWeb: слушателем data прочитано «${text}»`);

  const web = Readable.toWeb(createReadStream("письмо.txt"));
  console.log(`toWeb: ${web.constructor.name}, отмена — ${typeof web.cancel}, уничтожение — ${typeof web.destroy}`);
  const reader = web.getReader();
  const piece = await reader.read();
  console.log(`toWeb: read() дал { done: ${piece.done}, value: ${piece.value.constructor.name}(${piece.value.length}) }`);
  await reader.cancel();
  console.log(`toWeb: после cancel() read() дал { done: ${(await reader.read()).done} }`);

  server.close(() => console.log("сервер закрыт"));
});

node bridge.mjs

Сначала предположите, что выведет команда

вывод команды · код выхода 0

клиент: POST http://127.0.0.1:<порт>/
сервер: req — IncomingMessage < Readable < Stream < EventEmitter < Object
сервер: res — ServerResponse < OutgoingMessage < Stream < EventEmitter < Object, это Writable: false
сервер: тело запроса «привет»
клиент: 200, тело — ReadableStream, слушатели: undefined
клиент: куски ответа — Uint8Array(20), Uint8Array(25)
сервер: тело запроса «ещё раз»
fromWeb: Readable < Stream < EventEmitter < Object, это Readable: true
fromWeb: слушателем data прочитано «ответ идёт двумя кусками»
toWeb: ReadableStream, отмена — function, уничтожение — undefined
toWeb: read() дал { done: false, value: Uint8Array(92) }
toWeb: после cancel() read() дал { done: true }
сервер закрыт

Запрос — настоящий Readable, и его лестница наследования кончается тем же, чем миниатюра: IncomingMessage < Readable < Stream < EventEmitter — читающий поток поверх излучателя событий. А вот res от Writable не наследуется: ServerResponse < OutgoingMessage < Stream < EventEmitter. У него есть write, end, ответ false и событие drain — договор соблюдён, класс другой: OutgoingMessage старше, чем нынешний Writable. Мораль практическая: проверять потоки через instanceof — плохая идея, поток узнают по договору.

У клиента то же тело выглядит иначе: response.body — это ReadableStream, никаких on у него нет, а в куске лежит Uint8Array, а не Buffer. Кусков два, потому что сервер писал ответ дважды: тело приходит по мере того, как его пишут, а не целиком. Мост через Readable.fromWeb превращает его в обычный поток Node — со слушателями, pipe и pipeline. Обратный мост Readable.toWeb делает из потока файла ReadableStream: у него уже нет destroy, зато есть cancel, и после отмены read() отвечает done: true. Есть ещё Writable.toWeb, Writable.fromWeb и Duplex.toWeb — мосты построены во все стороны.

Отсюда правило выбора, и оно простое. На сервере живёт node:stream: на нём говорят fs, http, zlib, readline и pipeline с его уборкой, а каждый мост — лишний слой между ними. Web Streams берут там, где их даёт или требует чужой код: тело fetch, Response с потоком внутри, Service Worker, воркеры на краю сети — и там, где один и тот же файл должен работать и в браузере, и на сервере. Смешивать не нужно: на границе ставится мост, а внутри модуля мир один.

Полезно знать масштаб того, что мы упростили. В node 26.5.0 каталог внутренней реализации потоков — 31 файл и 14 704 строки (счёт на 29 сентября 2026 года; число меняется даже в минорной версии, так что смотреть на него стоит как на порядок, а не как на константу). Реализация Web Streams рядом — ещё 8 043 строки. Всё, что собрала глава, — 263 строки вместе с излучателем.

Уходит это на то, чего в миниатюре нет. Объектный режим и подсчёт порога в штуках. Duplex — поток, который читают и пишут одновременно, вроде сокета, — и Transform поверх него: сжатие, шифр, хеш. Сборка мелких записей в одну через cork и uncork. Защита от повторного входа, когда push зовут прямо из обработчика data. Наблюдатель finished, сборка цепочек compose, Readable.from для итераторов. И целый слой совместимости со старыми потоками, которым больше пятнадцати лет: режимы paused и flowing, read(size), unpipe.

Писать свой класс потока после этого не нужно — нужно уметь читать чужой. Модель, собранная за главу, для этого и нужна: когда fs.createReadStream зовёт read, когда write отвечает false, почему pipeline уничтожает оба конца и что случится с файлом, если выйти из цикла через break. Дальше эти потоки перестанут быть учебными: глава про HTTP начнётся с того, что из сокета приезжают куски, в которых лежит текст протокола, и разбирать их придётся самим, а статья про загрузку файлов проведёт такой поток мимо памяти прямо на диск.

Четырнадцать тысяч строк — это не «в Node всё сложно», а цена совместимости: старый код на потоках 2010 года обязан работать сегодня.

Свои потоки в Node писать почти не нужно: их дают fs, http, net, zlib и сам процесс. Это те же Readable и Writable с теми же событиями и порогом — новое в них только содержимое кусков.
В куске байты, а не текст: границу кладёт порог, и она спокойно проходит посреди двухбайтовой буквы. Целый текст дают Buffer.concat, опция encoding или readline, а не склейка toString по кускам.
Слушатель data не умеет ждать: async внутри него — тихая утечка памяти. for await читает следующий кусок после тела цикла, ошибку отдаёт исключением, а на выходе из цикла уничтожает поток (кроме iterator({ destroyOnReturn: false })).
Асинхронный итератор — это очередь кусков и обещание следующего: pause после куска, resume перед ожиданием, уборка в методе return. Обратное давление получается само собой.
Web Streams — тот же механизм другими словами и без единого события: enqueue, close, pull, cancel, reader.read(), await writer.ready. Порог считается в штуках, по умолчанию в одну.
На сервере работают с node:stream, Web Streams берут там, где их даёт чужой API (fetch, Response, Service Worker) или где код общий с браузером; на границе ставят Readable.fromWeb и Readable.toWeb.

1/4Файл с русским текстом читают потоком с порогом 16 байт и склеивают: text += chunk.toString(). Что получится?

2/4В обработчике stream.on("data", async (chunk) => { await save(chunk); }) сохранение медленное, файл большой. Что произойдёт?

3/4Цикл for await по файловому потоку прервали через break. Что стало с потоком?

4/4У WritableStream нет события drain. Как писатель узнаёт, что очередь разобрали и можно писать дальше?

Загрузка, которая дождалась

Сервер сохраняет загрузку на диск: на каждый кусок — асинхронная запись save. Код выглядит правильным, но к концу потока на диске пусто. Почините приём: поток не должен отдавать следующий кусок, пока не записан предыдущий. Трогать save и сам источник нельзя — только приём.

js

// Emitter, потоки, pipe и pipeline — в подложке
const disk = [];
const save = (chunk) =>
  new Promise((resolve) => setTimeout(() => {
    disk.push(chunk);
    resolve();
  }, 10));

let read = 0;
const upload = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    upload.push(read < 6 ? `[кусок ${read}]` : null);
  },
});

// ваш код: обработчик асинхронный, а поток об этом не знает
upload.on("data", async (chunk) => {
  await save(chunk);
});
upload.on("end", () => {
  console.log(`конец загрузки: на диске ${disk.length} из ${read - 1} кусков`);
});

страница примера

Сначала предположите, что выведет код

Слушатель обязан сам сказать потоку, что занят. Пауза действует с текущего куска, а снимают её после записи:

jsлистинг · не запускается

// Emitter, потоки, pipe и pipeline — в подложке
const disk = [];
const save = (chunk) =>
  new Promise((resolve) => setTimeout(() => {
    disk.push(chunk);
    resolve();
  }, 10));

let read = 0;
const upload = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    upload.push(read < 6 ? `[кусок ${read}]` : null);
  },
});

upload.on("data", async (chunk) => {
  // пауза действует с этого же куска: flow проверяет флаг на каждом
  upload.pause();
  await save(chunk);
  upload.resume();
});
upload.on("end", () => {
  console.log(`конец загрузки: на диске ${disk.length} из ${read - 1} кусков`);
});

// Emitter, потоки, pipe и pipeline — в подложке
const disk = [];
const save = (chunk) =>
  new Promise((resolve) => setTimeout(() => {
    disk.push(chunk);
    resolve();
  }, 10));

let read = 0;
const upload = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    upload.push(read < 6 ? `[кусок ${read}]` : null);
  },
});

upload.on("data", async (chunk) => {
  // пауза действует с этого же куска: flow проверяет флаг на каждом
  upload.pause();
  await save(chunk);
  upload.resume();
});
upload.on("end", () => {
  console.log(`конец загрузки: на диске ${disk.length} из ${read - 1} кусков`);
});

Вывод: конец загрузки: на диске 5 из 5 кусков вместо 0 из 5. Это и есть обратное давление, собранное руками из pause и resume, — и ровно это делает за вас for await, в котором следующий кусок читается после тела цикла. В настоящем коде на Node так и пишут: for await (const chunk of upload) await save(chunk);, а слушатель data оставляют для синхронной работы.

Мост наружу

Функция toWeb оборачивает наш поток в ReadableStream, чтобы отдать его коду, который умеет только Web Streams. Написана она наивно: источник вычитывается целиком, сколько бы кусков потребитель ни взял, а отмена до него не доходит. Доделайте так, чтобы диск читали по спросу, а cancel уничтожал источник. Подсказка: у Web-потока для этого есть pull и cancel, у нашего — pause, resume и destroy.

js

// Emitter, потоки, pipe и pipeline — в подложке
function toWeb(stream) {
  // ваш код: сейчас источник вычитывают без спросу, а отмена до него не доходит
  return new ReadableStream(
    {
      start(controller) {
        stream.on("data", (chunk) => controller.enqueue(chunk));
        stream.on("end", () => controller.close());
      },
    },
    { highWaterMark: 1 },
  );
}

let read = 0;
const file = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    console.log(`диск: читаю кусок ${read}`);
    file.push(read < 7 ? `[кусок ${read}]` : null);
  },
  destroy() {
    console.log("диск: файл закрыт");
  },
});

async function main() {
  const reader = toWeb(file).getReader();
  for (let i = 0; i < 2; i += 1) {
    const { value } = await reader.read();
    console.log(`обработка: ${value}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
  await reader.cancel();
  console.log(`ушли после двух кусков, а диск прочитал ${read}`);
}
main();

страница примера

Сначала предположите, что выведет код

Три правки: пауза после каждого куска, снятие паузы в pull, уничтожение в cancel. Заодно стоит перекинуть ошибку — иначе она потеряется по дороге:

jsлистинг · не запускается

// Emitter, потоки, pipe и pipeline — в подложке
function toWeb(stream) {
  return new ReadableStream(
    {
      start(controller) {
        stream.on("data", (chunk) => {
          // кусок уехал в очередь — дальше только по спросу
          stream.pause();
          controller.enqueue(chunk);
        });
        stream.on("end", () => controller.close());
        stream.on("error", (err) => controller.error(err));
      },
      pull() {
        stream.resume();
      },
      cancel() {
        stream.destroy();
      },
    },
    { highWaterMark: 1 },
  );
}

let read = 0;
const file = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    console.log(`диск: читаю кусок ${read}`);
    file.push(read < 7 ? `[кусок ${read}]` : null);
  },
  destroy() {
    console.log("диск: файл закрыт");
  },
});

async function main() {
  const reader = toWeb(file).getReader();
  for (let i = 0; i < 2; i += 1) {
    const { value } = await reader.read();
    console.log(`обработка: ${value}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
  await reader.cancel();
  console.log(`ушли после двух кусков, а диск прочитал ${read}`);
}
main();

// Emitter, потоки, pipe и pipeline — в подложке
function toWeb(stream) {
  return new ReadableStream(
    {
      start(controller) {
        stream.on("data", (chunk) => {
          // кусок уехал в очередь — дальше только по спросу
          stream.pause();
          controller.enqueue(chunk);
        });
        stream.on("end", () => controller.close());
        stream.on("error", (err) => controller.error(err));
      },
      pull() {
        stream.resume();
      },
      cancel() {
        stream.destroy();
      },
    },
    { highWaterMark: 1 },
  );
}

let read = 0;
const file = new Readable({
  highWaterMark: 1,
  read() {
    read += 1;
    console.log(`диск: читаю кусок ${read}`);
    file.push(read < 7 ? `[кусок ${read}]` : null);
  },
  destroy() {
    console.log("диск: файл закрыт");
  },
});

async function main() {
  const reader = toWeb(file).getReader();
  for (let i = 0; i < 2; i += 1) {
    const { value } = await reader.read();
    console.log(`обработка: ${value}`);
    await new Promise((resolve) => setTimeout(resolve, 10));
  }
  await reader.cancel();
  console.log(`ушли после двух кусков, а диск прочитал ${read}`);
}
main();

Было «ушли после двух кусков, а диск прочитал 7» — весь файл уехал в очередь Web-потока, хотя потребителю хватило двух кусков. Стало «прочитал 4»: два отданных, один про запас в очереди и один, который поток успел прочитать вперёд. Строка «диск: файл закрыт» есть в обоих случаях, но по разным причинам: в наивной версии файл кончился сам, в правильной его закрыла отмена. Ровно это и делает Readable.toWeb из node:stream — с той разницей, что там ещё переносятся ошибки, порог и режим байтов.

Обратное давление по-браузерному

Тот же протокол, но в Web Streams. Запись в файл медленная, порог очереди — два куска, а пишут шесть подряд. Посмотрите, что печатает desiredSize, и почините цикл так, чтобы место в очереди не уходило в минус.

js

const disk = [];
const file = new WritableStream(
  {
    write(chunk) {
      return new Promise((resolve) => setTimeout(() => {
        disk.push(chunk);
        resolve();
      }, 10));
    },
    close() {
      console.log(`файл закрыт, в нём кусков: ${disk.length}`);
    },
  },
  { highWaterMark: 2 },
);

async function save(parts) {
  const writer = file.getWriter();
  for (const part of parts) {
    console.log(`место в очереди: ${writer.desiredSize}`);
    // ваш код: очередь растёт, сколько бы кусков ни пришло
    writer.write(part);
  }
  await writer.close();
}

save(["[1]", "[2]", "[3]", "[4]", "[5]", "[6]"]);

Сначала предположите, что выведет код

Ждать нужно промис ready — это Web-версия drain:

jsлистинг · не запускается

const disk = [];
const file = new WritableStream(
  {
    write(chunk) {
      return new Promise((resolve) => setTimeout(() => {
        disk.push(chunk);
        resolve();
      }, 10));
    },
    close() {
      console.log(`файл закрыт, в нём кусков: ${disk.length}`);
    },
  },
  { highWaterMark: 2 },
);

async function save(parts) {
  const writer = file.getWriter();
  for (const part of parts) {
    console.log(`место в очереди: ${writer.desiredSize}`);
    await writer.ready;
    writer.write(part);
  }
  await writer.close();
}

save(["[1]", "[2]", "[3]", "[4]", "[5]", "[6]"]);

const disk = [];
const file = new WritableStream(
  {
    write(chunk) {
      return new Promise((resolve) => setTimeout(() => {
        disk.push(chunk);
        resolve();
      }, 10));
    },
    close() {
      console.log(`файл закрыт, в нём кусков: ${disk.length}`);
    },
  },
  { highWaterMark: 2 },
);

async function save(parts) {
  const writer = file.getWriter();
  for (const part of parts) {
    console.log(`место в очереди: ${writer.desiredSize}`);
    await writer.ready;
    writer.write(part);
  }
  await writer.close();
}

save(["[1]", "[2]", "[3]", "[4]", "[5]", "[6]"]);

Было «место в очереди: 2, 1, 0, −1, −2, −3» — три лишних куска висят в памяти. Стало «2, 1, 0, 0, 0, 0»: каждая следующая запись ждёт, пока предыдущая освободит место. Записаны в обоих случаях все шесть кусков, и в этом вся коварность: наивный код отличается от правильного не результатом, а памятью — на шести кусках разницы не видно, на шестигигабайтном файле она видна очень хорошо.

Настоящие потоки: fs, http, Web Streams и async-итерация

Видеоразбор темы за 5–7 минут

Файл: кусок — это байтыГраницу кладёт порог, а не строка и даже не символ

for await вместо слушателяСлушатель не умеет ждать, а цикл умеет

Web Streams: то же самое другими словамиНи одного события: контроллер, промисы и очередь в штуках

Мост: настоящий HTTP с обеих сторонЗапрос — поток Node, тело ответа — Web-поток, между ними две функции

Что внутри у настоящихМиниатюра — две с половиной сотни строк, оригинал — четырнадцать тысяч

Итогошпаргалка по теме

Проверьте себявопросы по теме

Задачизадачи с решениями

Загрузка, которая дождалась

Мост наружу

Обратное давление по-браузерному