Readable interface

Balíček:
@azure/storage-common
Extends

Stream

Vlastnosti

closed

Je po true'close' vysvícení.
destroyed

Je to true po readable.destroy() tom, co bylo vyhlášeno.
errored

Vrací chybu, pokud byl proud zničen chybou.
readable

Je true , pokud je bezpečné volat read, což znamená, že proud nebyl zničen nebo vysílán 'error' , nebo 'end'.
readableAborted

Vrátí, zda byl proud zničen nebo chybně použit, než vysílá 'end'.
readableDidRead

Vrací, zda 'data' bylo vysíláno.
readableEncoding

Getter pro vlastnost encoding daného Readable proudu. Vlastnost encoding lze nastavit pomocí metody setEncoded .
readableEnded

Stává se true , když 'end' je událost vysílána.
readableFlowing

Tato vlastnost odráží současný stav Readable potoka, jak je popsán v části Tři stavy .
readableHighWaterMark

Při vytváření tohoto vrátí highWaterMarkhodnotu předaného .Readable
readableLength

Tato vlastnost obsahuje počet bajtů (nebo objektů) ve frontě připravených ke čtení. Hodnota poskytuje introspekční data týkající se stavu .highWaterMark
readableObjectMode

Getter pro vlastnost objectMode daného Readable proudu.

Metody

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

Alias pro 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) je ekvivalent stream.compose(readable, s).

Tato metoda také umožňuje poskytnout a AbortSignal , které při přerušení zničí složený proud.

Další informace naleznete na stream.compose(...streams).
destroy(Error)

Zničte stream. Volitelně vyvolat 'error' událost a vydat událost 'close' (pokud emitClose není nastaveno na false). Po tomto hovoru čtitelný stream uvolní všechny interní zdroje a následné volání push() budou ignorována.

Jakmile destroy() je volán, všechny další volání budou no-op a žádné další chyby kromě , _destroy() které mohou být vydány jako 'error'.

Implementátoři by tuto metodu neměli přepsat, ale místo toho implementovat readable._destroy().
drop(number, Abortable)

Tato metoda vrací nový proud s prvními limitními bloky vypuštěnými od začátku.
emit(string | symbol, any[])
emit<E>(E, ReadableEventMap[E])

Synchronně volá všechny naslouchací procesy zaregistrované pro událost s názvem eventName, v pořadí, v jakém byly zaregistrovány, předávání zadaných argumentů každému.

Vrátí true, pokud událost měla naslouchací procesy, false jinak.

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">)

Tato metoda je podobná a Array.prototype.every volá fn na každý chunk v proudu, aby se ověřilo, zda všechny očekávané návratové hodnoty jsou pravdivé hodnoty fn. Jakmile je volání fn na hodnotě vrácení bloku awaitfalešné, proud je zničen a slib splněn s false. Pokud všechny volání fn na blokech vrátí pravdivou hodnotu, slib je splněn s true.
filter((data: any, options?: Abortable) => boolean | Promise<boolean>, ReadableOperatorOptions)

Tato metoda umožňuje filtrování proudu. Pro každý blok v proudu se volá funkce fn a pokud vrátí pravdivou hodnotu, blok bude předán výslednému proudu. Pokud funkce fn vrátí slib – ten slib bude ed await.
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">)

Tato metoda je podobná a Array.prototype.find volá fn na každý chunk v proudu, aby našla chunk s pravdivou hodnotou fn. Jakmile je očekávaná návratová hodnota fn volání pravdivá, stream je zničen a slib splněn hodnotou, pro kterou fn vrátil pravdivou hodnotu. Pokud všechny fn volání na blokech vrátí falešnou hodnotu, slib je splněn s undefined.
flatMap((data: any, options?: Abortable) => any, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

Tato metoda vrací nový proud tak, že na každý úsek proudu aplikuje daný callback a následně výsledek zplošťuje.

Je možné vrátit proud nebo jiný iterable či asynchronní iterable z fn a výsledné proudy budou sloučeny (zploštěny) do vráceného proudu.
forEach((data: any, options?: Abortable) => void | Promise<void>, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

Tato metoda umožňuje iteraci proudu. Pro každý chunk v proudu bude volána funkce fn . Pokud funkce fn vrátí slib – ten slib bude ed await.

Tato metoda se liší od for await...of smyček tím, že může volitelně zpracovávat bloky současně. Navíc iteraci lze zastavit pouze tím, forEach že prošla volba signal a přerušila příslušný AbortController, zatímco for await...of může být zastavena pomocí break nebo return. V obou případech bude proud zničen.

Tato metoda se liší od naslouchání události 'data' tím, že využívá událost readable v základním stroji a může omezit počet souběžných fn hovorů.
isPaused()

Metoda readable.isPaused() vrací aktuální provozní stav .Readable To se používá především mechanismem, který je základem této metody readable.pipe() . Ve většině typických případů není důvod tuto metodu používat přímo.

const readable = new stream.Readable();

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

Iterátor vytvořený touto metodou dává uživatelům možnost zrušit zničení proudu, pokud je smyčka ukončena for await...of , returnbreak, nebo throw, nebo pokud by iterátor měl proud zničit, pokud během iterace vydal chybu.
listenerCount(string | symbol, (args: any[]) => void)
listenerCount<E>(E, (args: ReadableEventMap[E]) => void)

Vrátí počet naslouchacích procesů naslouchajících události s názvem eventName. Pokud listener, vrátí se v seznamu naslouchacích procesů události, kolikrát se naslouchací proces najde.
listeners(string | symbol)
listeners<E>(E)

Vrátí kopii pole naslouchacích procesů pro událost s názvem 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)

Tato metoda umožňuje mapování přes proud. Funkce fn bude volána pro každý chunk v proudu. Pokud funkce fn vrátí slib – tento slib bude ed awaitpřed předáním do resultingového proudu.
off(string | symbol, (args: any[]) => void)
off<E>(E, (args: ReadableEventMap[E]) => void)

Alias pro emitter.removeListener().
on(string | symbol, (args: any[]) => void)
on<E>(E, (args: ReadableEventMap[E]) => void)

Přidá funkci listener na konec pole naslouchacích procesů pro událost s názvem eventName. Nejsou provedeny žádné kontroly, abyste zjistili, jestli už listener byla přidána. Více volání, která předávají stejnou kombinaci eventName a listener, způsobí přidání listener a volání několikrát.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

Ve výchozím nastavení se naslouchací procesy událostí vyvolávají v pořadí, v jakém se přidávají. Metodu emitter.prependListener() lze použít jako alternativu k přidání naslouchacího procesu událostí na začátek pole naslouchacích procesů.

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)

