Поток¶

v18.x.x

Поток - это абстрактный интерфейс для работы с потоковыми данными в Node.js. Модуль node:stream предоставляет API для реализации интерфейса потока.

В Node.js существует множество объектов потока. Например, запрос к HTTP-серверу и process.stdout являются экземплярами потока.

Потоки могут быть доступны для чтения, записи или и для того, и для другого. Все потоки являются экземплярами EventEmitter.

Чтобы получить доступ к модулю node:stream:

1

const stream = require('node:stream');

Модуль node:stream полезен для создания новых типов экземпляров потоков. Обычно нет необходимости использовать модуль node:stream для потребления потоков.

Организация данного документа¶

Этот документ содержит два основных раздела и третий раздел для примечаний. В первом разделе объясняется, как использовать существующие потоки в приложении. Во втором разделе объясняется, как создавать новые типы потоков.

Типы потоков¶

В Node.js существует четыре основных типа потоков:

• Writable: потоки, в которые можно записывать данные (например, fs.createWriteStream()).
• Readable: потоки, из которых можно читать данные (например, fs.createReadStream()).
• Duplex: потоки, которые являются одновременно Readable и Writable (например, net.Socket).
• Transform: дуплексные потоки, которые могут изменять или преобразовывать данные по мере их записи и чтения (например, zlib.createDeflate()).

Кроме того, в этот модуль входят служебные функции stream.pipeline(), stream.finished(), stream.Readable.from() и stream.addAbortSignal().

Promise API потоков¶

API stream/promises предоставляет альтернативный набор асинхронных служебных функций для потоков, которые возвращают объекты Promise, а не используют обратные вызовы. API доступен через require('node:stream/promises') или require('node:stream').promises.

stream.pipeline¶

1

stream.pipeline(source[, ...transforms], destination[, options])

1

stream.pipeline(streams[, options])

• streams {Stream[]} | {Iterable[]} | {AsyncIterable[]} | {Function[]}
• source <Stream> | <Iterable> | <AsyncIterable> | <Function>
  • Возвращает: <Promise> | <AsyncIterable>
• ...transforms <Stream> | <Function>
  • source <AsyncIterable>
  • Возвращает: <Promise> | <AsyncIterable>
• destination <Stream> | <Function>
  • source <AsyncIterable>
  • Возвращает: <Promise> | <AsyncIterable>
• options <Object>
  • сигнал <AbortSignal>
  • end <boolean>
• Возвращает: <Promise> Выполняется, когда конвейер завершен.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
    await pipeline(
        fs.createReadStream('archive.tar'),
        zlib.createGzip(),
        fs.createWriteStream('archive.tar.gz')
    );
    console.log('Pipeline succeeded.');
}

run().catch(console.error);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import { pipeline } from 'node:stream/promises';
import {
    createReadStream,
    createWriteStream,
} from 'node:fs';
import { createGzip } from 'node:zlib';

await pipeline(
    createReadStream('archive.tar'),
    createGzip(),
    createWriteStream('archive.tar.gz')
);
console.log('Pipeline succeeded.');

Чтобы использовать AbortSignal, передайте его внутри объекта options в качестве последнего аргумента. Когда сигнал будет прерван, на базовом конвейере будет вызвана команда destroy с сообщением AbortError.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
    const ac = new AbortController();
    const signal = ac.signal;

    setImmediate(() => ac.abort());
    await pipeline(
        fs.createReadStream('archive.tar'),
        zlib.createGzip(),
        fs.createWriteStream('archive.tar.gz'),
        { signal }
    );
}

run().catch(console.error); // AbortError

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

import { pipeline } from 'node:stream/promises';
import {
    createReadStream,
    createWriteStream,
} from 'node:fs';
import { createGzip } from 'node:zlib';

const ac = new AbortController();
const { signal } = ac;
setImmediate(() => ac.abort());
try {
    await pipeline(
        createReadStream('archive.tar'),
        createGzip(),
        createWriteStream('archive.tar.gz'),
        { signal }
    );
} catch (err) {
    console.error(err); // AbortError
}

API pipeline также поддерживает асинхронные генераторы:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
    await pipeline(
        fs.createReadStream('lowercase.txt'),
        async function* (source, { signal }) {
            source.setEncoding('utf8'); // Работаем со строками, а не с `буфером`.
            for await (const chunk of source) {
                yield await processChunk(chunk, { signal });
            }
        },
        fs.createWriteStream('uppercase.txt')
    );
    console.log('Pipeline succeeded.');
}

run().catch(console.error);

1
2
3
4
5
6
7

import { pipeline } from 'node:stream/promises';
import fs from 'node:fs';
await pipeline(async function* ({ signal }) {
    await someLongRunningfn({ signal });
    yield 'asd';
}, fs.createWriteStream('uppercase.txt'));
console.log('Pipeline succeeded.');

API pipeline предоставляет версию обратного вызова:

stream.finished¶

1

stream.finished(stream[, options])

• stream <Stream>
• options <Object>
  • error <boolean> | <undefined>
  • readable <boolean> | <undefined>
  • writable <boolean> | <undefined>
  • сигнал: <AbortSignal> | <undefined>
• Возвращает: <Promise> Выполняется, когда поток больше не доступен для чтения или записи.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

async function run() {
    await finished(rs);
    console.log('Поток закончил чтение.');
}

run().catch(console.error);
rs.resume(); // Слить поток.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

import { finished } from 'node:stream/promises';
import { createReadStream } from 'node:fs';

const rs = createReadStream('archive.tar');

async function run() {
    await finished(rs);
    console.log('Поток закончил чтение.');
}

run().catch(console.error);
rs.resume(); // Слить поток.

API finished предоставляет версию обратного вызова:

Объектный режим¶

Все потоки, создаваемые API Node.js, работают исключительно со строками и объектами Buffer (или Uint8Array). Однако, возможно, что реализация потоков может работать с другими типами значений JavaScript (за исключением null, который служит специальной цели в потоках). Такие потоки считаются работающими в "объектном режиме".

Экземпляры потоков переводятся в объектный режим с помощью опции objectMode при создании потока. Попытка переключить существующий поток в объектный режим небезопасна.

Буферизация¶

Потоки Writable и Readable будут хранить данные во внутреннем буфере.

Объем потенциально буферизуемых данных зависит от параметра highWaterMark, передаваемого в конструктор потока. Для обычных потоков опция highWaterMark определяет общее количество байт. Для потоков, работающих в объектном режиме, параметр highWaterMark указывает общее количество объектов.

Данные буферизуются в потоках Readable, когда реализация вызывает stream.push(chunk). Если потребитель потока не вызывает stream.read(), данные будут находиться во внутренней очереди, пока не будут потреблены.

Когда общий размер внутреннего буфера чтения достигнет порога, заданного параметром highWaterMark, поток временно прекратит чтение данных из базового ресурса, пока данные, находящиеся в буфере, не будут потреблены (то есть поток перестанет вызывать внутренний метод readable._read(), который используется для заполнения буфера чтения).

Буферизация данных в потоках Writable происходит при многократном вызове метода writable.write(chunk). Пока общий размер внутреннего буфера записи ниже порога, установленного highWaterMark, вызовы writable.write() будут возвращать true. Как только размер внутреннего буфера достигнет или превысит highWaterMark, будет возвращена false.

