Readable interface

Пакет:: @azure/storage-common

Extends: Stream

Свойства

closed	Это `true` после `'close'` было излучено.
destroyed	Это `true` было `readable.destroy()` объявлено.
errored	Возвращает ошибку, если поток был уничтожен с ошибкой.
readable	Безопасно ли вызывать `true`, то есть поток не был уничтожен и не излучался`'error'`, или `'end'`.
readableAborted	Возвращает, был ли поток уничтожен или допущен ошибка до излучения `'end'`.
readableDidRead	Показывает, был `'data'` ли излучён.
readableEncoding	Getter для свойства `encoding` данного `Readable` потока. Свойство `encoding` можно задать с помощью метода setEncoding .
readableEnded	Становится `true` при `'end'` испускании события.
readableFlowing	Это свойство отражает текущее состояние `Readable` ручья, описанное в разделе «Три состояния ».
readableHighWaterMark	Возвращает значение `highWaterMark` переданных при создании этого `Readable`.
readableLength	Это свойство содержит количество байт (или объектов) в очереди, готовых к чтению. Значение предоставляет данные интроспекции относительно состояния `highWaterMark`.
readableObjectMode	Getter для свойства `objectMode` данного `Readable` потока.

Методы

addListener(string \| symbol, (args: any[]) => void)
addListener<E>(E, (args: ReadableEventMap[E]) => void)	Псевдоним для `emitter.on(eventName, listener)`.
compose(WritableStream \| WritableStream \| TransformStream \| (source: any) => void, Abortable)	`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(['text passed through', 'composed stream']).compose(splitToWords); const words = await wordsStream.toArray(); console.log(words); // prints ['text', 'passed', 'through', 'composed', 'stream']` `readable.compose(s)` эквивалентен `stream.compose(readable, s)`. Этот метод также позволяет предоставить , `AbortSignal` который уничтожает составленный поток при прекращении. См. `stream.compose(...streams)` для получения дополнительной информации.
destroy(Error)	Уничтожить поток. Опционально выпустить `'error'` событие и выпустить `'close'` событие (если `emitClose` только не установлено на `false`). После этого вызова читаемый поток освобождает любые внутренние ресурсы, и последующие вызовы `push()` будут игнорироваться. После `destroy()` вызова любые дальнейшие вызовы будут no-op, и никаких ошибок, кроме , `_destroy()` не может быть издана как `'error'`. Реализаторы не должны отменять этот метод, а вместо этого реализовывать `readable._destroy()`.
drop(number, Abortable)	Этот метод возвращает новый поток с первыми предельными блоками, отбрасывающимися с самого начала.
emit(string \| symbol, any[])
emit<E>(E, ReadableEventMap[E])	Синхронно вызывает каждый из прослушивателей, зарегистрированных для события с именем `eventName`, в том порядке, в который они были зарегистрированы, передавая предоставленные аргументы каждому из них. Возвращает `true`, если событие имело прослушиватели, `false` в противном случае. import { EventEmitter } from 'node:events'; const myEmitter = new EventEmitter(); // First listener myEmitter.on('event', function firstListener() { console.log('Helloooo! first listener'); }); // Second listener myEmitter.on('event', function secondListener(arg1, arg2) { console.log(`event with parameters ${arg1}, ${arg2} in second listener`); }); // Third listener myEmitter.on('event', function thirdListener(...args) { const parameters = args.join(', '); console.log(`event with parameters ${parameters} in third listener`); }); console.log(myEmitter.listeners('event')); myEmitter.emit('event', 1, 2, 3, 4, 5); // Prints: // [ // [Function: firstListener], // [Function: secondListener], // [Function: thirdListener] // ] // Helloooo! first listener // event with parameters 1, 2 in second listener // event with parameters 1, 2, 3, 4, 5 in third listener
every((data: any, options?: Abortable) => boolean \| Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	Этот метод похож на `Array.prototype.every` и вызывает fn на каждом чанке потока, чтобы проверить, являются ли все ожидаемые значения возврата истинными для fn. Как только fn-вызов на `await`chunk ed return-значение ложно, поток уничтожается, и обещание выполняется с `false`. Если все fn-запросы на чанки возвращают истинное значение, обещание выполняется с `true`.
filter((data: any, options?: Abortable) => boolean \| Promise<boolean>, ReadableOperatorOptions)	Этот метод позволяет фильтровать поток. Для каждого блока в потоке вызывается функция fn , и если она возвращает истинное значение, этот блок передаётся в поток результата. Если функция fn возвращает обещание — это обещание будет `await`ed.
find((data: any, options?: Abortable) => boolean \| Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)
find<T>((data: any, options?: Abortable) => data is T, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	Этот метод похож на `Array.prototype.find` и вызывает fn на каждом чанке потока, чтобы найти блок с истинным значением fn. Как только ожидаемое возвратное значение fn-звонка становится истинным, поток уничтожается, и обещание исполняется со значением, для которого fn возвращает истинное значение. Если все fn-запросы на блоках возвращают ложное значение, обещание выполняется с .`undefined`
flatMap((data: any, options?: Abortable) => any, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	Этот метод возвращает новый поток, применяя данный callback к каждому блоку потока и затем выравнивая результат. Возможно вернуть поток или другую итерируемую или асинхронную итерируемую из fn , и итоговые потоки будут объединены (плоски) с возвращённым потоком.
forEach((data: any, options?: Abortable) => void \| Promise<void>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	Этот метод позволяет итерировать поток. Для каждого блока в потоке будет вызвана функция fn . Если функция fn возвращает обещание — это обещание будет `await`ed. Этот метод отличается от `for await...of` циклов тем, что может по желанию одновременно обрабатывать чанки. Кроме того, `forEach` итерацию можно остановить только после прохождения опции `signal` и прерывания соответствующего AbortController, пока `for await...of` можно остановить с `break` помощью или `return`. В любом случае поток будет уничтожен. Этот метод отличается от прослушивания `'data'` события тем, что использует событие `readable` в базовой машине и может ограничить количество одновременных вызовов fn .
isPaused()	Метод возвращает `readable.isPaused()` текущее рабочее состояние `Readable`. Это используется в первую очередь механизмом, лежащим в основе метода `readable.pipe()` . В большинстве случаев нет смысла использовать этот метод напрямую. `const readable = new stream.Readable(); readable.isPaused(); // === false readable.pause(); readable.isPaused(); // === true readable.resume(); readable.isPaused(); // === false`
iterator(ReadableIteratorOptions)	Итератор, созданный этим методом, даёт пользователям возможность отменить разрушение потока, если `for await...of` цикл выходит на `return`, `break`, или `throw`, или если итератор должен уничтожить поток, если поток издал ошибку во время итерации.
listenerCount(string \| symbol, (args: any[]) => void)
listenerCount<E>(E, (args: ReadableEventMap[E]) => void)	Возвращает число прослушивателей, прослушивающих событие с именем `eventName`. Если указана `listener`, он возвращает количество раз, когда прослушиватель найден в списке прослушивателей события.
listeners(string \| symbol)
listeners<E>(E)	Возвращает копию массива прослушивателей для события с именем `eventName`. `server.on('connection', (stream) => { console.log('someone connected!'); }); console.log(util.inspect(server.listeners('connection'))); // Prints: [ [Function] ]`
map((data: any, options?: Abortable) => any, ReadableOperatorOptions)	Этот метод позволяет отображать по потоку. Функция fn будет вызываться для каждого чанка в потоке. Если функция fn возвращает обещание — это обещание будет `await`ed до передачи в поток результатов.
off(string \| symbol, (args: any[]) => void)
off<E>(E, (args: ReadableEventMap[E]) => void)	Псевдоним для `emitter.removeListener()`.
on(string \| symbol, (args: any[]) => void)
on<E>(E, (args: ReadableEventMap[E]) => void)	Добавляет функцию `listener` в конец массива прослушивателей для события с именем `eventName`. Проверка не выполняется, чтобы узнать, добавлена ли `listener`. Несколько вызовов, проходящих одно и то же сочетание `eventName` и `listener`, приведет к добавлению `listener` и вызову несколько раз. `server.on('connection', (stream) => { console.log('someone connected!'); });` Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны. По умолчанию прослушиватели событий вызываются в том порядке, в который они добавляются. Метод `emitter.prependListener()` можно использовать в качестве альтернативы для добавления прослушивателя событий в начало массива прослушивателей. `import { EventEmitter } from 'node:events'; const myEE = new EventEmitter(); myEE.on('foo', () => console.log('a')); myEE.prependListener('foo', () => console.log('b')); myEE.emit('foo'); // Prints: // b // a`
once(string \| symbol, (args: any[]) => void)
once<E>(E, (args: ReadableEventMap[E]) => void)	Добавляет функцию однократного для события с именем . При следующем запуске `eventName` этот прослушиватель удаляется и вызывается. `server.once('connection', (stream) => { console.log('Ah, we have our first user!'); });` Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны. По умолчанию прослушиватели событий вызываются в том порядке, в который они добавляются. Метод `emitter.prependOnceListener()` можно использовать в качестве альтернативы для добавления прослушивателя событий в начало массива прослушивателей. `import { EventEmitter } from 'node:events'; const myEE = new EventEmitter(); myEE.once('foo', () => console.log('a')); myEE.prependOnceListener('foo', () => console.log('b')); myEE.emit('foo'); // Prints: // b // a`
pause()	Этот `readable.pause()` метод заставляет поток в режиме потока прекратить `'data'` выпускать события, переключаясь из режима потока. Любые доступные данные останутся во внутреннем буфере. const readable = getReadableStreamSomehow(); readable.on('data', (chunk) => { console.log(`Received ${chunk.length} bytes of data.`); readable.pause(); console.log('There will be no additional data for 1 second.'); setTimeout(() => { console.log('Now data will start flowing again.'); readable.resume(); }, 1000); }); Метод `readable.pause()` не действует, если есть `'readable'` слушатель события.
prependListener(string \| symbol, (args: any[]) => void)
prependListener<E>(E, (args: ReadableEventMap[E]) => void)	Добавляет функцию `listener` в начала массива прослушивателей для события с именем `eventName`. Проверка не выполняется, чтобы узнать, добавлена ли `listener`. Несколько вызовов, проходящих одно и то же сочетание `eventName` и `listener`, приведет к добавлению `listener` и вызову несколько раз. `server.prependListener('connection', (stream) => { console.log('someone connected!'); });` Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны.
prependOnceListener(string \| symbol, (args: any[]) => void)
prependOnceListener<E>(E, (args: ReadableEventMap[E]) => void)	Добавляет функцию однократного для события с именем в начиная массива прослушивателей. При следующем запуске `eventName` этот прослушиватель удаляется, а затем вызывается. `server.prependOnceListener('connection', (stream) => { console.log('Ah, we have our first user!'); });` Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны.
push(any, BufferEncoding)
rawListeners(string \| symbol)
rawListeners<E>(E)	Возвращает копию массива прослушивателей для события с именем `eventName`, включая любые оболочки (например, созданные `.once()`). import { EventEmitter } from 'node:events'; const emitter = new EventEmitter(); emitter.once('log', () => console.log('log once')); // Returns a new Array with a function `onceWrapper` which has a property // `listener` which contains the original listener bound above const listeners = emitter.rawListeners('log'); const logFnWrapper = listeners[0]; // Logs "log once" to the console and does not unbind the `once` event logFnWrapper.listener(); // Logs "log once" to the console and removes the listener logFnWrapper(); emitter.on('log', () => console.log('log persistently')); // Will return a new Array with a single function bound by `.on()` above const newListeners = emitter.rawListeners('log'); // Logs "log persistently" twice newListeners[0](); emitter.emit('log');
read(number)	`readable.read()` Метод считывает данные из внутреннего буфера и возвращает их. Если данные для чтения отсутствуют, `null` возвращается. По умолчанию данные возвращаются как `Buffer` объект, если только кодировка не была задана с помощью `readable.setEncoding()` метода или поток не работает в объектном режиме. Необязательный `size` аргумент указывает определённое количество байт для чтения. Если `size` байты недоступны для чтения, `null` они возвращаются, если поток не завершился, тогда все данные, оставшиеся во внутреннем буфере, будут возвращены. Если аргумент `size` не указан, все данные, содержащиеся во внутреннем буфере, будут возвращены. Аргумент `size` должен быть меньше или равен 1 ГиБ. Метод должен вызываться только на `readable.read()` потоках, `Readable` работающих в паузированном режиме. В режиме потока `readable.read()` вызывается автоматически, пока внутренний буфер не будет полностью осушен. const readable = getReadableStreamSomehow(); // 'readable' may be triggered multiple times as data is buffered in readable.on('readable', () => { let chunk; console.log('Stream is readable (new data received in buffer)'); // Use a loop to make sure we read all currently available data while (null !== (chunk = readable.read())) { console.log(`Read ${chunk.length} bytes of data...`); } }); // 'end' will be triggered once when there is no more data available readable.on('end', () => { console.log('Reached end of stream.'); }); Каждый вызов возвращает `readable.read()` фрагмент данных, или `null`. Куски не соединяются. Цикл необходим для поглощения всех данных, `while` находящихся в буфере. При чтении большого файла `.read()` может вернуться `null`, уже поглотив весь буферизованный контент, но ещё больше данных, которые ещё не буферизированы. В этом случае новое `'readable'` событие возникает, когда в буфере появляется больше данных. В конце `'end'` событий событие будет опубликовано, когда не останется новых данных. Поэтому, чтобы прочитать весь содержимое файла из `readable`, необходимо собрать чанки между несколькими `'readable'` событиями: `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'` событие также будет излучаться. Вызов прочитан после передачи `'end'` события вернётся `null`. Ошибка во время выполнения не возникает.
reduce<T>((previous: any, data: any, options?: Abortable) => T)	Этот метод вызывает fn на каждом участке потока в порядке, передавая ему результат вычисления по предыдущему элементу. Она возвращает обещание на итоговое значение редукции. Если начальное значение не предоставляется, в качестве начального значения используется первый кусок потока. Если поток пуст, обещание отклоняется с помощью `TypeError` свойства кода`ERR_INVALID_ARGS`. Функция редуктора повторяет поток по элементам, что означает, что параметр параллелизма или параллелизма отсутствует. Чтобы выполнить параллельное сокращение, можно извлечь метод асинхронной функции to `readable.map` .
reduce<T>((previous: T, data: any, options?: Abortable) => T, T, Abortable)
removeAllListeners(string \| symbol)
removeAllListeners<E>(E)	Удаляет все прослушиватели или те из указанных `eventName`. Не рекомендуется удалять прослушиватели, добавленные в другое место в коде, особенно если экземпляр `EventEmitter` был создан другим компонентом или модулем (например, сокетами или потоками файлов). Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны.
removeListener(string \| symbol, (args: any[]) => void)
removeListener<E>(E, (args: ReadableEventMap[E]) => void)	Удаляет указанный `listener` из массива прослушивателя для события с именем `eventName`. `const callback = (stream) => { console.log('someone connected!'); }; server.on('connection', callback); // ... server.removeListener('connection', callback);` `removeListener()` удаляет в большинстве случаев один экземпляр прослушивателя из массива прослушивателя. Если один прослушиватель был добавлен несколько раз в массив прослушивателя для указанного `eventName`, `removeListener()` необходимо вызывать несколько раз, чтобы удалить каждый экземпляр. После создания события все прослушиватели, подключенные к нему во время создания, вызываются в порядке. Это означает, что любые `removeListener()` вызовы `removeAllListeners()`OR после эмитирования и до завершения выполнения последним слушателем не удалят их из `emit()` процесса. Последующие события ведут себя должным образом. import { EventEmitter } from 'node:events'; class MyEmitter extends EventEmitter {} const myEmitter = new MyEmitter(); const callbackA = () => { console.log('A'); myEmitter.removeListener('event', callbackB); }; const callbackB = () => { console.log('B'); }; myEmitter.on('event', callbackA); myEmitter.on('event', callbackB); // callbackA removes listener callbackB but it will still be called. // Internal listener array at time of emit [callbackA, callbackB] myEmitter.emit('event'); // Prints: // A // B // callbackB is now removed. // Internal listener array [callbackA] myEmitter.emit('event'); // Prints: // A Поскольку слушатели управляются с помощью внутреннего массива, вызов этого параметра изменяет индексы позиций любого слушателя, зарегистрированного после удаления слушателя. Это не повлияет на порядок вызова прослушивателей, но это означает, что все копии массива прослушивателя, возвращенные методом `emitter.listeners()`, должны быть воссозданы. Если одна функция была добавлена в обработчик несколько раз для одного события (как в примере ниже), `removeListener()` удалит последний добавленный экземпляр. В примере прослушивателя `once('ping')` удаляется: `import { EventEmitter } from 'node:events'; const ee = new EventEmitter(); function pong() { console.log('pong'); } ee.on('ping', pong); ee.once('ping', pong); ee.removeListener('ping', pong); ee.emit('ping'); ee.emit('ping');` Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны.
resume()	Этот `readable.resume()` метод заставляет явно приостановленный `Readable` поток возобновить события излучения `'data'` , переключая поток в режим потока. Этот `readable.resume()` метод можно использовать для полного потребления данных из потока без фактической обработки этих данных: `getReadableStreamSomehow() .resume() .on('end', () => { console.log('Reached the end, but did not read anything.'); });` Метод `readable.resume()` не действует, если есть `'readable'` слушатель события.
setEncoding(BufferEncoding)	Метод задаёт кодировку символов для данных, читаемых `readable.setEncoding()` из потока `Readable` . По умолчанию кодировка не назначается, и данные потока возвращаются в виде `Buffer` объектов. Установка кодировки приводит к тому, что данные потока возвращаются в виде строк указанного кодирования, а не как `Buffer` объекты. Например, вызов `readable.setEncoding('utf8')` приводит к интерпретации выходных данных как UTF-8 и передачи в виде строк. Вызов `readable.setEncoding('hex')` приводит к тому, что данные будут закодированы в шестнадцатеричном строчном формате. Поток корректно обрабатывает многобайтовые символы, `Readable` передаваемые через поток, которые в противном случае были бы некорректно декодированы, если просто извлечь из потока как `Buffer` объекты. `const readable = getReadableStreamSomehow(); readable.setEncoding('utf8'); readable.on('data', (chunk) => { assert.equal(typeof chunk, 'string'); console.log('Got %d characters of string data:', chunk.length); });`
some((data: any, options?: Abortable) => boolean \| Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	Этот метод похож на `Array.prototype.some` и вызывает fn на каждом чанке потока до тех пор, пока ожидаемое возвращаемое значение не достигнет (или любого истинного значения).`true` Как только fn-вызов на chunk `await`ed возвращает значение true, поток уничтожается, и обещание исполняется с `true`. Если ни один из fn-призывов к чанкам не возвращает истинное значение, обещание выполняется с `false`.
take(number, Abortable)	Этот метод возвращает новый поток с первыми предельными чанками.
toArray(Abortable)	Этот метод позволяет легко получать содержимое потока. Поскольку этот метод считывает весь поток в память, он нивелирует преимущества потоков. Он предназначен для совместимости и удобства, а не как основной способ потребления потоков.
unpipe(WritableStream)	Метод отсоединяет ручей`readable.unpipe()`, `Writable` ранее присоединённый с помощью метода трубы. Если не указано `destination` , то все трубы отсоединяются. Если указано, `destination` но труба для неё не установлена, то метод ничего не делает. `import fs from 'node:fs'; const readable = getReadableStreamSomehow(); const writable = fs.createWriteStream('file.txt'); // All the data from readable goes into 'file.txt', // but only for the first second. readable.pipe(writable); setTimeout(() => { console.log('Stop writing to file.txt.'); readable.unpipe(writable); console.log('Manually close the file stream.'); writable.end(); }, 1000);`
unshift(any, BufferEncoding)	Передача `chunk` в `null` виде сигнализирует конец потока (EOF) и ведёт себя так же, как `readable.push(null)`, после чего больше данные не могут быть записаны. Сигнал EOF размещается в конце буфера, и любые буферированные данные всё равно будут промыты. Метод `readable.unshift()` возвращает часть данных обратно во внутренний буфер. Это полезно в определённых ситуациях, когда поток поглощается кодом, который должен «перепотребить» часть данных, оптимистично извлеченных из источника, чтобы данные могли быть переданы другой стороне. `stream.unshift(chunk)` Метод нельзя вызвать после выпуска `'end'` события, иначе будет сброшена ошибка во время выполнения. Разработчикам, использующим `stream.unshift()` частую пользу, стоит рассмотреть возможность перехода на `Transform` поток. См. `API for stream implementers` раздел для получения дополнительной информации. // Pull off a header delimited by \n\n. // Use unshift() if we get too much. // Call the callback with (error, header, stream). import { StringDecoder } from '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')) { // Found the header boundary. 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); // Remove the 'readable' listener before unshifting. stream.removeListener('readable', onReadable); if (buf.length) stream.unshift(buf); // Now the body of the message can be read from the stream. callback(null, header, stream); return; } // Still reading the header. header += str; } } } В отличие от push, `stream.unshift(chunk)` процесс чтения не завершается сбросом внутреннего состояния потока. Это может привести к неожиданным результатам, если `readable.unshift()` вызов вызывается во время чтения (то есть внутри реализации _read на пользовательском потоке). Следуя за вызовом To `readable.unshift()` с помощью немедленного натиска , состояние считывания сбросит правильно, однако лучше просто избегать вызова `readable.unshift()` во время выполнения чтения.
wrap(ReadableStream)	До Node.js 0.10 потоки не реализовывали весь `node:stream` API модуля в нынешнем виде. (См. `Compatibility` больше информации.) При использовании старой Node.js-библиотеки, которая генерирует `'data'` события и имеет метод паузы , который является только рекомендацией, этот `readable.wrap()` метод может быть использован для создания `Readable` потока, использующего старый поток в качестве источника данных. Это редко потребуется `readable.wrap()` использовать, но этот метод был предоставлен для удобства взаимодействия со старыми Node.js приложениями и библиотеками. `import { OldReader } from './old-api-module.js'; import { Readable } from 'node:stream'; const oreader = new OldReader(); const myReader = new Readable().wrap(oreader); myReader.on('readable', () => { myReader.read(); // etc. });`
[asyncDispose]()	Звонит `readable.destroy()` с кнопкой `AbortError` и возвращает обещание, которое выполняется после завершения стрима.
[asyncIterator]()
[toAsyncStreamable]()	Когда `--experimental-stream-iter` флаг включен, `Readable` потоки реализуют протокол `Stream.toAsyncStreamable` , обеспечивая эффективное использование API `stream/iter` . Это обеспечивает пакетный асинхронный итератор, который сбрасывает внутренний буфер потока на `Uint8Array[]` партии, амортизируя накладные расходы Promise на каждый чанк стандартного `Symbol.asyncIterator` пути. Для потоков в байтовом режиме блоки приводятся напрямую в виде `Buffer` экземпляров (которые являются `Uint8Array` подклассами). Для потоков в объектном режиме или кодирования каждый блок нормализуется до `Uint8Array` пакетирования. Возвращаемый итератор помечен как проверенный источник, поэтому `from()` пропускает его без дополнительной нормализации. `import { Readable } from 'node:stream'; import { text, from } from 'node:stream/iter'; const readable = new Readable({ read() { this.push('hello'); this.push(null); }, }); // Readable is automatically consumed via toAsyncStreamable console.log(await text(from(readable))); // 'hello'` Без `--experimental-stream-iter` флага вызов этого метода приводит `ERR_STREAM_ITER_MISSING_FLAG`к .

Наследуемые методы

eventNames()	Возвращает массив, в котором перечислены события, для которых у поставщика зарегистрированы прослушиватели. `import { EventEmitter } from 'node:events'; const myEE = new EventEmitter(); myEE.on('foo', () => {}); myEE.on('bar', () => {}); const sym = Symbol('symbol'); myEE.on(sym, () => {}); console.log(myEE.eventNames()); // Prints: [ 'foo', 'bar', Symbol(symbol) ]`
getMaxListeners()	Возвращает текущее значение максимального прослушивателя для `EventEmitter`, заданное `emitter.setMaxListeners(n)` или значением по умолчанию `events.defaultMaxListeners`.
pipe<T>(T, PipeOptions)
setMaxListeners(number)	По умолчанию `EventEmitter`будет выводить предупреждение, если для определенного события добавляются более `10` прослушивателей. Это полезное значение по умолчанию, которое помогает находить утечки памяти. Метод `emitter.setMaxListeners()` позволяет изменить ограничение для этого конкретного экземпляра `EventEmitter`. Для указания неограниченного количества прослушивателей можно задать значение `Infinity` (или `0`). Возвращает ссылку на `EventEmitter`, чтобы вызовы могли быть связаны.
[captureRejectionSymbol](Error, string \| symbol, any[])	Метод `Symbol.for('nodejs.rejection')` вызывается в случае, если при выпуске события происходит отказ обещания и `captureRejections` активируется на эмиттере. Возможно использовать `events.captureRejectionSymbol` вместо `Symbol.for('nodejs.rejection')`. `import { EventEmitter, captureRejectionSymbol } from 'node:events'; class MyClass extends EventEmitter { constructor() { super({ captureRejections: true }); } [captureRejectionSymbol](err, event, ...args) { console.log('rejection happened for', event, 'with', err, ...args); this.destroy(err); } destroy(err) { // Tear the resource down here. } }`

Сведения о свойстве

closed

Это true после 'close' было излучено.

closed: boolean

Значение свойства

boolean

destroyed

Это true было readable.destroy() объявлено.

destroyed: boolean

Значение свойства

boolean

errored

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

errored: Error | null

Значение свойства

Error | null

readable

Безопасно ли вызывать true, то есть поток не был уничтожен и не излучался'error', или 'end'.

readable: boolean

Значение свойства

boolean

readableAborted

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

readableAborted: boolean

Значение свойства

boolean

readableDidRead

Показывает, был 'data' ли излучён.

readableDidRead: boolean

Значение свойства

boolean

readableEncoding

Getter для свойства encoding данного Readable потока. Свойство encoding можно задать с помощью метода setEncoding .

readableEncoding: BufferEncoding | null

Значение свойства

BufferEncoding | null

readableEnded

Становится true при 'end' испускании события.

readableEnded: boolean

Значение свойства

boolean

readableFlowing

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

readableFlowing: boolean | null

Значение свойства

boolean | null

readableHighWaterMark

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

readableHighWaterMark: number

Значение свойства

number

readableLength

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

readableLength: number

Значение свойства

number

readableObjectMode

Getter для свойства objectMode данного Readable потока.

readableObjectMode: boolean

Значение свойства

boolean

Сведения о методе

addListener(string | symbol, (args: any[]) => void)

function addListener(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

addListener<E>(E, (args: ReadableEventMap[E]) => void)

Псевдоним для emitter.on(eventName, listener).

function addListener<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

listener: (args: ReadableEventMap[E]) => void

Возвращаемое значение

this

compose(WritableStream | WritableStream | TransformStream | (source: any) => void, Abortable)

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(['text passed through', 'composed stream']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['text', 'passed', 'through', 'composed', 'stream']

readable.compose(s) эквивалентен stream.compose(readable, s).

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

См. stream.compose(...streams) для получения дополнительной информации.

function compose(stream: WritableStream | WritableStream | TransformStream | (source: any) => void, options?: Abortable): Duplex

Параметры

stream: WritableStream | WritableStream | TransformStream | (source: any) => void

options: Abortable

Возвращаемое значение

Duplex

ручей, составленный из потока stream.

destroy(Error)

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

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

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

function destroy(error?: Error): this

Параметры

error: Error

Ошибка, которая будет передана в виде полезной нагрузки в 'error' случае события

Возвращаемое значение

this

drop(number, Abortable)

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

function drop(limit: number, options?: Abortable): Readable

Параметры

limit: number

Количество чанков, которые можно выпасть из читаемого.

options: Abortable

Возвращаемое значение

Readable

поток с лимитными чанками, выпавшими с самого начала.

emit(string | symbol, any[])

function emit(eventName: string | symbol, args: any[]): boolean

Параметры

eventName: string | symbol

args: any[]

Возвращаемое значение

boolean

emit<E>(E, ReadableEventMap[E])

Синхронно вызывает каждый из прослушивателей, зарегистрированных для события с именем eventName, в том порядке, в который они были зарегистрированы, передавая предоставленные аргументы каждому из них.

Возвращает true, если событие имело прослушиватели, false в противном случае.

import { EventEmitter } from 'node:events';
const myEmitter = new EventEmitter();

// First listener
myEmitter.on('event', function firstListener() {
  console.log('Helloooo! first listener');
});
// Second listener
myEmitter.on('event', function secondListener(arg1, arg2) {
  console.log(`event with parameters ${arg1}, ${arg2} in second listener`);
});
// Third listener
myEmitter.on('event', function thirdListener(...args) {
  const parameters = args.join(', ');
  console.log(`event with parameters ${parameters} in third listener`);
});

console.log(myEmitter.listeners('event'));

myEmitter.emit('event', 1, 2, 3, 4, 5);

// Prints:
// [
//   [Function: firstListener],
//   [Function: secondListener],
//   [Function: thirdListener]
// ]
// Helloooo! first listener
// event with parameters 1, 2 in second listener
// event with parameters 1, 2, 3, 4, 5 in third listener

function emit<E>(eventName: E, args: ReadableEventMap[E]): boolean

Параметры

eventName: E

args: ReadableEventMap[E]

Возвращаемое значение

boolean

every((data: any, options?: Abortable) => boolean | Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

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

function every(fn: (data: any, options?: Abortable) => boolean | Promise<boolean>, options?: Pick<ReadableOperatorOptions, "concurrency" | "signal">): Promise<boolean>

Параметры

fn: (data: any, options?: Abortable) => boolean | Promise<boolean>

функция для вызова каждого блока потока. Асинхронно или нет.

options: Pick<ReadableOperatorOptions, "concurrency" | "signal">

Возвращаемое значение

Promise<boolean>

Обещание, вычисляющее для true Если FN возвращает истинное значение для каждого из чанков.

filter((data: any, options?: Abortable) => boolean | Promise<boolean>, ReadableOperatorOptions)

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

function filter(fn: (data: any, options?: Abortable) => boolean | Promise<boolean>, options?: ReadableOperatorOptions): Readable

Параметры

fn: (data: any, options?: Abortable) => boolean | Promise<boolean>

функция фильтрации чанков из потока. Асинхронно или нет.

options: ReadableOperatorOptions

Возвращаемое значение

Readable

поток, отфильтрованный предикатом fn.

find((data: any, options?: Abortable) => boolean | Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

function find(fn: (data: any, options?: Abortable) => boolean | Promise<boolean>, options?: Pick<ReadableOperatorOptions, "concurrency" | "signal">): Promise<any>

Параметры

fn: (data: any, options?: Abortable) => boolean | Promise<boolean>

options: Pick<ReadableOperatorOptions, "concurrency" | "signal">

Возвращаемое значение

Promise<any>

find<T>((data: any, options?: Abortable) => data is T, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

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

function find<T>(fn: (data: any, options?: Abortable) => data is T, options?: Pick<ReadableOperatorOptions, "concurrency" | "signal">): Promise<T | undefined>

Параметры

fn: (data: any, options?: Abortable) => data is T

функция для вызова каждого блока потока. Асинхронно или нет.

options: Pick<ReadableOperatorOptions, "concurrency" | "signal">

Возвращаемое значение

Promise<T | undefined>

Обещание, вычисляющее для первого блока, для которого fn оценивается с истинным значением, или undefined если элемент не найден.

flatMap((data: any, options?: Abortable) => any, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

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

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

function flatMap(fn: (data: any, options?: Abortable) => any, options?: Pick<ReadableOperatorOptions, "concurrency" | "signal">): Readable

Параметры

fn: (data: any, options?: Abortable) => any

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

options: Pick<ReadableOperatorOptions, "concurrency" | "signal">

Возвращаемое значение

Readable

поток с плоским отображением с функцией fn.

forEach((data: any, options?: Abortable) => void | Promise<void>, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

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

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

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

function forEach(fn: (data: any, options?: Abortable) => void | Promise<void>, options?: Pick<ReadableOperatorOptions, "concurrency" | "signal">): Promise<void>

Параметры

fn: (data: any, options?: Abortable) => void | Promise<void>

функция для вызова каждого блока потока. Асинхронно или нет.

options: Pick<ReadableOperatorOptions, "concurrency" | "signal">

Возвращаемое значение

Promise<void>

обещание на то, что стрим закончится.

isPaused()

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

const readable = new stream.Readable();

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

function isPaused(): boolean

Возвращаемое значение

boolean

iterator(ReadableIteratorOptions)

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

function iterator(options?: ReadableIteratorOptions): AsyncIterator<any>

Параметры

options: ReadableIteratorOptions

Возвращаемое значение

AsyncIterator<any>

listenerCount(string | symbol, (args: any[]) => void)

function listenerCount(eventName: string | symbol, listener?: (args: any[]) => void): number

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

number

listenerCount<E>(E, (args: ReadableEventMap[E]) => void)

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

function listenerCount<E>(eventName: E, listener?: (args: ReadableEventMap[E]) => void): number

Параметры

eventName: E

Имя прослушиваемого события

listener: (args: ReadableEventMap[E]) => void

Функция обработчика событий

Возвращаемое значение

number

listeners(string | symbol)

function listeners(eventName: string | symbol): (args: any[]) => void[]

Параметры

eventName: string | symbol

Возвращаемое значение

(args: any[]) => void[]

listeners<E>(E)

Возвращает копию массива прослушивателей для события с именем eventName.

server.on('connection', (stream) => {
  console.log('someone connected!');
});
console.log(util.inspect(server.listeners('connection')));
// Prints: [ [Function] ]

function listeners<E>(eventName: E): (args: ReadableEventMap[E]) => void[]

Параметры

eventName: E

Возвращаемое значение

(args: ReadableEventMap[E]) => void[]

map((data: any, options?: Abortable) => any, ReadableOperatorOptions)

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

function map(fn: (data: any, options?: Abortable) => any, options?: ReadableOperatorOptions): Readable

Параметры

fn: (data: any, options?: Abortable) => any

функция для отображения по каждому чанку в потоке. Асинхронно или нет.

options: ReadableOperatorOptions

Возвращаемое значение

Readable

поток, отображаемый с функцией fn.

off(string | symbol, (args: any[]) => void)

function off(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

off<E>(E, (args: ReadableEventMap[E]) => void)

Псевдоним для emitter.removeListener().

function off<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

listener: (args: ReadableEventMap[E]) => void

Возвращаемое значение

this

on(string | symbol, (args: any[]) => void)

function on(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

on<E>(E, (args: ReadableEventMap[E]) => void)

Добавляет функцию listener в конец массива прослушивателей для события с именем eventName. Проверка не выполняется, чтобы узнать, добавлена ли listener. Несколько вызовов, проходящих одно и то же сочетание eventName и listener, приведет к добавлению listener и вызову несколько раз.

server.on('connection', (stream) => {
  console.log('someone connected!');
});

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

По умолчанию прослушиватели событий вызываются в том порядке, в который они добавляются. Метод emitter.prependListener() можно использовать в качестве альтернативы для добавления прослушивателя событий в начало массива прослушивателей.

import { EventEmitter } from 'node:events';
const myEE = new EventEmitter();
myEE.on('foo', () => console.log('a'));
myEE.prependListener('foo', () => console.log('b'));
myEE.emit('foo');
// Prints:
//   b
//   a

function on<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

Название мероприятия.

listener: (args: ReadableEventMap[E]) => void

Функция обратного вызова

Возвращаемое значение

this

once(string | symbol, (args: any[]) => void)

function once(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

once<E>(E, (args: ReadableEventMap[E]) => void)

Добавляет функцию однократного для события с именем . При следующем запуске eventName этот прослушиватель удаляется и вызывается.

server.once('connection', (stream) => {
  console.log('Ah, we have our first user!');
});

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

По умолчанию прослушиватели событий вызываются в том порядке, в который они добавляются. Метод emitter.prependOnceListener() можно использовать в качестве альтернативы для добавления прослушивателя событий в начало массива прослушивателей.

import { EventEmitter } from 'node:events';
const myEE = new EventEmitter();
myEE.once('foo', () => console.log('a'));
myEE.prependOnceListener('foo', () => console.log('b'));
myEE.emit('foo');
// Prints:
//   b
//   a

function once<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

Название мероприятия.

listener: (args: ReadableEventMap[E]) => void

Функция обратного вызова

Возвращаемое значение

this

pause()

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

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  readable.pause();
  console.log('There will be no additional data for 1 second.');
  setTimeout(() => {
    console.log('Now data will start flowing again.');
    readable.resume();
  }, 1000);
});

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

function pause(): this

Возвращаемое значение

this

prependListener(string | symbol, (args: any[]) => void)

function prependListener(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

prependListener<E>(E, (args: ReadableEventMap[E]) => void)

Добавляет функцию listener в начала массива прослушивателей для события с именем eventName. Проверка не выполняется, чтобы узнать, добавлена ли listener. Несколько вызовов, проходящих одно и то же сочетание eventName и listener, приведет к добавлению listener и вызову несколько раз.

server.prependListener('connection', (stream) => {
  console.log('someone connected!');
});

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

function prependListener<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

Название мероприятия.

listener: (args: ReadableEventMap[E]) => void

Функция обратного вызова

Возвращаемое значение

this

prependOnceListener(string | symbol, (args: any[]) => void)

function prependOnceListener(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

prependOnceListener<E>(E, (args: ReadableEventMap[E]) => void)

Добавляет функцию однократного для события с именем в начиная массива прослушивателей. При следующем запуске eventName этот прослушиватель удаляется, а затем вызывается.

server.prependOnceListener('connection', (stream) => {
  console.log('Ah, we have our first user!');
});

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

function prependOnceListener<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

Название мероприятия.

listener: (args: ReadableEventMap[E]) => void

Функция обратного вызова

Возвращаемое значение

this

push(any, BufferEncoding)

function push(chunk: any, encoding?: BufferEncoding): boolean

Параметры

chunk: any

encoding: BufferEncoding

Возвращаемое значение

boolean

rawListeners(string | symbol)

function rawListeners(eventName: string | symbol): (args: any[]) => void[]

Параметры

eventName: string | symbol

Возвращаемое значение

(args: any[]) => void[]

rawListeners<E>(E)

Возвращает копию массива прослушивателей для события с именем eventName, включая любые оболочки (например, созданные .once()).

import { EventEmitter } from 'node:events';
const emitter = new EventEmitter();
emitter.once('log', () => console.log('log once'));

// Returns a new Array with a function `onceWrapper` which has a property
// `listener` which contains the original listener bound above
const listeners = emitter.rawListeners('log');
const logFnWrapper = listeners[0];

// Logs "log once" to the console and does not unbind the `once` event
logFnWrapper.listener();

// Logs "log once" to the console and removes the listener
logFnWrapper();

emitter.on('log', () => console.log('log persistently'));
// Will return a new Array with a single function bound by `.on()` above
const newListeners = emitter.rawListeners('log');

// Logs "log persistently" twice
newListeners[0]();
emitter.emit('log');

function rawListeners<E>(eventName: E): (args: ReadableEventMap[E]) => void[]

Параметры

eventName: E

Возвращаемое значение

(args: ReadableEventMap[E]) => void[]

read(number)

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

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

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

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

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

const readable = getReadableStreamSomehow();

// 'readable' may be triggered multiple times as data is buffered in
readable.on('readable', () => {
  let chunk;
  console.log('Stream is readable (new data received in buffer)');
  // Use a loop to make sure we read all currently available data
  while (null !== (chunk = readable.read())) {
    console.log(`Read ${chunk.length} bytes of data...`);
  }
});

// 'end' will be triggered once when there is no more data available
readable.on('end', () => {
  console.log('Reached end of stream.');
});

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

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

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' событие также будет излучаться.

Вызов прочитан после передачи 'end' события вернётся null. Ошибка во время выполнения не возникает.

function read(size?: number): any

Параметры

size: number

Необязательный аргумент для указания, сколько данных читать.

Возвращаемое значение

any

reduce<T>((previous: any, data: any, options?: Abortable) => T)

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

Если начальное значение не предоставляется, в качестве начального значения используется первый кусок потока. Если поток пуст, обещание отклоняется с помощью TypeError свойства кодаERR_INVALID_ARGS.

Функция редуктора повторяет поток по элементам, что означает, что параметр параллелизма или параллелизма отсутствует. Чтобы выполнить параллельное сокращение, можно извлечь метод асинхронной функции to readable.map .

function reduce<T>(fn: (previous: any, data: any, options?: Abortable) => T): Promise<T>

Параметры

fn: (previous: any, data: any, options?: Abortable) => T

функция редуктора, вызывающая каждый чанк в потоке. Асинхронно или нет.

Возвращаемое значение

Promise<T>

обещание конечного значения редукции.

reduce<T>((previous: T, data: any, options?: Abortable) => T, T, Abortable)

function reduce<T>(fn: (previous: T, data: any, options?: Abortable) => T, initial: T, options?: Abortable): Promise<T>

Параметры

fn: (previous: T, data: any, options?: Abortable) => T

initial: T

options: Abortable

Возвращаемое значение

Promise<T>

removeAllListeners(string | symbol)

function removeAllListeners(eventName?: string | symbol): this

Параметры

eventName: string | symbol

Возвращаемое значение

this

removeAllListeners<E>(E)

Удаляет все прослушиватели или те из указанных eventName.

Не рекомендуется удалять прослушиватели, добавленные в другое место в коде, особенно если экземпляр EventEmitter был создан другим компонентом или модулем (например, сокетами или потоками файлов).

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

function removeAllListeners<E>(eventName?: E): this

Параметры

eventName: E

Возвращаемое значение

this

removeListener(string | symbol, (args: any[]) => void)

function removeListener(eventName: string | symbol, listener: (args: any[]) => void): this

Параметры

eventName: string | symbol

listener: (args: any[]) => void

Возвращаемое значение

this

removeListener<E>(E, (args: ReadableEventMap[E]) => void)

Удаляет указанный listener из массива прослушивателя для события с именем eventName.

const callback = (stream) => {
  console.log('someone connected!');
};
server.on('connection', callback);
// ...
server.removeListener('connection', callback);

removeListener() удаляет в большинстве случаев один экземпляр прослушивателя из массива прослушивателя. Если один прослушиватель был добавлен несколько раз в массив прослушивателя для указанного eventName, removeListener() необходимо вызывать несколько раз, чтобы удалить каждый экземпляр.

После создания события все прослушиватели, подключенные к нему во время создания, вызываются в порядке. Это означает, что любые removeListener() вызовы removeAllListeners()OR после эмитирования и до завершения выполнения последним слушателем не удалят их из emit() процесса. Последующие события ведут себя должным образом.

import { EventEmitter } from 'node:events';
class MyEmitter extends EventEmitter {}
const myEmitter = new MyEmitter();

const callbackA = () => {
  console.log('A');
  myEmitter.removeListener('event', callbackB);
};

const callbackB = () => {
  console.log('B');
};

myEmitter.on('event', callbackA);

myEmitter.on('event', callbackB);

// callbackA removes listener callbackB but it will still be called.
// Internal listener array at time of emit [callbackA, callbackB]
myEmitter.emit('event');
// Prints:
//   A
//   B

// callbackB is now removed.
// Internal listener array [callbackA]
myEmitter.emit('event');
// Prints:
//   A

Поскольку слушатели управляются с помощью внутреннего массива, вызов этого параметра изменяет индексы позиций любого слушателя, зарегистрированного после удаления слушателя. Это не повлияет на порядок вызова прослушивателей, но это означает, что все копии массива прослушивателя, возвращенные методом emitter.listeners(), должны быть воссозданы.

Если одна функция была добавлена в обработчик несколько раз для одного события (как в примере ниже), removeListener() удалит последний добавленный экземпляр. В примере прослушивателя once('ping') удаляется:

import { EventEmitter } from 'node:events';
const ee = new EventEmitter();

function pong() {
  console.log('pong');
}

ee.on('ping', pong);
ee.once('ping', pong);
ee.removeListener('ping', pong);

ee.emit('ping');
ee.emit('ping');

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

function removeListener<E>(eventName: E, listener: (args: ReadableEventMap[E]) => void): this

Параметры

eventName: E

listener: (args: ReadableEventMap[E]) => void

Возвращаемое значение

this

resume()

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

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

getReadableStreamSomehow()
  .resume()
  .on('end', () => {
    console.log('Reached the end, but did not read anything.');
  });

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

function resume(): this

Возвращаемое значение

this

setEncoding(BufferEncoding)

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

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

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

const readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', (chunk) => {
  assert.equal(typeof chunk, 'string');
  console.log('Got %d characters of string data:', chunk.length);
});

function setEncoding(encoding: BufferEncoding): this

Параметры

encoding: BufferEncoding

Используемая кодировка.

Возвращаемое значение

this

some((data: any, options?: Abortable) => boolean | Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

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

function some(fn: (data: any, options?: Abortable) => boolean | Promise<boolean>, options?: Pick<ReadableOperatorOptions, "concurrency" | "signal">): Promise<boolean>

Параметры

fn: (data: any, options?: Abortable) => boolean | Promise<boolean>

функция для вызова каждого блока потока. Асинхронно или нет.

options: Pick<ReadableOperatorOptions, "concurrency" | "signal">

Возвращаемое значение

Promise<boolean>

Обещание, оценивающее true , если FN возвращает истинное значение хотя бы для одного из чанков.

take(number, Abortable)

Этот метод возвращает новый поток с первыми предельными чанками.

function take(limit: number, options?: Abortable): Readable

Параметры

limit: number

количество фрагментов, которые нужно взять из читаемого материала.

options: Abortable

Возвращаемое значение

Readable

Поток с ограниченными чанками.

toArray(Abortable)

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

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

function toArray(options?: Abortable): Promise<any[]>

Параметры

options: Abortable

Возвращаемое значение

Promise<any[]>

обещание, содержащее массив с содержимым потока.

unpipe(WritableStream)

Метод отсоединяет ручейreadable.unpipe(), Writable ранее присоединённый с помощью метода трубы.

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

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

import fs from 'node:fs';
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second.
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt.');
  readable.unpipe(writable);
  console.log('Manually close the file stream.');
  writable.end();
}, 1000);

function unpipe(destination?: WritableStream): this

Параметры

destination: WritableStream

Опциональный специфический поток для отключения

Возвращаемое значение

this

unshift(any, BufferEncoding)

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

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

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

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

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
import { StringDecoder } from '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')) {
        // Found the header boundary.
        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);
        // Remove the 'readable' listener before unshifting.
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // Now the body of the message can be read from the stream.
        callback(null, header, stream);
        return;
      }
      // Still reading the header.
      header += str;
    }
  }
}

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

function unshift(chunk: any, encoding?: BufferEncoding)

Параметры

chunk: any

Кусок данных, который нужно отложить в очередь чтения. Для потоков, не работающих в объектном режиме, chunk должны быть {string}, {Buffer}, {TypedArray}, {DataView} или null. Для потоков в объектном режиме может chunk быть любое значение JavaScript.

encoding: BufferEncoding

Кодирование строковых чанков. Должна быть действительная Buffer кодировка, например 'utf8' или 'ascii'.

wrap(ReadableStream)

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

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

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

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

myReader.on('readable', () => {
  myReader.read(); // etc.
});

function wrap(stream: ReadableStream): this

Параметры

stream: ReadableStream

«Старинный» читаемый поток

Возвращаемое значение

this

[asyncDispose]()

Звонит readable.destroy() с кнопкой AbortError и возвращает обещание, которое выполняется после завершения стрима.

function [asyncDispose](): Promise<void>

Возвращаемое значение

Promise<void>

[asyncIterator]()

function [asyncIterator](): AsyncIterator<any>

Возвращаемое значение

AsyncIterator<any>

AsyncIterator чтобы полностью поглотить поток.

[toAsyncStreamable]()

Когда --experimental-stream-iter флаг включен, Readable потоки реализуют протокол Stream.toAsyncStreamable , обеспечивая эффективное использование API stream/iter .

Это обеспечивает пакетный асинхронный итератор, который сбрасывает внутренний буфер потока на Uint8Array[] партии, амортизируя накладные расходы Promise на каждый чанк стандартного Symbol.asyncIterator пути. Для потоков в байтовом режиме блоки приводятся напрямую в виде Buffer экземпляров (которые являются Uint8Array подклассами). Для потоков в объектном режиме или кодирования каждый блок нормализуется до Uint8Array пакетирования.

Возвращаемый итератор помечен как проверенный источник, поэтому from() пропускает его без дополнительной нормализации.

import { Readable } from 'node:stream';
import { text, from } from 'node:stream/iter';

const readable = new Readable({
  read() { this.push('hello'); this.push(null); },
});

// Readable is automatically consumed via toAsyncStreamable
console.log(await text(from(readable))); // 'hello'

Без --experimental-stream-iter флага вызов этого метода приводит ERR_STREAM_ITER_MISSING_FLAGк .

function [toAsyncStreamable](): ByteReadableStream

Возвращаемое значение

ByteReadableStream

Сведения о наследуемом методе

eventNames()

Возвращает массив, в котором перечислены события, для которых у поставщика зарегистрированы прослушиватели.

import { EventEmitter } from 'node:events';

const myEE = new EventEmitter();
myEE.on('foo', () => {});
myEE.on('bar', () => {});

const sym = Symbol('symbol');
myEE.on(sym, () => {});

console.log(myEE.eventNames());
// Prints: [ 'foo', 'bar', Symbol(symbol) ]

function eventNames(): (string | symbol)[]

Возвращаемое значение

(string | symbol)[]

Унаследованный от Stream.eventNames

getMaxListeners()

Возвращает текущее значение максимального прослушивателя для EventEmitter, заданное emitter.setMaxListeners(n) или значением по умолчанию events.defaultMaxListeners.

function getMaxListeners(): number

Возвращаемое значение

number

Унаследованный от Стрим.getMaxСлушатели

pipe<T>(T, PipeOptions)

function pipe<T>(destination: T, options?: PipeOptions): T

Параметры

destination: T

options: PipeOptions

Возвращаемое значение

Унаследованный от Stream.pipe

setMaxListeners(number)

По умолчанию EventEmitterбудет выводить предупреждение, если для определенного события добавляются более 10 прослушивателей. Это полезное значение по умолчанию, которое помогает находить утечки памяти. Метод emitter.setMaxListeners() позволяет изменить ограничение для этого конкретного экземпляра EventEmitter. Для указания неограниченного количества прослушивателей можно задать значение Infinity (или 0).

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

function setMaxListeners(n: number): this

Параметры

n: number

Возвращаемое значение

this

Унаследованный от Stream.setMaxListeners

[captureRejectionSymbol](Error, string | symbol, any[])

Метод Symbol.for('nodejs.rejection') вызывается в случае, если при выпуске события происходит отказ обещания и captureRejections активируется на эмиттере. Возможно использовать events.captureRejectionSymbol вместо Symbol.for('nodejs.rejection').

import { EventEmitter, captureRejectionSymbol } from 'node:events';

class MyClass extends EventEmitter {
  constructor() {
    super({ captureRejections: true });
  }

  [captureRejectionSymbol](err, event, ...args) {
    console.log('rejection happened for', event, 'with', err, ...args);
    this.destroy(err);
  }

  destroy(err) {
    // Tear the resource down here.
  }
}

function [captureRejectionSymbol](error: Error, event: string | symbol, args: any[])

Параметры

error: Error

event: string | symbol

args: any[]

Унаследовано от Потока. [символ захватаОтвержения]