Přidá jednorázovou funkcilistener pro událost s názvem eventName. Při příštím spuštění eventName se tento naslouchací proces odebere a potom se vyvolá.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

Ve výchozím nastavení se naslouchací procesy událostí vyvolávají v pořadí, v jakém se přidávají. Metodu emitter.prependOnceListener() lze použít jako alternativu k přidání naslouchacího procesu událostí na začátek pole naslouchacích procesů.

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()

Tato readable.pause() metoda způsobí, že proud v režimu proudění přestane vysílat 'data' události a přepne se z režimu proudění. Jakákoli data, která se stanou dostupná, zůstanou v interním bufferu.

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);
});

Metoda readable.pause() nemá žádný účinek, pokud je zde 'readable' event listener.
prependListener(string | symbol, (args: any[]) => void)
prependListener<E>(E, (args: ReadableEventMap[E]) => void)

Přidá funkci listener do začínající pole naslouchacích procesů pro událost s názvem eventName. Nejsou provedeny žádné kontroly, abyste zjistili, jestli už listener byla přidána. Více volání, která předávají stejnou kombinaci eventName a listener, způsobí přidání listener a volání několikrát.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.
prependOnceListener(string | symbol, (args: any[]) => void)
prependOnceListener<E>(E, (args: ReadableEventMap[E]) => void)

Přidá jednorázovélistener funkce pro událost s názvem eventName do začínající pole naslouchacích procesů. Při příštím spuštění eventName se tento naslouchací proces odebere a potom se vyvolá.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.
push(any, BufferEncoding)
rawListeners(string | symbol)
rawListeners<E>(E)

Vrátí kopii pole naslouchacích procesů pro událost s názvem eventName, včetně všech obálky (například těch vytvořených .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)

Metoda readable.read() čte data z interního bufferu a vrací je. Pokud nejsou k dispozici žádná data ke čtení, vrací se. null Ve výchozím nastavení jsou data vrácena jako objekt, Buffer pokud nebylo zadáno kódování pomocí readable.setEncoding() metody nebo pokud proud nepracuje v objektovém režimu.

Volitelný size argument určuje konkrétní počet bajtů k přečtení. Pokud size nejsou k dispozici bajty ke čtení, vrátí se, nullpokud proud neskončil, v takovém případě budou vrácena všechna data zbývající v interním bufferu.

Pokud argument není specifikován, size vrátí se všechna data obsažená v interním bufferu.

Argument size musí být menší nebo roven 1 GiB.

Metoda readable.read() by měla být vyvolána pouze u Readable streamů pracujících v režimu pozastavení. V režimu průtoku se automaticky volá, readable.read() dokud není vnitřní buffer zcela vyprázdněn.

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.');
});

Každé volání vrací readable.read() část dat, nebo null. Kusy nejsou spojovány. K požití všech dat, která jsou aktuálně v bufferu, je nutná smyčka while . Při čtení velkého souboru .read() se může vrátit null, že jste spotřebovali veškerý dosud uložený obsah, ale stále je k dispozici další data, která ještě nejsou uložena. V tomto případě bude vyslána nová 'readable' událost, když je v bufferu více dat. Nakonec bude událost 'end' vysílána, až nebudou k dispozici další data.

Proto pro čtení celého obsahu souboru z z , readableje nutné shromáždit bloky napříč více 'readable' událostmi:

const chunks = [];

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

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

Proud Readable v objektovém režimu vždy vrátí jednu položku z volání do readable.read(size), bez ohledu na hodnotu size argumentu.

Pokud readable.read() metoda vrátí úsek dat, 'data' bude také vygenerována událost.

Volání po vysílání události 'end' se vrátí.null Chyba za běhu se nezobrazí.
reduce<T>((previous: any, data: any, options?: Abortable) => T)

Tato metoda volá fn na každém bloku proudu v pořadí a předává mu výsledek z výpočtu na předchozím prvku. Vrací slib konečné hodnoty redukce.

Pokud není uvedena žádná počáteční hodnota, použije se první část proudu jako počáteční hodnota. Pokud je proud prázdný, slib je odmítnut vlastností s TypeErrorERR_INVALID_ARGS kódem.

Funkce reduktora iteruje proud element po prvku, což znamená, že neexistuje parametr souběžnosti ani paralelizismus. Pro současné provedení redukce můžete extrahovat asynchronní funkci do readable.map metody.
reduce<T>((previous: T, data: any, options?: Abortable) => T, T, Abortable)
removeAllListeners(string | symbol)
removeAllListeners<E>(E)