Ключевой целью API stream, в частности метода stream.pipe(), является ограничение буферизации данных до приемлемого уровня, чтобы источники и пункты назначения с разной скоростью не перегружали доступную память.

Опция highWaterMark - это порог, а не предел: она определяет количество данных, которое поток буферизирует, прежде чем перестанет запрашивать больше данных. Она не обеспечивает жесткого ограничения памяти в целом. Конкретные реализации потоков могут выбрать более строгие ограничения, но это необязательно.

Поскольку потоки Duplex и Transform являются одновременно Readable и Writable, каждый из них поддерживает два отдельных внутренних буфера, используемых для чтения и записи, что позволяет каждой стороне работать независимо от другой, поддерживая при этом соответствующий и эффективный поток данных. Например, экземпляры net.Socket представляют собой Duplex потоки, чья Readable сторона позволяет потреблять данные, полученные из сокета, и чья Writable сторона позволяет записывать данные в сокет. Поскольку данные могут записываться в сокет быстрее или медленнее, чем приниматься, каждая сторона должна работать (и буферизироваться) независимо от другой.

Механика внутренней буферизации является внутренней деталью реализации и может быть изменена в любое время. Однако для некоторых продвинутых реализаций внутренние буферы могут быть r

API для потребителей потоков¶

Почти все приложения Node.js, независимо от того, насколько они просты, в той или иной мере используют потоки. Ниже приведен пример использования потоков в приложении Node.js, реализующем HTTP-сервер:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

const http = require('node:http');

const server = http.createServer((req, res) => {
    // `req` - это http.IncomingMessage, который является читаемым потоком.
    // `res` - это http.ServerResponse, который является записываемым потоком.

    let body = '';
    // Получаем данные в виде строк utf8.
    // Если кодировка не задана, будут получены объекты Buffer.
    req.setEncoding('utf8');

    // Читаемые потоки испускают события 'data' после добавления слушателя.
    req.on('data', (chunk) => {
        body += chunk;
    });

    // Событие 'end' означает, что все тело было получено.
    req.on('end', () => {
        try {
            const data = JSON.parse(body);
            // Записываем обратно что-нибудь интересное для пользователя:
            res.write(typeof data);
            res.end();
        } catch (er) {
            // ой! плохой json!
            res.statusCode = 400;
            return res.end(`error: ${er.message}`);
        }
    });
});

server.listen(1337);

// $ curl localhost:1337 -d "{}"
// объект
// $ curl localhost:1337 -d "\"foo\""
// строка
// $ curl localhost:1337 -d "not json"
// ошибка: Unexpected token 'o', "not json" is not valid JSON

Writable потоки (такие как res в примере) раскрывают такие методы как write() и end(), которые используются для записи данных в поток.

Потоки Readable используют API EventEmitter для уведомления кода приложения, когда данные доступны для чтения из потока. Эти доступные данные могут быть считаны из потока несколькими способами.

Потоки Writable и Readable используют API EventEmitter различными способами для передачи текущего состояния потока.

Потоки Duplex и Transform являются одновременно Writable и Readable.

Приложениям, которые записывают данные в поток или потребляют данные из потока, не требуется реализовывать интерфейсы потоков напрямую, и у них, как правило, нет причин вызывать require('node:stream').

Разработчики, желающие реализовать новые типы потоков, должны обратиться к разделу "API для реализаторов потоков".

Записываемые потоки¶

Записываемые потоки - это абстракция для назначения, в которое записываются данные.

Примеры Writable потоков включают:

• HTTP-запросы, на клиенте
• HTTP ответы, на сервере
• потоки записи fs
• потоки zlib
• crypto streams
• TCP сокеты
• stdin дочернего процесса
• process.stdout, process.stderr

Некоторые из этих примеров на самом деле являются потоками Duplex, которые реализуют интерфейс Writable.

Все потоки Writable реализуют интерфейс, определенный классом stream.Writable.

Хотя конкретные экземпляры потоков Writable могут различаться различными способами, все потоки Writable следуют одной и той же основной схеме использования, как показано в примере ниже:

1
2
3
4

const myStream = getWritableStreamSomehow();
myStream.write('некоторые данные');
myStream.write('еще немного данных');
myStream.end('закончил запись данных');

stream.Writable¶

Событие: close¶

Событие 'close' генерируется, когда поток и любой из его базовых ресурсов (например, дескриптор файла) закрыты. Это событие указывает на то, что больше не будет испускаться никаких событий, и никаких дальнейших вычислений не будет.

Поток Writable всегда будет испускать событие close, если он создан с опцией emitClose.

Событие: drain¶

Если вызов stream.write(chunk) возвращает false, событие 'drain' будет выдано, когда будет уместно возобновить запись данных в поток.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// Записываем данные в предоставленный поток с возможностью записи миллион раз.
// Будьте внимательны к обратному давлению.
function writeOneMillionTimes(
    writer,
    data,
    encoding,
    callback
) {
    let i = 1000000;
    write();
    function write() {
        let ok = true;
        do {
            i--;
            if (i === 0) {
                // Последний раз!
                writer.write(data, encoding, callback);
            } else {
                // Узнайте, должны ли мы продолжить или подождать.
                // Не передавайте обратный вызов, потому что мы еще не закончили.
                ok = writer.write(data, encoding);
            }
        } while (i > 0 && ok);
        if (i > 0) {
            // Пришлось остановиться раньше времени!
            // Напишем еще немного, когда все стечет.
            writer.once('drain', write);
        }
    }
}

Событие: error¶

• <Error>

Событие 'error' генерируется, если во время записи или передачи данных произошла ошибка. При вызове обратного вызова слушателя ему передается единственный аргумент Error.

Поток закрывается при возникновении события 'error', если только опция autoDestroy не была установлена в false при создании потока.

После error больше не должно происходить никаких событий, кроме close (включая события error).

Событие: finish¶

Событие finish возникает после вызова метода stream.end(), и все данные были переданы в базовую систему.

1
2
3
4
5
6
7
8

const writer = getWritableStreamSomehow();
for (let i = 0; i < 100; i++) {
    writer.write(`hello, #${i}!\n`);
}
writer.on('finish', () => {
    console.log('Все записи завершены');
});
writer.end('Это конец\n');

Событие: pipe¶

• src <stream.Readable> исходный поток, который передается по трубопроводу в этот объект записи

Событие pipe возникает, когда метод stream.pipe() вызывается на потоке readable, добавляя этот writable к его набору пунктов назначения.

1
2
3
4
5
6
7

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('pipe', (src) => {
    console.log('Что-то передается в писатель');
    assert.equal(src, reader);
});
reader.pipe(writer);

Событие: unpipe¶

• src <stream.Readable> Исходный поток, который unpipeed этот writable

Событие unpipe испускается, когда метод stream.unpipe() вызывается на потоке Readable, удаляя этот Writable из его набора пунктов назначения.

Это также происходит в случае, если этот поток Writable выдает ошибку при передаче в него потока Readable.