Odebere všechny naslouchací procesy nebo naslouchací procesy zadané eventName.

Je vhodné odebrat naslouchací procesy přidané jinde v kódu, zejména pokud byla instance EventEmitter vytvořená nějakou jinou komponentou nebo modulem (např. sokety nebo streamy souborů).

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.
removeListener(string | symbol, (args: any[]) => void)
removeListener<E>(E, (args: ReadableEventMap[E]) => void)

Odebere zadanou listener z pole naslouchacího procesu pro událost s názvem eventName.

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

removeListener() odebere z pole naslouchacího procesu nanejvýš jednu instanci naslouchacího procesu. Pokud byl některý jeden naslouchací proces přidán vícekrát do pole naslouchacího procesu pro zadané eventName, pak removeListener() musí být volána vícekrát, aby bylo možné odebrat každou instanci.

Jakmile se událost vygeneruje, všechny naslouchací procesy, které jsou k ní připojeny v době generování, se volají v pořadí. To znamená, že žádné removeListener() nebo volání removeAllListeners() vysílání a před dokončením posledního posluchače je neodstraní z emit() probíhajících. Následné události se chovají podle očekávání.

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

Protože jsou posluchači spravováni pomocí interního pole, volání tohoto pole změní poziční indexy každého posluchače registrovaného po jeho odstranění. To nebude mít vliv na pořadí, ve kterém se volají naslouchací procesy, ale znamená to, že všechny kopie pole naslouchacího procesu vrácené metodou emitter.listeners() bude potřeba znovu vytvořit.

Když je jedna funkce přidána jako obslužná rutina několikrát pro jednu událost (jako v příkladu níže), removeListener() odebere naposledy přidanou instanci. V příkladu se odebere naslouchací proces 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');

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.
resume()

Tato readable.resume() metoda způsobí, že explicitně Readable pozastavený proud obnoví vysílání 'data' událostí a přepne proud do režimu proudění.

Metoda readable.resume() může být použita k úplnému využití dat z proudu, aniž by se tato data skutečně zpracovávala:

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

Metoda readable.resume() nemá žádný účinek, pokud je zde 'readable' event listener.
setEncoding(BufferEncoding)

Metoda nastavuje readable.setEncoding() kódování znaků pro data čtená Readable ze streamu.

Ve výchozím nastavení není přiřazeno žádné kódování a data ze streamu jsou vrácena jako Buffer objekty. Nastavení kódování způsobí, že data proudu jsou vrácena jako řetězce zadaného kódování, nikoli jako Buffer objekty. Například volání readable.setEncoding('utf8') způsobí, že výstupní data budou interpretována jako UTF-8 data a předána jako řetězce. Volání readable.setEncoding('hex') způsobí, že data budou zakódována v hexadecimálním řetězcovém formátu.

Proud Readable správně zpracuje vícebajtové znaky doručené proudem, které by jinak byly nesprávně dekódovány, pokud by byly jednoduše odebrány jako Buffer objekty.

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">)

Tato metoda je podobná a Array.prototype.some volá fn na každý chunk v proudu, dokud není očekávaná návratová hodnota true (nebo jakákoli pravdivá hodnota). Jakmile je volací hodnota fn na úseku awaited return pravdivá, stream je zničen a slib splněn s true. Pokud žádné z volání fn na blokech nevrátí pravdivou hodnotu, slib je splněn pomocí false.
take(number, Abortable)

Tato metoda vrací nový proud s prvními limitními bloky.
toArray(Abortable)

Tato metoda umožňuje snadné získání obsahu proudu.

Protože tato metoda načte celý stream do paměti, neguje výhody streamů. Je určen pro interoperabilitu a pohodlí, ne jako hlavní způsob konzumace streamů.
unpipe(WritableStream)

Metoda readable.unpipe() odpojí proud, Writable který byl dříve připojen, pomocí metody potrubí .

Pokud není specifikováno destination , jsou všechny trubky odpojeny.

Pokud je specifikováno destination , ale není pro něj nastaveno žádné potrubí, metoda nic neudělá.

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)

Passing chunk as null signalizuje konec proudu (EOF) a chová se stejně jako readable.push(null), po čemž již nelze zapisovat další data. EOF signál je umístěn na konci bufferu a všechna uložená data budou stále vyčištěna.

Metoda vkládá readable.unshift() část dat zpět do interního bufferu. To je užitečné v určitých situacích, kdy je stream spotřebován kódem, který potřebuje "odkonzumovat" určité množství dat, která optimisticky vytáhl ze zdroje, aby mohla být data předána jiné straně.

Metoda nemůže být volána stream.unshift(chunk) po vyvolání události 'end' , jinak bude vyhozena chyba za běhu.

Vývojáři, kteří často používají stream.unshift() , by měli zvážit přechod na používání Transform streamu. Více informací najdete v této sekci.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;
    }
  }
}

Na rozdíl od push neukončí proces čtení resetováním vnitřního stavu čtecího toku (push). stream.unshift(chunk) To může způsobit neočekávané výsledky, pokud readable.unshift() je vyvoláno během čtení (tj. z _read implementace na vlastním toku). Následování volání na readable.unshift() s okamžitým push správně resetuje stav čtení, avšak je nejlepší se volání readable.unshift() během čtení jednoduše vyhnout.
wrap(ReadableStream)

Před Node.js 0.10 streamy neimplementovaly node:stream celé API modulů tak, jak je nyní definováno. (Více informací naleznete zde Compatibility .)