1
2
3
4
5
6
7
8

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('unpipe', (src) => {
    console.log('Что-то перестало поступать в писатель');
    assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer);

writable.cork¶

1

writable.cork();

Метод writable.cork() заставляет все записанные данные буферизироваться в памяти. Буферизованные данные будут удалены при вызове методов stream.uncork() или stream.end().

Основная цель метода writable.cork() - приспособиться к ситуации, когда в поток записывается несколько небольших фрагментов в быстрой последовательности. Вместо того, чтобы немедленно пересылать их по назначению, writable.cork() буферизирует все куски до вызова writable.uncork(), который передаст их все в writable._writev(), если таковой имеется. Это предотвращает ситуацию блокировки в голове строки, когда данные буферизируются в ожидании обработки первого небольшого фрагмента. Однако использование writable.cork() без реализации writable._writev() может негативно сказаться на пропускной способности.

См. также: writable.uncork(), writable._writev().

writable.destroy¶

1

writable.destroy([error]);

• error <Error> Необязательно, ошибка, которую нужно выдать с событием 'error'.
• Возвращает: <this>

Уничтожить поток. Опционально выдает событие 'error' и выдает событие 'close' (если emitClose не установлено в false). После этого вызова поток, доступный для записи, завершен, и последующие вызовы write() или end() приведут к ошибке ERR_STREAM_DESTROYED. Это деструктивный и немедленный способ уничтожения потока. Предыдущие вызовы write() могут не уничтожить поток и вызвать ошибку ERR_STREAM_DESTROYED. Используйте end() вместо destroy, если данные должны быть удалены до закрытия, или дождитесь события 'drain' перед уничтожением потока.

1
2
3
4
5
6
7
8
9

const { Writable } = require('node:stream');

const myStream = new Writable();

const fooErr = new Error('ошибка foo');
myStream.destroy(fooErr);
myStream.on('error', (fooErr) =>
    console.error(fooErr.message)
); // ошибка foo

1
2
3
4
5
6

const { Writable } = require('node:stream');

const myStream = new Writable();

myStream.destroy();
myStream.on('error', function wontHappen() {});

1
2
3
4
5
6
7

const { Writable } = require('node:stream');

const myStream = new Writable();
myStream.destroy();

myStream.write('foo', (error) => console.error(error.code));
// ERR_STREAM_DESTROYED

После вызова destroy() любые дальнейшие вызовы будут бесполезны, и никакие другие ошибки, кроме _destroy(), не могут быть выданы как 'error'.

Реализаторы не должны переопределять этот метод, а вместо этого реализовать writable._destroy().

writable.closed¶

• <boolean>

Является true после испускания close.

writable.destroyed¶

• <boolean>

Является true после вызова writable.destroy().

1
2
3
4
5
6
7

const { Writable } = require('node:stream');

const myStream = new Writable();

console.log(myStream.destroyed); // false
myStream.destroy();
console.log(myStream.destroyed); // true

writable.end¶

1

writable.end([chunk[, encoding]][, callback])

• chunk <string> | <Buffer> | <Uint8Array> | <any> Необязательные данные для записи. Для потоков, не работающих в объектном режиме, chunk должен быть строкой, Buffer или Uint8Array. Для потоков, работающих в объектном режиме, chunk может быть любым значением JavaScript, кроме null.
• encoding <string> Кодировка, если chunk является строкой.
• callback <Function> Обратный вызов для завершения потока.
• Возвращает: <this>

Вызов метода writable.end() сигнализирует о том, что данные больше не будут записываться в Writable. Необязательные аргументы chunk и encoding позволяют записать последний дополнительный фрагмент данных непосредственно перед закрытием потока.

Вызов метода stream.write() после вызова stream.end() приведет к ошибке.

1
2
3
4
5
6

// Напишите 'hello, ', а затем закончите 'world!'.
const fs = require('node:fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Писать больше сейчас запрещено!

writable.setDefaultEncoding¶

1

writable.setDefaultEncoding(encoding);

• encoding <string> Новая кодировка по умолчанию
• Возвращает: <this>

Метод writable.setDefaultEncoding() устанавливает кодировку по умолчанию для потока Writable.

writable.uncork¶

1

writable.uncork();

Метод writable.uncork() очищает все данные, буферизованные с момента вызова stream.cork().

При использовании writable.cork() и writable.uncork() для управления буферизацией записей в поток, отложите вызов writable.uncork() с помощью process.nextTick(). Это позволяет выполнять пакетную обработку всех вызовов writable.write(), которые происходят в данной фазе цикла событий Node.js.

1
2
3
4

stream.cork();
stream.write('some ');
stream.write('data ');
process.nextTick(() => stream.uncork());

Если метод writable.cork() вызывается несколько раз на потоке, то такое же количество вызовов writable.uncork() должно быть вызвано для промывки буферизованных данных.

1
2
3
4
5
6
7
8
9

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
    stream.uncork();
    // Данные не будут удалены до тех пор, пока функция uncork() не будет вызвана во второй раз.
    stream.uncork();
});

См. также: writable.cork().

writable.writable¶

• <boolean>

Является true, если безопасно вызывать writable.write(), что означает, что поток не был уничтожен, ошибочен или завершен.

writable.writableAborted¶

• <boolean>

Возвращает, был ли поток уничтожен или ошибочен перед выдачей 'finish'.

writable.writableEnded¶

• <boolean>

Является true после вызова writable.end(). Это свойство не указывает, были ли данные выгружены, для этого используйте writable.writableFinished.

writable.writableCorked¶

• <integer>

Количество раз, которое необходимо вызвать writable.uncork(), чтобы полностью откупорить поток.

writable.errored¶

• <Error>

Возвращает ошибку, если поток был уничтожен с ошибкой.

writable.writableFinished¶

• <boolean>

Устанавливается в true непосредственно перед испусканием события 'finish'.

writable.writableHighWaterMark¶

• <number>

Возвращает значение highWaterMark, переданное при создании этой writable.

writable.writableLength¶

• <number>

Это свойство содержит количество байтов (или объектов) в очереди, готовых к записи. Значение предоставляет данные интроспекции относительно состояния highWaterMark.

writable.writableNeedDrain¶

• <boolean>

Является true, если буфер потока был заполнен и поток будет издавать сигнал 'drain'.

writable.writableObjectMode¶

• <boolean>

Получатель для свойства objectMode данного потока Writable.

writable.write¶

1

writable.write(chunk[, encoding][, callback])

• chunk <string> | <Buffer> | <Uint8Array> | <any> Необязательные данные для записи. Для потоков, не работающих в объектном режиме, chunk должен быть строкой, Buffer или Uint8Array. Для потоков, работающих в объектном режиме, chunk может быть любым значением JavaScript, кроме null.
• encoding <string> | <null> Кодировка, если chunk является строкой. По умолчанию: 'utf8'.
• callback <Function> Обратный вызов, когда этот фрагмент данных будет удален.
• Возвращает: <boolean> false, если поток желает, чтобы вызывающий код дождался события 'drain', прежде чем продолжить запись дополнительных данных; иначе true.

Метод writable.write() записывает некоторые данные в поток и вызывает предоставленный callback, когда данные будут полностью обработаны. Если произошла ошибка, будет вызван callback с ошибкой в качестве первого аргумента. Вызов callback происходит асинхронно и до того, как будет выдана 'error'.

Возвращаемое значение равно true, если внутренний буфер меньше, чем highWaterMark, настроенный при создании потока после приема chunk. Если возвращается false, дальнейшие попытки записи данных в поток должны быть прекращены до тех пор, пока не произойдет событие 'drain'.