Při použití starší knihovny Node.js, která generuje 'data' události a má pauzovací metodu pouze poradní, lze tuto metodu použít k vytvoření proudureadable.wrap(), Readable který používá starý proud jako zdroj dat.

Používání readable.wrap() bude zřídka nutné, ale metoda byla poskytnuta jako pohodlí pro interakci se staršími Node.js aplikacemi a knihovnami.

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]()

Volá readable.destroy() s a AbortError vrací slib, který se splní, až stream skončí.
[asyncIterator]()
[toAsyncStreamable]()

Když je příznak povolen, --experimental-stream-iter streamy implementují protokol, Readable což umožňuje efektivní spotřebu Stream.toAsyncStreamable API pro tento příznakstream/iter.

To poskytuje dávkový asynchronní iterátor, který odčerpává vnitřní buffer toku do Uint8Array[] dávek, čímž amortizuje režii Symbol.asyncIterator Promise na jeden chunk standardní cesty. U toků v bajtovém režimu jsou bloky přímo vydávány jako Buffer instance (které jsou Uint8Array podtřídami). Pro objektové nebo kódované toky je každý blok normalizován na před Uint8Array batchováním.

Vrácený iterátor je označen jako ověřený zdroj, takže from() jej předává bez další normalizace.

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'

Bez vlajky --experimental-stream-iter volání této metody hodí ERR_STREAM_ITER_MISSING_FLAG.

Zděděné metody

eventNames()

Vrátí pole se seznamem událostí, pro které má emiter zaregistrované naslouchací procesy.

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()

Vrátí aktuální maximální hodnotu naslouchacího procesu pro EventEmitter, která je buď nastavena emitter.setMaxListeners(n), nebo je výchozí hodnota events.defaultMaxListeners.
pipe<T>(T, PipeOptions)
setMaxListeners(number)

Ve výchozím nastavení EventEmitters zobrazí upozornění, pokud se pro konkrétní událost přidá více než 10 naslouchací procesy. Toto je užitečné výchozí nastavení, které pomáhá najít nevrácenou paměť. Metoda emitter.setMaxListeners() umožňuje upravit limit pro tuto konkrétní instanci EventEmitter. Hodnotu lze nastavit na Infinity (nebo 0), aby bylo možné označit neomezený počet naslouchacích procesů.

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.
[captureRejectionSymbol](Error, string | symbol, any[])

Metoda Symbol.for('nodejs.rejection') se používá v případě, že při vysílání události dojde k odmítnutí slibu a captureRejections je na emitoru povolena. Je možné použít events.captureRejectionSymbol místo .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.
  }
}

Podrobnosti vlastnosti

closed

Je po true'close' vysvícení.

closed: boolean

Hodnota vlastnosti

boolean

destroyed

Je to true po readable.destroy() tom, co bylo vyhlášeno.

destroyed: boolean

Hodnota vlastnosti

boolean

errored

Vrací chybu, pokud byl proud zničen chybou.

errored: Error | null

Hodnota vlastnosti

Error | null

readable

Je true , pokud je bezpečné volat read, což znamená, že proud nebyl zničen nebo vysílán 'error' , nebo 'end'.

readable: boolean

Hodnota vlastnosti

boolean

readableAborted

Vrátí, zda byl proud zničen nebo chybně použit, než vysílá 'end'.

readableAborted: boolean

Hodnota vlastnosti

boolean

readableDidRead

Vrací, zda 'data' bylo vysíláno.

readableDidRead: boolean

Hodnota vlastnosti

boolean

readableEncoding

Getter pro vlastnost encoding daného Readable proudu. Vlastnost encoding lze nastavit pomocí metody setEncoded .

readableEncoding: BufferEncoding | null

Hodnota vlastnosti

BufferEncoding | null

readableEnded

Stává se true , když 'end' je událost vysílána.

readableEnded: boolean

Hodnota vlastnosti

boolean

readableFlowing

Tato vlastnost odráží současný stav Readable potoka, jak je popsán v části Tři stavy .

readableFlowing: boolean | null

Hodnota vlastnosti

boolean | null

readableHighWaterMark

Při vytváření tohoto vrátí highWaterMarkhodnotu předaného .Readable

readableHighWaterMark: number

Hodnota vlastnosti

number

readableLength

Tato vlastnost obsahuje počet bajtů (nebo objektů) ve frontě připravených ke čtení. Hodnota poskytuje introspekční data týkající se stavu .highWaterMark

readableLength: number

Hodnota vlastnosti

number

readableObjectMode

Getter pro vlastnost objectMode daného Readable proudu.

readableObjectMode: boolean

Hodnota vlastnosti

boolean

Podrobnosti metody

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Alias pro emitter.on(eventName, listener).

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

Parametry

eventName

E

listener

(args: ReadableEventMap[E]) => void

Návraty

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) je ekvivalent stream.compose(readable, s).

Tato metoda také umožňuje poskytnout a AbortSignal , které při přerušení zničí složený proud.

Další informace naleznete na stream.compose(...streams).

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

Parametry

stream

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

options

Abortable

Návraty

Duplex

proud složený z proudu stream.

destroy(Error)

Zničte stream. Volitelně vyvolat 'error' událost a vydat událost 'close' (pokud emitClose není nastaveno na false). Po tomto hovoru čtitelný stream uvolní všechny interní zdroje a následné volání push() budou ignorována.

Jakmile destroy() je volán, všechny další volání budou no-op a žádné další chyby kromě , _destroy() které mohou být vydány jako 'error'.

Implementátoři by tuto metodu neměli přepsat, ale místo toho implementovat readable._destroy().

function destroy(error?: Error): this

Parametry

error