Пока поток не осушен, вызовы write() будут буферизировать chunk и возвращать false. Когда все текущие буферизованные фрагменты будут осушены (приняты операционной системой для доставки), произойдет событие 'drain'. Если write() возвращает false, не записывайте больше чанков, пока не произойдет событие 'drain'. Хотя вызов write() на потоке, который не осушается, разрешен, Node.js будет буферизировать все записанные фрагменты до тех пор, пока не произойдет максимальное использование памяти, после чего произойдет безусловное прерывание. Даже до прерывания, высокое использование памяти приведет к плохой работе сборщика мусора и высокой RSS (которая обычно не возвращается в систему, даже после того, как память больше не требуется). Поскольку TCP-сокеты могут никогда не разряжаться, если удаленный пир не читает данные, запись в сокет, который не разряжается, может привести к уязвимости, которую можно использовать удаленно.

Запись данных, пока поток не иссякает, особенно проблематична для Transform, поскольку потоки Transform по умолчанию приостанавливаются до тех пор, пока они не будут переданы по трубопроводу или не будет добавлен обработчик событий data или readable.

Если данные для записи могут быть сгенерированы или получены по требованию, рекомендуется инкапсулировать логику в Readable и использовать stream.pipe(). Однако, если вызов write() предпочтительнее, можно соблюсти обратное давление и избежать проблем с памятью, используя событие 'drain':

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

function write(data, cb) {
    if (!stream.write(data)) {
        stream.once('drain', cb);
    } else {
        process.nextTick(cb);
    }
}

// Wait for cb to be called before doing any other write.
write('hello', () => {
    console.log('Write completed, do more writes now.');
});

Поток Writable в объектном режиме всегда будет игнорировать аргумент encoding.

Читаемые потоки¶

Читаемые потоки - это абстракция для источника, из которого потребляются данные.

Примеры Readable потоков включают:

• HTTP ответы, на клиенте
• HTTP-запросы, на сервере
• потоки чтения fs
• потоки zlib
• crypto streams
• TCP сокеты
• stdout и stderr дочернего процесса
• process.stdin

Все потоки Readable реализуют интерфейс, определенный классом stream.Readable.

Два режима чтения¶

Потоки Readable эффективно работают в одном из двух режимов: текущем и приостановленном. Эти режимы отличаются от объектного режима. Поток Readable может быть в объектном режиме или нет, независимо от того, находится ли он в потоковом режиме или в режиме паузы.

• В режиме потока данные считываются из базовой системы автоматически и предоставляются приложению как можно быстрее с помощью событий через интерфейс EventEmitter.
• В режиме паузы для чтения фрагментов данных из потока необходимо явно вызывать метод stream.read().

Все потоки Readable начинаются в режиме паузы, но могут быть переключены в режим потока одним из следующих способов:

• Добавление обработчика события 'data'.
• Вызов метода stream.resume().
• Вызов метода stream.pipe() для отправки данных на Writable.

Readable может переключиться обратно в режим паузы, используя одно из следующих действий:

• Если нет мест назначения, вызвав метод stream.pause().
• Если есть места назначения труб, то путем удаления всех мест назначения труб. Несколько мест назначения труб можно удалить, вызвав метод stream.unpipe().

Важно помнить, что Readable не будет генерировать данные, пока не будет предоставлен механизм для потребления или игнорирования этих данных. Если механизм потребления отключен или убран, Readable будет пытаться прекратить генерировать данные.

По причинам обратной совместимости, удаление обработчиков событий 'data' не будет автоматически приостанавливать поток. Кроме того, если есть конечные пункты назначения, то вызов stream.pause() не гарантирует, что поток останется приостановленным, когда эти пункты назначения иссякнут и запросят больше данных.

Если Readable переключается в режим потока и нет потребителей, доступных для обработки данных, эти данные будут потеряны. Это может произойти, например, когда метод readable.resume() вызывается без слушателя, присоединенного к событию 'data', или когда обработчик события 'data' удаляется из потока.

Добавление обработчика события 'readable' автоматически прекращает поток, и данные должны быть потреблены через readable.read(). Если обработчик события 'readable' удален, то поток снова начнет течь, если есть обработчик события 'data'.

Три состояния¶

Два режима работы потока Readable - это упрощенная абстракция для более сложного внутреннего управления состояниями, которое происходит в реализации потока Readable.

В частности, в любой момент времени каждый Readable находится в одном из трех возможных состояний:

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true.

Когда readable.readableFlowing имеет значение null, механизм потребления данных потока не предусмотрен. Поэтому поток не будет генерировать данные. В этом состоянии прикрепление слушателя для события 'data', вызов метода readable.pipe() или вызов метода readable.resume() переключит readable.readableFlowing в true, заставляя Readable начать активно генерировать события по мере генерации данных.

Вызов readable.pause(), readable.unpipe() или получение обратного давления приведет к тому, что readable.readableFlowing будет установлен как false, временно останавливая поток событий, но не останавливая генерацию данных. Находясь в этом состоянии, прикрепление слушателя для события 'data' не переключит readable.readableFlowing в true.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

const { PassThrough, Writable } = require('node:stream');
const pass = new PassThrough();
const writable = new Writable();

pass.pipe(writable);
pass.unpipe(writable);
// readableFlowing теперь false.

pass.on('data', (chunk) => {
    console.log(chunk.toString());
});
// readableFlowing все еще ложно.
pass.write('ok'); // Не будет выдавать 'data'.
pass.resume(); // Должен быть вызван, чтобы поток выдал 'данные'.
// readableFlowing теперь истина.

Пока readable.readableFlowing имеет значение false, данные могут накапливаться во внутреннем буфере потока.

Выберите один стиль API¶

API потока Readable развивался на протяжении нескольких версий Node.js и предоставляет несколько методов потребления данных потока. В целом, разработчики должны выбрать один из методов потребления данных и никогда не должны использовать несколько методов для потребления данных из одного потока. В частности, использование комбинации on('data'), on('readable'), pipe() или асинхронных итераторов может привести к неинтуитивному поведению.

stream.Readable¶

Событие: close¶

Событие close генерируется, когда поток и любой из его базовых ресурсов (например, дескриптор файла) закрыты. Это событие указывает на то, что больше не будет испускаться никаких событий, и никаких дальнейших вычислений не будет.

Поток Readable всегда будет испускать событие close, если он создан с опцией emitClose.

Событие: data¶

• chunk <Buffer> | <string> | <any> Кусок данных. Для потоков, не работающих в объектном режиме, чанк будет либо строкой, либо буфером. Для потоков, работающих в объектном режиме, чанк может быть любым значением JavaScript, кроме null.

Событие 'data' генерируется всякий раз, когда поток передает право собственности на кусок данных потребителю. Это может происходить всякий раз, когда поток переключается в режим потока, вызывая readable.pipe(), readable.resume() или присоединяя обратный вызов слушателя к событию 'data'. Событие 'data' также будет возникать всякий раз, когда вызывается метод readable.read() и фрагмент данных доступен для возврата.

Прикрепление слушателя события 'data' к потоку, который не был явно приостановлен, переключит поток в режим потока. Данные будут передаваться, как только они станут доступны.

В обратный вызов слушателя будет передан фрагмент данных в виде строки, если для потока была задана кодировка по умолчанию с помощью метода readable.setEncoding(); в противном случае данные будут переданы в виде Buffer.