Error

Chyba, která bude předána jako payload v události 'error'

Návraty

this

drop(number, Abortable)

Tato metoda vrací nový proud s prvními limitními bloky vypuštěnými od začátku.

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

Parametry

limit

number

počet bloků, které se mají vynechat z čitelného.

options

Abortable

Návraty

Readable

Stream s limitními bloky vypuštěnými od začátku.

emit(string | symbol, any[]) 

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

Parametry

eventName

string | symbol

args

any[]

Návraty

boolean

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

Synchronně volá všechny naslouchací procesy zaregistrované pro událost s názvem eventName, v pořadí, v jakém byly zaregistrovány, předávání zadaných argumentů každému.

Vrátí true, pokud událost měla naslouchací procesy, false jinak.

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

Parametry

eventName

E

args

ReadableEventMap[E]

Návraty

boolean

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

Tato metoda je podobná a Array.prototype.every volá fn na každý chunk v proudu, aby se ověřilo, zda všechny očekávané návratové hodnoty jsou pravdivé hodnoty fn. Jakmile je volání fn na hodnotě vrácení bloku awaitfalešné, proud je zničen a slib splněn s false. Pokud všechny volání fn na blokech vrátí pravdivou hodnotu, slib je splněn s true.

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

Parametry

fn

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

funkce pro volání každého bloku proudu. Ať už asynchronní nebo ne.

options

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

Návraty

Promise<boolean>

Slib vyhodnocující na if truefn vrací pravdivou hodnotu pro každý blok.

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

Tato metoda umožňuje filtrování proudu. Pro každý blok v proudu se volá funkce fn a pokud vrátí pravdivou hodnotu, blok bude předán výslednému proudu. Pokud funkce fn vrátí slib – ten slib bude ed await.

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

Parametry

fn

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

funkce pro filtrování částí proudu. Ať už asynchronní nebo ne.

options

ReadableOperatorOptions

Návraty

Readable

Proud filtrovaný predikátem 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>

Parametry

fn

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

options

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

Návraty

Promise<any>

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

Tato metoda je podobná a Array.prototype.find volá fn na každý chunk v proudu, aby našla chunk s pravdivou hodnotou fn. Jakmile je očekávaná návratová hodnota fn volání pravdivá, stream je zničen a slib splněn hodnotou, pro kterou fn vrátil pravdivou hodnotu. Pokud všechny fn volání na blokech vrátí falešnou hodnotu, slib je splněn s undefined.

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

Parametry

fn

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

funkce pro volání každého bloku proudu. Ať už asynchronní nebo ne.

options

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

Návraty

Promise<T | undefined>

Slib vyhodnocující na první blok, pro který FN vyhodnocoval s pravdivou hodnotou, nebo undefined pokud nebyl nalezen žádný prvek.

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

Tato metoda vrací nový proud tak, že na každý úsek proudu aplikuje daný callback a následně výsledek zplošťuje.

Je možné vrátit proud nebo jiný iterable či asynchronní iterable z fn a výsledné proudy budou sloučeny (zploštěny) do vráceného proudu.

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

Parametry

fn

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

funkce pro mapování přes každý chunk v proudu. Možná asynchronní. Může to být stream nebo generátor.

options

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

Návraty

Readable

Proud plochý s funkcí fn.

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

Tato metoda umožňuje iteraci proudu. Pro každý chunk v proudu bude volána funkce fn . Pokud funkce fn vrátí slib – ten slib bude ed await.

Tato metoda se liší od for await...of smyček tím, že může volitelně zpracovávat bloky současně. Navíc iteraci lze zastavit pouze tím, forEach že prošla volba signal a přerušila příslušný AbortController, zatímco for await...of může být zastavena pomocí break nebo return. V obou případech bude proud zničen.

Tato metoda se liší od naslouchání události 'data' tím, že využívá událost readable v základním stroji a může omezit počet souběžných fn hovorů.

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

Parametry

fn

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

funkce pro volání každého bloku proudu. Ať už asynchronní nebo ne.

options

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

Návraty

Promise<void>

slib, až stream skončí.

isPaused()

Metoda readable.isPaused() vrací aktuální provozní stav .Readable To se používá především mechanismem, který je základem této metody readable.pipe() . Ve většině typických případů není důvod tuto metodu používat přímo.

const readable = new stream.Readable();

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

function isPaused(): boolean

Návraty

boolean

iterator(ReadableIteratorOptions)

Iterátor vytvořený touto metodou dává uživatelům možnost zrušit zničení proudu, pokud je smyčka ukončena for await...of , returnbreak, nebo throw, nebo pokud by iterátor měl proud zničit, pokud během iterace vydal chybu.

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

Parametry

options

ReadableIteratorOptions

Návraty

AsyncIterator<any>

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

number

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

Vrátí počet naslouchacích procesů naslouchajících události s názvem eventName. Pokud listener, vrátí se v seznamu naslouchacích procesů události, kolikrát se naslouchací proces najde.

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

Parametry

eventName

E

Název události, pro kterou se naslouchá

listener

(args: ReadableEventMap[E]) => void

Funkce obslužné rutiny události

Návraty

number

listeners(string | symbol) 

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

Parametry

eventName

string | symbol

Návraty

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

listeners<E>(E)

Vrátí kopii pole naslouchacích procesů pro událost s názvem 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[]

Parametry

eventName

E

Návraty

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

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

Tato metoda umožňuje mapování přes proud. Funkce fn bude volána pro každý chunk v proudu. Pokud funkce fn vrátí slib – tento slib bude ed awaitpřed předáním do resultingového proudu.

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