1
2
3
4

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
    console.log(`Принято ${chunk.length} байт данных.`);
});

Событие: end¶

Событие 'end' происходит, когда больше нет данных для потребления из потока.

Событие 'end' не будет вызвано, пока данные не будут полностью израсходованы. Этого можно добиться, переключив поток в режим потока, или вызывая stream.read() несколько раз, пока все данные не будут потреблены.

1
2
3
4
5
6
7

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
    console.log(`Получено ${chunk.length} байт данных.`);
});
readable.on('end', () => {
    console.log('Больше данных не будет.');
});

Событие: error¶

• <Error>

Событие 'error' может быть вызвано реализацией Readable в любое время. Как правило, это может произойти, если базовый поток не может генерировать данные из-за внутреннего сбоя или когда реализация потока пытается передать недопустимый фрагмент данных.

Обратному вызову слушателя будет передан единственный объект Error.

Событие: pause¶

Событие pause происходит, когда вызывается stream.pause() и readableFlowing не равно false.

Событие: readable¶

Событие 'readable' генерируется, когда из потока доступны данные для чтения или когда достигнут конец потока. По сути, событие 'readable' указывает на то, что в потоке есть новая информация. Если данные доступны, stream.read() вернет эти данные.

1
2
3
4
5
6
7
8
9

const readable = getReadableStreamSomehow();
readable.on('readable', function () {
    // Теперь есть некоторые данные для чтения.
    let data;

    while ((data = this.read()) !== null) {
        console.log(data);
    }
});

Если достигнут конец потока, вызов stream.read() вернет null и вызовет событие 'end'. Это также верно, если никогда не было данных для чтения. Например, в следующем примере foo.txt является пустым файлом:

1
2
3
4
5
6
7
8

const fs = require('node:fs');
const rr = fs.createReadStream('foo.txt');
rr.on('readable', () => {
    console.log(`readable: ${rr.read()}`);
});
rr.on('end', () => {
    console.log('end');
});

Результатом выполнения этого скрипта будет:

1
2
3

$ node test.js
readable: null
end

В некоторых случаях прикрепление слушателя для события 'readable' приведет к считыванию некоторого количества данных во внутренний буфер.

В целом, механизмы событий readable.pipe() и 'data' проще для понимания, чем событие 'readable'. Однако обработка 'readable' может привести к увеличению пропускной способности.

Если одновременно используются readable и 'data', 'readable' имеет приоритет в управлении потоком, т. е. 'data' будет выдаваться только при вызове stream.read(). Свойство readableFlowing станет false. Если есть слушатели 'data', когда 'readable' будет удалено, поток начнет течь, т. е. события 'data' будут испускаться без вызова .resume().

Событие: resume¶

Событие 'resume' происходит, когда вызывается stream.resume() и readableFlowing не является true.

readable.destroy¶

1

readable.destroy([error]);

• error <Error> Ошибка, которая будет передана в качестве полезной нагрузки в событии 'error'.
• Возвращает: <this>

Уничтожить поток. Опционально испускает событие 'error' и испускает событие 'close' (если emitClose не установлено в false). После этого вызова читаемый поток освободит все внутренние ресурсы, и последующие вызовы push() будут игнорироваться.

После вызова destroy() все последующие вызовы будут бесполезны, и никакие другие ошибки, кроме _destroy(), не могут быть выданы как 'error'.

Реализаторы не должны переопределять этот метод, а вместо этого реализовать readable._destroy().

readable.closed¶

• <boolean>

Является true после испускания close.

readable.destroyed¶

• <boolean>

Является true после вызова readable.destroy().

readable.isPaused¶

1

readable.isPaused();

• Возвращает: <boolean>

Метод readable.isPaused() возвращает текущее рабочее состояние Readable. Он используется в основном механизмом, который лежит в основе метода readable.pipe(). В большинстве типичных случаев нет причин использовать этот метод напрямую.

1
2
3
4
5
6
7

const readable = new stream.Readable();

readable.isPaused(); // === false
readable.pause();
readable.isPaused(); // === true
readable.resume();
readable.isPaused(); // === false

readable.pause¶

1

readable.pause();

• Возвращает: <this>

Метод readable.pause() заставит поток в режиме потока прекратить испускать события 'data', переходя из режима потока. Любые данные, которые становятся доступными, остаются во внутреннем буфере.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
    console.log(`Получено ${chunk.length} байт данных.`);
    readable.pause();
    console.log(
        'Дополнительных данных не будет в течение 1 секунды.'
    );
    setTimeout(() => {
        console.log(
            'Теперь данные начнут поступать снова.'
        );
        readable.resume();
    }, 1000);
});

Метод readable.pause() не имеет эффекта, если существует слушатель событий 'readable'.

readable.pipe¶

1

readable.pipe(destination[, options])

• destination <stream.Writable> Место назначения для записи данных
• options <Object> Опции трубы
  • end <boolean> Завершить запись при завершении чтения. По умолчанию: true.
• Возвращает: <stream.Writable> конечный пункт, позволяющий создавать цепочку труб, если это поток Duplex или Transform.

Метод readable.pipe() присоединяет поток Writable к readable, заставляя его автоматически переключаться в режим потока и передавать все свои данные в присоединенный Writable. Поток данных будет автоматически управляться таким образом, чтобы конечный поток Writable не был перегружен более быстрым потоком Readable.

В следующем примере все данные из readable передаются в файл с именем file.txt:

1
2
3
4
5

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// Все данные из readable попадают в 'file.txt'.
readable.pipe(writable);

Можно присоединить несколько потоков Writable к одному потоку Readable.

Метод readable.pipe() возвращает ссылку на поток назначения, что позволяет создавать цепочки потоков, передаваемых по трубопроводу:

1
2
3
4
5
6

const fs = require('node:fs');
const zlib = require('node:zlib');
const r = fs.createReadStream('file.txt');
const z = zlib.createGzip();
const w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w);

По умолчанию, stream.end() вызывается на конечном Writable потоке, когда исходный Readable поток испускает 'end', так что конечный поток больше не доступен для записи. Чтобы отключить это поведение по умолчанию, опцию end можно передать как false, в результате чего поток назначения останется открытым:

1
2
3
4

reader.pipe(writer, { end: false });
reader.on('end', () => {
    writer.end('Goodbye\n');
});

Важной оговоркой является то, что если поток Readable выдает ошибку во время обработки, направление Writable не закрывается автоматически. Если произойдет ошибка, необходимо будет ручно закрыть каждый поток, чтобы предотвратить утечку памяти.

Потоки process.stderr и process.stdout Writable никогда не закрываются до выхода процесса Node.js, независимо от указанных опций.

readable.read¶

1

readable.read([size]);

• size <number> Необязательный аргумент, указывающий, сколько данных нужно прочитать.
• Возвращает: <string> | <Buffer> | <null> | <any>

Метод readable.read() считывает данные из внутреннего буфера и возвращает их. Если данные не доступны для чтения, возвращается null. По умолчанию данные возвращаются в виде объекта Buffer, если только кодировка не была указана с помощью метода readable.setEncoding() или поток работает в объектном режиме.

Необязательный аргумент size задает определенное количество байт для чтения. Если size байт недоступен для чтения, будет возвращен null, если только поток не завершился, в этом случае будут возвращены все данные, оставшиеся во внутреннем буфере.