Parametry

fn

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

funkce pro mapování přes každý chunk v proudu. Ať už asynchronní nebo ne.

options

ReadableOperatorOptions

Návraty

Readable

Stream mapovaný funkcí fn.

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Alias pro emitter.removeListener().

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

Parametry

eventName

E

listener

(args: ReadableEventMap[E]) => void

Návraty

this

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Přidá funkci listener na konec pole naslouchacích procesů pro událost s názvem eventName. Nejsou provedeny žádné kontroly, abyste zjistili, jestli už listener byla přidána. Více volání, která předávají stejnou kombinaci eventName a listener, způsobí přidání listener a volání několikrát.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

Ve výchozím nastavení se naslouchací procesy událostí vyvolávají v pořadí, v jakém se přidávají. Metodu emitter.prependListener() lze použít jako alternativu k přidání naslouchacího procesu událostí na začátek pole naslouchacích procesů.

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

Parametry

eventName

E

Název události.

listener

(args: ReadableEventMap[E]) => void

Funkce zpětného volání

Návraty

this

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Přidá jednorázovou funkcilistener pro událost s názvem eventName. Při příštím spuštění eventName se tento naslouchací proces odebere a potom se vyvolá.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

Ve výchozím nastavení se naslouchací procesy událostí vyvolávají v pořadí, v jakém se přidávají. Metodu emitter.prependOnceListener() lze použít jako alternativu k přidání naslouchacího procesu událostí na začátek pole naslouchacích procesů.

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

Parametry

eventName

E

Název události.

listener

(args: ReadableEventMap[E]) => void

Funkce zpětného volání

Návraty

this

pause()

Tato readable.pause() metoda způsobí, že proud v režimu proudění přestane vysílat 'data' události a přepne se z režimu proudění. Jakákoli data, která se stanou dostupná, zůstanou v interním bufferu.

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);
});

Metoda readable.pause() nemá žádný účinek, pokud je zde 'readable' event listener.

function pause(): this

Návraty

this

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Přidá funkci listener do začínající pole naslouchacích procesů pro událost s názvem eventName. Nejsou provedeny žádné kontroly, abyste zjistili, jestli už listener byla přidána. Více volání, která předávají stejnou kombinaci eventName a listener, způsobí přidání listener a volání několikrát.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

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

Parametry

eventName

E

Název události.

listener

(args: ReadableEventMap[E]) => void

Funkce zpětného volání

Návraty

this

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Přidá jednorázovélistener funkce pro událost s názvem eventName do začínající pole naslouchacích procesů. Při příštím spuštění eventName se tento naslouchací proces odebere a potom se vyvolá.

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

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

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

Parametry

eventName

E

Název události.

listener

(args: ReadableEventMap[E]) => void

Funkce zpětného volání

Návraty

this

push(any, BufferEncoding) 

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

Parametry

chunk

any

encoding

BufferEncoding

Návraty

boolean

rawListeners(string | symbol) 

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

Parametry

eventName

string | symbol

Návraty

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

rawListeners<E>(E)

Vrátí kopii pole naslouchacích procesů pro událost s názvem eventName, včetně všech obálky (například těch vytvořených .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[]

Parametry

eventName

E

Návraty

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

read(number)

Metoda readable.read() čte data z interního bufferu a vrací je. Pokud nejsou k dispozici žádná data ke čtení, vrací se. null Ve výchozím nastavení jsou data vrácena jako objekt, Buffer pokud nebylo zadáno kódování pomocí readable.setEncoding() metody nebo pokud proud nepracuje v objektovém režimu.

Volitelný size argument určuje konkrétní počet bajtů k přečtení. Pokud size nejsou k dispozici bajty ke čtení, vrátí se, nullpokud proud neskončil, v takovém případě budou vrácena všechna data zbývající v interním bufferu.

Pokud argument není specifikován, size vrátí se všechna data obsažená v interním bufferu.

Argument size musí být menší nebo roven 1 GiB.

Metoda readable.read() by měla být vyvolána pouze u Readable streamů pracujících v režimu pozastavení. V režimu průtoku se automaticky volá, readable.read() dokud není vnitřní buffer zcela vyprázdněn.

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.');
});

Každé volání vrací readable.read() část dat, nebo null. Kusy nejsou spojovány. K požití všech dat, která jsou aktuálně v bufferu, je nutná smyčka while . Při čtení velkého souboru .read() se může vrátit null, že jste spotřebovali veškerý dosud uložený obsah, ale stále je k dispozici další data, která ještě nejsou uložena. V tomto případě bude vyslána nová 'readable' událost, když je v bufferu více dat. Nakonec bude událost 'end' vysílána, až nebudou k dispozici další data.

Proto pro čtení celého obsahu souboru z z , readableje nutné shromáždit bloky napříč více 'readable' událostmi:

const chunks = [];

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

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

Proud Readable v objektovém režimu vždy vrátí jednu položku z volání do readable.read(size), bez ohledu na hodnotu size argumentu.

Pokud readable.read() metoda vrátí úsek dat, 'data' bude také vygenerována událost.

Volání po vysílání události 'end' se vrátí.null Chyba za běhu se nezobrazí.

function read(size?: number): any

Parametry

size

number

Volitelný argument pro určení, kolik dat má být přečteno.

Návraty

any

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

Tato metoda volá fn na každém bloku proudu v pořadí a předává mu výsledek z výpočtu na předchozím prvku. Vrací slib konečné hodnoty redukce.

Pokud není uvedena žádná počáteční hodnota, použije se první část proudu jako počáteční hodnota. Pokud je proud prázdný, slib je odmítnut vlastností s TypeErrorERR_INVALID_ARGS kódem.