Если аргумент size не указан, будут возвращены все данные, содержащиеся во внутреннем буфере.

Аргумент size должен быть меньше или равен 1 GiB.

Метод readable.read() следует вызывать только на потоках Readable, работающих в приостановленном режиме. В потоковом режиме readable.read() вызывается автоматически, пока внутренний буфер не будет полностью опустошен.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

const readable = getReadableStreamSomehow();

// 'readable' может быть вызван несколько раз по мере буферизации данных
readable.on('readable', () => {
    let chunk;
    console.log(
        'Stream is readable (новые данные получены в буфер)'
    );
    // Используйте цикл, чтобы убедиться, что мы прочитали все доступные в данный момент данные
    while (null !== (chunk = readable.read())) {
        console.log(
            `Прочитано ${chunk.length} байт данных...`
        );
    }
});

// 'end' будет срабатывать один раз, когда больше не будет данных
readable.on('end', () => {
    console.log('Достигнут конец потока.');
});

Каждый вызов readable.read() возвращает фрагмент данных или null. Куски не конкатенируются. Цикл while необходим для потребления всех данных, находящихся в буфере. При чтении большого файла .read() может вернуть null, израсходовав все буферизованное содержимое, но есть еще больше данных, которые еще не буферизованы. В этом случае новое событие 'readable' будет выдано, когда в буфере будет больше данных. Наконец, событие 'end' будет вызвано, когда больше не будет данных.

Поэтому, чтобы прочитать все содержимое файла из readable, необходимо собирать фрагменты в несколько событий 'readable':

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

const chunks = [];

readable.on('readable', () => {
    let chunk;
    while (null !== (chunk = readable.read())) {
        chunks.push(chunk);
    }
});

readable.on('end', () => {
    const content = chunks.join('');
});

Поток Readable в объектном режиме всегда будет возвращать один элемент из вызова readable.read(size), независимо от значения аргумента size.

Если метод readable.read() возвращает фрагмент данных, также будет выдано событие 'data'.

Вызов stream.read([size]) после того, как было выдано событие 'end', вернет null. Никакой ошибки во время выполнения не возникнет.

readable.readable¶

• <boolean>

Является true, если безопасно вызывать readable.read(), что означает, что поток не был уничтожен или выдал 'error' или 'end'.

readable.readableAborted¶

• <boolean>

Возвращает, был ли поток уничтожен или ошибочен перед выдачей 'end'.

readable.readableDidRead¶

• <boolean>

Возвращает, были ли испущены данные.

readable.readableEncoding¶

• <null> | <string>

Получатель свойства encoding для данного потока Readable. Свойство encoding может быть установлено с помощью метода readable.setEncoding().

readable.readableEnded¶

• <boolean>

Становится true, когда испускается событие 'end'.

readable.errored¶

• <Error>

Возвращает ошибку, если поток был уничтожен с ошибкой.

readable.readableFlowing¶

• <boolean>

Это свойство отражает текущее состояние потока Readable, как описано в разделе Три состояния.

readable.readableHighWaterMark¶

• <number>

Возвращает значение highWaterMark, переданное при создании этого Readable.

readable.readableLength¶

• <number>

Это свойство содержит количество байтов (или объектов) в очереди, готовых к чтению. Значение предоставляет данные интроспекции относительно состояния highWaterMark.

readable.readableObjectMode¶

• <boolean>

Получатель для свойства objectMode данного потока Readable.

readable.resume¶

1

readable.resume();

• Возвращает: <this>

Метод readable.resume() заставляет явно приостановленный поток Readable возобновить испускание событий 'data', переводя поток в режим потока.

Метод readable.resume() можно использовать для полного потребления данных из потока без фактической обработки этих данных:

1
2
3
4
5

getReadableStreamSomehow()
    .resume()
    .on('end', () => {
        console.log('Достиг конца, но ничего не прочитал.');
    });

Метод readable.resume() не имеет эффекта, если существует слушатель события 'readable'.

readable.setEncoding¶

1

readable.setEncoding(encoding);

• encoding <string> Кодировка, которую следует использовать.
• Возвращает: <this>

Метод readable.setEncoding() устанавливает кодировку символов для данных, считываемых из потока Readable.

По умолчанию кодировка не задается, и данные потока будут возвращаться в виде объектов Buffer. Установка кодировки приводит к тому, что данные потока будут возвращаться в виде строк указанной кодировки, а не в виде объектов Buffer. Например, вызов readable.setEncoding('utf8') приведет к тому, что выходные данные будут интерпретированы как данные UTF-8 и переданы как строки. Вызов readable.setEncoding('hex') приведет к тому, что данные будут закодированы в шестнадцатеричном формате строк.

Поток Readable будет правильно обрабатывать многобайтовые символы, передаваемые через поток, которые в противном случае были бы неправильно декодированы, если бы просто извлекались из потока как объекты Buffer.

1
2
3
4
5
6
7
8
9

const readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', (chunk) => {
    assert.equal(typeof chunk, 'string');
    console.log(
        'Получено %d символов строковых данных:',
        chunk.length
    );
});

readable.unpipe¶

1

readable.unpipe([destination]);

• destination <stream.Writable> Необязательный конкретный поток для распайки
• Возвращает: <this>

Метод readable.unpipe() отсоединяет поток Writable, ранее присоединенный с помощью метода stream.pipe().

Если destination не указан, то отсоединяются все трубы.

Если назначение указано, но для него не установлена труба, то метод ничего не делает.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// Все данные из readable попадают в 'file.txt',
// но только в течение первой секунды.
readable.pipe(writable);
setTimeout(() => {
    console.log('Остановить запись в файл.txt.');
    readable.unpipe(writable);
    console.log('Вручную закрыть поток файлов.');
    writable.end();
}, 1000);

readable.unshift¶

1

readable.unshift(chunk[, encoding])

• chunk <Buffer> | <Uint8Array> | <string> | <null> | <any> Кусок данных для выгрузки в очередь чтения. Для потоков, не работающих в объектном режиме, chunk должен быть строкой, Buffer, Uint8Array или null. Для потоков, работающих в объектном режиме, chunk может быть любым значением JavaScript.
• encoding <string> Кодировка кусков строки. Должна быть правильной кодировкой Buffer, такой как 'utf8 или 'ascii.

Передача chunk как null сигнализирует о конце потока (EOF) и ведет себя так же, как readable.push(null), после чего данные больше не могут быть записаны. Сигнал EOF ставится в конце буфера, и все буферизованные данные все равно будут смыты.

Метод readable.unshift() выталкивает фрагмент данных обратно во внутренний буфер. Это полезно в некоторых ситуациях, когда поток потребляется кодом, которому нужно "отменить потребление" некоторого количества данных, которые он оптимистично извлек из источника, чтобы эти данные могли быть переданы другой стороне.

Метод stream.unshift(chunk) не может быть вызван после того, как произошло событие 'end', иначе будет выдана ошибка времени выполнения.

Разработчикам, часто использующим stream.unshift(), следует рассмотреть возможность перехода на использование потока Transform вместо этого. Дополнительную информацию смотрите в разделе "API для реализаторов потоков".

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// Вытаскиваем заголовок, разделенный \n\n.
// Используем unshift(), если получаем слишком много.
// Вызываем обратный вызов с (error, header, stream).
const { StringDecoder } = require('node:string_decoder');
function parseHeader(stream, callback) {
    stream.on('error', callback);
    stream.on('readable', onReadable);
    const decoder = new StringDecoder('utf8');
    let header = '';
    function onReadable() {
        let chunk;
        while (null !== (chunk = stream.read())) {
            const str = decoder.write(chunk);
            if (str.includes('\n\n')) {
                // Найдена граница заголовка.
                const split = str.split(/\n\n/);
                header += split.shift();
                const remaining = split.join('\n\n');
                const buf = Buffer.from(remaining, 'utf8');
                stream.removeListener('error', callback);
                // Удалите слушателя 'readable' перед разгруппировкой.
                stream.removeListener(
                    'readable',
                    onReadable
                );
                if (buf.length) stream.unshift(buf);
                // Теперь тело сообщения может быть прочитано из потока.
                callback(null, header, stream);
                return;
            }
            // Продолжаем читать заголовок.
            header += str;
        }
    }
}

В отличие от stream.push(chunk), stream.unshift(chunk) не завершает процесс чтения, сбрасывая внутреннее состояние потока. Это может привести к неожиданным результатам, если readable.unshift() вызывается во время чтения (например, из реализации stream._read() на пользовательском потоке). После вызова readable.unshift() с немедленным stream.push('') будет сброшен параметр

readable.wrap¶

1

readable.wrap(stream);

• stream <Stream> Читаемый поток "старого стиля"
• Возвращает: <this>

До версии Node.js 0.10 потоки не реализовывали весь API модуля node:stream, как он определен в настоящее время. (Более подробную информацию смотрите в "Совместимость").

При использовании старой библиотеки Node.js, которая испускает события 'data' и имеет метод stream.pause(), который является только рекомендательным, метод readable.wrap() можно использовать для создания потока Readable, который использует старый поток в качестве источника данных.

Использование readable.wrap() потребуется редко, но метод был предоставлен в качестве удобства для взаимодействия со старыми приложениями и библиотеками Node.js.

1
2
3
4
5
6
7
8

const { OldReader } = require('./old-api-module.js');
const { Readable } = require('node:stream');
const oreader = new OldReader();
const myReader = new Readable().wrap(oreader);

myReader.on('readable', () => {
    myReader.read(); // и т.д.
});

readable[Symbol.asyncIterator]¶

1

readable[Symbol.asyncIterator]();

• Возвращает: <AsyncIterator> для полного потребления потока.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

const fs = require('node:fs');

async function print(readable) {
    readable.setEncoding('utf8');
    let data = '';
    for await (const chunk of readable) {
        data += chunk;
    }
    console.log(data);
}

print(fs.createReadStream('file')).catch(console.error);

Если цикл завершится с break, return или throw, поток будет уничтожен. Другими словами, итерация над потоком будет полностью его потреблять. Поток будет считываться кусками размером, равным параметру highWaterMark. В приведенном выше примере данные будут в одном куске, если файл имеет размер менее 64 KiB, потому что опция highWaterMark не предоставляется в fs.createReadStream().

readable.compose¶

1

readable.compose(stream[, options])

• stream <Stream> | <Iterable> | <AsyncIterable> | <Function>
• options <Object>
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: {Duplex} поток, составленный с потоком stream.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import { Readable } from 'node:stream';

async function* splitToWords(source) {
    for await (const chunk of source) {
        const words = String(chunk).split(' ');

        for (const word of words) {
            yield word;
        }
    }
}

const wordsStream = Readable.from([
    'this is',
    'compose as operator',
]).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // печатает ['this', 'is', 'compose', 'as', 'operator']

Дополнительную информацию смотрите в stream.compose.

readable.iterator¶

1

readable.iterator([options]);

• options <Object>
  • destroyOnReturn <boolean> Если установлено значение false, вызов return на асинхронном итераторе или завершение итерации for await...of с помощью break, return или throw не будет уничтожать поток. По умолчанию: true.
• Возвращает: <AsyncIterator> для потребления потока.

Итератор, созданный этим методом, дает пользователям возможность отменить уничтожение потока, если цикл for await...of будет завершен return, break или throw, или если итератор должен уничтожить поток, если поток выдал ошибку во время итерации.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

const { Readable } = require('node:stream');

async function printIterator(readable) {
    for await (const chunk of readable.iterator({
        destroyOnReturn: false,
    })) {
        console.log(chunk); // 1
        break;
    }

    console.log(readable.destroyed); // false

    for await (const chunk of readable.iterator({
        destroyOnReturn: false,
    })) {
        console.log(chunk); // Будет выведено 2, а затем 3
    }

    console.log(readable.destroyed); // True, поток был полностью уничтожен
}

async function printSymbolAsyncIterator(readable) {
    for await (const chunk of readable) {
        console.log(chunk); // 1
        break;
    }

    console.log(readable.destroyed); // true
}

async function showBoth() {
    await printIterator(Readable.from([1, 2, 3]));
    await printSymbolAsyncIterator(
        Readable.from([1, 2, 3])
    );
}

showBoth();

readable.map¶

1

readable.map(fn[, options])

• fn <Function> | {AsyncFunction} функция для отображения каждого куска данных в потоке.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: {Readable} поток, отображенный с помощью функции fn.

Этот метод позволяет выполнять отображение над потоком. Функция fn будет вызываться для каждого чанка в потоке. Если функция fn возвращает обещание - это обещание будет ожидаться перед передачей в поток результатов.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// С синхронным маппером.
for await (const chunk of Readable.from([1, 2, 3, 4]).map(
    (x) => x * 2
)) {
    console.log(chunk); // 2, 4, 6, 8
}
// С асинхронным маппером, делая не более 2 запросов за раз.
const resolver = new Resolver();
const dnsResults = Readable.from([
    'nodejs.org',
    'openjsf.org',
    'www.linuxfoundation.org',
]).map((domain) => resolver.resolve4(domain), {
    concurrency: 2,
});
for await (const result of dnsResults) {
    console.log(result); // Выводит в журнал DNS-результат resolver.resolve4.
}

readable.filter¶

1

readable.filter(fn[, options])

• fn <Function> | {AsyncFunction} функция для фильтрации фрагментов из потока.
  • data <any> кусок данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: {Readable} поток, отфильтрованный с помощью предиката fn.

Этот метод позволяет фильтровать поток. Для каждого куска в потоке будет вызвана функция fn, и если она вернет истинное значение, то кусок будет передан в поток результатов. Если функция fn возвращает обещание - это обещание будет ожидаться.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// С синхронным предикатом.
for await (const chunk of Readable.from([
    1,
    2,
    3,
    4,
]).filter((x) => x > 2)) {
    console.log(chunk); // 3, 4
}
// С асинхронным предикатом, делая не более 2 запросов за раз.
const resolver = new Resolver();
const dnsResults = Readable.from([
    'nodejs.org',
    'openjsf.org',
    'www.linuxfoundation.org',
]).filter(
    async (domain) => {
        const { address } = await resolver.resolve4(
            domain,
            {
                ttl: true,
            }
        );
        return address.ttl > 60;
    },
    { concurrency: 2 }
);
for await (const result of dnsResults) {
    // Заносит в журнал домены с разрешенной dns-записью более 60 секунд.
    console.log(result);
}

readable.forEach¶

1