Funkce reduktora iteruje proud element po prvku, což znamená, že neexistuje parametr souběžnosti ani paralelizismus. Pro současné provedení redukce můžete extrahovat asynchronní funkci do readable.map metody.

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

Parametry

fn

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

redukční funkce, která volá každý chunk v streamu. Ať už asynchronní nebo ne.

Návraty

Promise<T>

slib konečné hodnoty snížení.

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>

Parametry

fn

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

initial

T

options

Abortable

Návraty

Promise<T>

removeAllListeners(string | symbol) 

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

Parametry

eventName

string | symbol

Návraty

this

removeAllListeners<E>(E)

Odebere všechny naslouchací procesy nebo naslouchací procesy zadané eventName.

Je vhodné odebrat naslouchací procesy přidané jinde v kódu, zejména pokud byla instance EventEmitter vytvořená nějakou jinou komponentou nebo modulem (např. sokety nebo streamy souborů).

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

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

Parametry

eventName

E

Návraty

this

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

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

Parametry

eventName

string | symbol

listener

(args: any[]) => void

Návraty

this

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

Odebere zadanou listener z pole naslouchacího procesu pro událost s názvem eventName.

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

removeListener() odebere z pole naslouchacího procesu nanejvýš jednu instanci naslouchacího procesu. Pokud byl některý jeden naslouchací proces přidán vícekrát do pole naslouchacího procesu pro zadané eventName, pak removeListener() musí být volána vícekrát, aby bylo možné odebrat každou instanci.

Jakmile se událost vygeneruje, všechny naslouchací procesy, které jsou k ní připojeny v době generování, se volají v pořadí. To znamená, že žádné removeListener() nebo volání removeAllListeners() vysílání a před dokončením posledního posluchače je neodstraní z emit() probíhajících. Následné události se chovají podle očekávání.

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

Protože jsou posluchači spravováni pomocí interního pole, volání tohoto pole změní poziční indexy každého posluchače registrovaného po jeho odstranění. To nebude mít vliv na pořadí, ve kterém se volají naslouchací procesy, ale znamená to, že všechny kopie pole naslouchacího procesu vrácené metodou emitter.listeners() bude potřeba znovu vytvořit.

Když je jedna funkce přidána jako obslužná rutina několikrát pro jednu událost (jako v příkladu níže), removeListener() odebere naposledy přidanou instanci. V příkladu se odebere naslouchací proces 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');

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

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

Parametry

eventName

E

listener

(args: ReadableEventMap[E]) => void

Návraty

this

resume()

Tato readable.resume() metoda způsobí, že explicitně Readable pozastavený proud obnoví vysílání 'data' událostí a přepne proud do režimu proudění.

Metoda readable.resume() může být použita k úplnému využití dat z proudu, aniž by se tato data skutečně zpracovávala:

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

Metoda readable.resume() nemá žádný účinek, pokud je zde 'readable' event listener.

function resume(): this

Návraty

this

setEncoding(BufferEncoding)

Metoda nastavuje readable.setEncoding() kódování znaků pro data čtená Readable ze streamu.

Ve výchozím nastavení není přiřazeno žádné kódování a data ze streamu jsou vrácena jako Buffer objekty. Nastavení kódování způsobí, že data proudu jsou vrácena jako řetězce zadaného kódování, nikoli jako Buffer objekty. Například volání readable.setEncoding('utf8') způsobí, že výstupní data budou interpretována jako UTF-8 data a předána jako řetězce. Volání readable.setEncoding('hex') způsobí, že data budou zakódována v hexadecimálním řetězcovém formátu.

Proud Readable správně zpracuje vícebajtové znaky doručené proudem, které by jinak byly nesprávně dekódovány, pokud by byly jednoduše odebrány jako Buffer objekty.

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

Parametry

encoding

BufferEncoding

Kódování, které se má použít.

Návraty

this

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

Tato metoda je podobná a Array.prototype.some volá fn na každý chunk v proudu, dokud není očekávaná návratová hodnota true (nebo jakákoli pravdivá hodnota). Jakmile je volací hodnota fn na úseku awaited return pravdivá, stream je zničen a slib splněn s true. Pokud žádné z volání fn na blokech nevrátí pravdivou hodnotu, slib je splněn pomocí false.

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

Parametry

fn

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

funkce pro volání každého bloku proudu. Ať už asynchronní nebo ne.

options

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

Návraty

Promise<boolean>

slib vyhodnocující na , true pokud fn vrací pravdivou hodnotu alespoň pro jeden z bloků.

take(number, Abortable)

Tato metoda vrací nový proud s prvními limitními bloky.

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

Parametry

limit

number

Počet částí, které lze vzít z čitelného.

options

Abortable

Návraty

Readable

stream s omezeným limitem úseků.

toArray(Abortable)

Tato metoda umožňuje snadné získání obsahu proudu.

Protože tato metoda načte celý stream do paměti, neguje výhody streamů. Je určen pro interoperabilitu a pohodlí, ne jako hlavní způsob konzumace streamů.

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

Parametry

options

Abortable

Návraty

Promise<any[]>

slib obsahující pole s obsahem proudu.

unpipe(WritableStream)

Metoda readable.unpipe() odpojí proud, Writable který byl dříve připojen, pomocí metody potrubí .

Pokud není specifikováno destination , jsou všechny trubky odpojeny.

Pokud je specifikováno destination , ale není pro něj nastaveno žádné potrubí, metoda nic neudělá.

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

Parametry

destination

WritableStream