readable.forEach(fn[, options])

• fn <Function> | {AsyncFunction} функция для вызова на каждом фрагменте потока.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: <Promise> обещание о завершении потока.

Этот метод позволяет итерировать поток. Для каждого куска в потоке будет вызвана функция fn. Если функция fn возвращает обещание - это обещание будет await.

Этот метод отличается от циклов for await...of тем, что он может обрабатывать фрагменты одновременно. Кроме того, итерацию forEach можно остановить только передав опцию signal и прервав соответствующий AbortController, в то время как for await...of можно остановить с помощьюbreakилиreturn. В любом случае поток будет уничтожен.

Этот метод отличается от прослушивания события 'data' тем, что он использует событие readable в базовой машине и может ограничить количество одновременных вызовов fn.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// С синхронным предикатом.
for await (const chunk of Readable.from([
    1,
    2,
    3,
    4,
]).filter((x) => x > 2)) {
    console.log(chunk); // 3, 4
}
// С асинхронным предикатом, делая не более 2 запросов за раз.
const resolver = new Resolver();
const dnsResults = Readable.from([
    'nodejs.org',
    'openjsf.org',
    'www.linuxfoundation.org',
]).map(
    async (domain) => {
        const { address } = await resolver.resolve4(
            domain,
            {
                ttl: true,
            }
        );
        return address;
    },
    { concurrency: 2 }
);
await dnsResults.forEach((result) => {
    // Выводит результат в журнал, аналогично `for await (const result of dnsResults)`.
    console.log(result);
});
console.log('done'); // Поток завершен

readable.toArray¶

1

readable.toArray([options]);

• options <Object>
  • signal <AbortSignal> позволяет отменить операцию toArray, если сигнал прерван.
• Возвращает: <Promise> обещание, содержащее массив с содержимым потока.

Этот метод позволяет легко получить содержимое потока.

Поскольку этот метод считывает весь поток в память, он сводит на нет преимущества потоков. Он предназначен для совместимости и удобства, а не как основной способ потребления потоков.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

await Readable.from([1, 2, 3, 4]).toArray(); // [1, 2, 3, 4]

// Выполняем параллельные dns-запросы с помощью .map и собираем
// результаты в массив с помощью toArray
const dnsResults = await Readable.from([
    'nodejs.org',
    'openjsf.org',
    'www.linuxfoundation.org',
])
    .map(
        async (domain) => {
            const { address } = await resolver.resolve4(
                domain,
                {
                    ttl: true,
                }
            );
            return address;
        },
        { concurrency: 2 }
    )
    .toArray();

readable.some¶

1

readable.some(fn[, options])

• fn <Function> | {AsyncFunction} функция для вызова на каждом фрагменте потока.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: <Promise> обещание, оценивающее true, если fn вернул истинное значение хотя бы для одного из чанков.

Этот метод похож на Array.prototype.some и вызывает fn на каждом куске в потоке, пока ожидаемое возвращаемое значение не станет true (или любым истинным значением). Как только вызов fn на куске, ожидающем возврата значения, становится истинным, поток уничтожается и обещание выполняется с true. Если ни один из вызовов fn на чанках не возвращает истинное значение, обещание выполняется с false.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// С синхронным предикатом.
await Readable.from([1, 2, 3, 4]).some((x) => x > 2); // true
await Readable.from([1, 2, 3, 4]).some((x) => x < 0); // false

// С асинхронным предикатом, выполняющим не более 2 проверок файлов за раз.
const anyBigFile = await Readable.from([
    'file1',
    'file2',
    'file3',
]).some(
    async (fileName) => {
        const stats = await stat(fileName);
        return stats.size > 1024 * 1024;
    },
    { concurrency: 2 }
);
console.log(anyBigFile); // `true`, если любой файл в списке больше 1MB
console.log('done'); // Поток завершен

readable.find¶

1

readable.find(fn[, options])

• fn <Function> | {AsyncFunction} функция для вызова на каждом фрагменте потока.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: <Promise> обещание, оценивающее первый чанк, для которого fn имеет истинностное значение, или undefined, если элемент не был найден.

Этот метод похож на Array.prototype.find и вызывает fn на каждом куске в потоке, чтобы найти кусок с истинностным значением для fn. Как только ожидаемое возвращаемое значение вызова fn становится истинным, поток уничтожается, а обещание выполняется значением, для которого fn вернул истинное значение. Если все вызовы fn в чанках возвращают ложное значение, обещание выполняется с undefined.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// С синхронным предикатом.
await Readable.from([1, 2, 3, 4]).find((x) => x > 2); // 3
await Readable.from([1, 2, 3, 4]).find((x) => x > 0); // 1
await Readable.from([1, 2, 3, 4]).find((x) => x > 10); // неопределено

// С асинхронным предикатом, выполняющим не более 2 проверок файлов за раз.
const foundBigFile = await Readable.from([
    'file1',
    'file2',
    'file3',
]).find(
    async (fileName) => {
        const stats = await stat(fileName);
        return stats.size > 1024 * 1024;
    },
    { concurrency: 2 }
);
console.log(foundBigFile); // Имя файла большого файла, если какой-либо файл в списке больше 1MB
console.log('done'); // Поток завершен

readable.every¶

1

readable.every(fn[, options])

• fn <Function> | {AsyncFunction} функция для вызова на каждом куске потока.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: <Promise> обещание, оценивающее true, если fn вернул истинное значение для всех чанков.

Этот метод похож на Array.prototype.every и вызывает fn на каждом куске в потоке, чтобы проверить, являются ли все ожидаемые возвращаемые значения истинным значением для fn. Как только вызов fn на чанке, ожидающем возврата значения, оказывается ложным, поток уничтожается, а обещание выполняется с false. Если все вызовы fn на чанках возвращают истинное значение, обещание выполняется с true.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// С синхронным предикатом.
await Readable.from([1, 2, 3, 4]).every((x) => x > 2); // false
await Readable.from([1, 2, 3, 4]).every((x) => x > 0); // true

// С асинхронным предикатом, выполняющим не более 2 проверок файлов за раз.
const allBigFiles = await Readable.from([
    'file1',
    'file2',
    'file3',
]).every(
    async (fileName) => {
        const stats = await stat(fileName);
        return stats.size > 1024 * 1024;
    },
    { concurrency: 2 }
);
// `true`, если все файлы в списке больше 1MiB
console.log(allBigFiles);
console.log('done'); // Поток завершен

readable.flatMap¶

1

readable.flatMap(fn[, options])

• fn {Function|AsyncGeneratorFunction|AsyncFunction} функция для отображения каждого куска в потоке.
  • data <any> фрагмент данных из потока.
  • options <Object>
    • signal <AbortSignal> прерывается, если поток уничтожается, позволяя прервать вызов fn раньше времени.
• options <Object>
  • concurrency <number> максимальное количество одновременных вызовов fn для потока. По умолчанию: 1.
  • signal <AbortSignal> позволяет уничтожить поток, если сигнал прерван.
• Возвращает: {Readable} поток, отображенный с помощью функции fn.

Этот метод возвращает новый поток, применяя заданный обратный вызов к каждому фрагменту потока и затем сглаживая результат.

Можно вернуть поток или другую итерабельную или асинхронную итерабельную функцию из fn, и потоки результатов будут объединены (сплющены) в возвращаемый поток.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23