Volitelné specifické proudění pro odpojení potrubí

Návraty

this

unshift(any, BufferEncoding)

Passing chunk as null signalizuje konec proudu (EOF) a chová se stejně jako readable.push(null), po čemž již nelze zapisovat další data. EOF signál je umístěn na konci bufferu a všechna uložená data budou stále vyčištěna.

Metoda vkládá readable.unshift() část dat zpět do interního bufferu. To je užitečné v určitých situacích, kdy je stream spotřebován kódem, který potřebuje "odkonzumovat" určité množství dat, která optimisticky vytáhl ze zdroje, aby mohla být data předána jiné straně.

Metoda nemůže být volána stream.unshift(chunk) po vyvolání události 'end' , jinak bude vyhozena chyba za běhu.

Vývojáři, kteří často používají stream.unshift() , by měli zvážit přechod na používání Transform streamu. Více informací najdete v této sekci.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;
    }
  }
}

Na rozdíl od push neukončí proces čtení resetováním vnitřního stavu čtecího toku (push). stream.unshift(chunk) To může způsobit neočekávané výsledky, pokud readable.unshift() je vyvoláno během čtení (tj. z _read implementace na vlastním toku). Následování volání na readable.unshift() s okamžitým push správně resetuje stav čtení, avšak je nejlepší se volání readable.unshift() během čtení jednoduše vyhnout.

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

Parametry

chunk

any

Část dat, která se musí vrátit zpět do četné fronty. Pro proudy, které nefungují v objektovém režimu, chunk musí být {string}, {Buffer}, {TypedArray}, {DataView} nebo null. Pro objektové proudy může být chunk jakákoli hodnota JavaScriptu.

encoding

BufferEncoding

Kódování bloků řetězců. Musí být platné kódování Buffer , například 'utf8' nebo 'ascii'.

wrap(ReadableStream)

Před Node.js 0.10 streamy neimplementovaly node:stream celé API modulů tak, jak je nyní definováno. (Více informací naleznete zde Compatibility .)

Při použití starší knihovny Node.js, která generuje 'data' události a má pauzovací metodu pouze poradní, lze tuto metodu použít k vytvoření proudureadable.wrap(), Readable který používá starý proud jako zdroj dat.

Používání readable.wrap() bude zřídka nutné, ale metoda byla poskytnuta jako pohodlí pro interakci se staršími Node.js aplikacemi a knihovnami.

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

Parametry

stream

ReadableStream

"Starý styl" čitelný stream

Návraty

this

[asyncDispose]()

Volá readable.destroy() s a AbortError vrací slib, který se splní, až stream skončí.

function [asyncDispose](): Promise<void>

Návraty

Promise<void>

[asyncIterator]() 

function [asyncIterator](): AsyncIterator<any>

Návraty

AsyncIterator<any>

AsyncIterator aby potok plně spotřeboval.

[toAsyncStreamable]()

Když je příznak povolen, --experimental-stream-iter streamy implementují protokol, Readable což umožňuje efektivní spotřebu Stream.toAsyncStreamable API pro tento příznakstream/iter.

To poskytuje dávkový asynchronní iterátor, který odčerpává vnitřní buffer toku do Uint8Array[] dávek, čímž amortizuje režii Symbol.asyncIterator Promise na jeden chunk standardní cesty. U toků v bajtovém režimu jsou bloky přímo vydávány jako Buffer instance (které jsou Uint8Array podtřídami). Pro objektové nebo kódované toky je každý blok normalizován na před Uint8Array batchováním.

Vrácený iterátor je označen jako ověřený zdroj, takže from() jej předává bez další normalizace.

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'

Bez vlajky --experimental-stream-iter volání této metody hodí ERR_STREAM_ITER_MISSING_FLAG.

function [toAsyncStreamable](): ByteReadableStream

Návraty

ByteReadableStream

Podrobnosti zděděných metod

eventNames()

Vrátí pole se seznamem událostí, pro které má emiter zaregistrované naslouchací procesy.

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)[]

Návraty

(string | symbol)[]

Děděno od Stream.eventNames

getMaxListeners()

Vrátí aktuální maximální hodnotu naslouchacího procesu pro EventEmitter, která je buď nastavena emitter.setMaxListeners(n), nebo je výchozí hodnota events.defaultMaxListeners.

function getMaxListeners(): number

Návraty

number

Děděno od Stream.getMaxListeners

pipe<T>(T, PipeOptions) 

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

Parametry

destination

T

options

PipeOptions

Návraty

T

Děděno od Stream.pipe

setMaxListeners(number)

Ve výchozím nastavení EventEmitters zobrazí upozornění, pokud se pro konkrétní událost přidá více než 10 naslouchací procesy. Toto je užitečné výchozí nastavení, které pomáhá najít nevrácenou paměť. Metoda emitter.setMaxListeners() umožňuje upravit limit pro tuto konkrétní instanci EventEmitter. Hodnotu lze nastavit na Infinity (nebo 0), aby bylo možné označit neomezený počet naslouchacích procesů.

Vrátí odkaz na EventEmitter, aby volání bylo možné zřetězený.

function setMaxListeners(n: number): this

Parametry

n

number

Návraty

this

Děděno od Stream.setMaxPosluchači

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

Metoda Symbol.for('nodejs.rejection') se používá v případě, že při vysílání události dojde k odmítnutí slibu a captureRejections je na emitoru povolena. Je možné použít events.captureRejectionSymbol místo .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[])

Parametry

error

Error

event

string | symbol

args

any[]

Zděděno z Streamu. [zachyceOdmítnutíSymbol]