Readable interface

Paket:
@azure/storage-common
Memperluas

Stream

Properti

closed

Adalah true setelah 'close' telah dipancarkan.
destroyed

Adalah true setelah readable.destroy() telah dipanggil.
errored

Mengembalikan kesalahan jika aliran telah dihancurkan dengan kesalahan.
readable

Apakah true jika aman untuk memanggil baca, yang berarti aliran belum dihancurkan atau dipancarkan 'error' atau 'end'.
readableAborted

Mengembalikan apakah aliran dihancurkan atau error sebelum dipancarkan 'end'.
readableDidRead

Mengembalikan apakah 'data' telah dipancarkan.
readableEncoding

Pengambil untuk properti encoding aliran tertentu Readable . Properti encoding dapat diatur menggunakan metode setEncoding .
readableEnded

Menjadi true ketika 'end' peristiwa dipancarkan.
readableFlowing

Properti ini mencerminkan status Readable aliran saat ini seperti yang dijelaskan di bagian Tiga status .
readableHighWaterMark

Mengembalikan nilai lulus highWaterMark saat membuat ini Readable.
readableLength

Properti ini berisi jumlah byte (atau objek) dalam antrean yang siap dibaca. Nilai tersebut memberikan data introspeksi mengenai status highWaterMark.
readableObjectMode

Pengambil untuk properti objectMode aliran tertentu Readable .

Metode

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

Alias untuk 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) setara dengan stream.compose(readable, s).

Metode ini juga memungkinkan untuk AbortSignal disediakan, yang akan menghancurkan aliran yang disusun saat dibatalkan.

Lihat stream.compose(...streams) untuk informasi lebih lanjut.
destroy(Error)

Hancurkan alirannya. Secara opsional memancarkan peristiwa 'error' , dan memancarkan 'close' peristiwa (kecuali diatur emitClose ke false). Setelah panggilan ini, aliran yang dapat dibaca akan melepaskan sumber daya internal apa pun dan panggilan berikutnya akan push() diabaikan.

Setelah destroy() dipanggil, panggilan lebih lanjut akan menjadi no-op dan tidak ada kesalahan lebih lanjut kecuali dari _destroy() yang dapat dipancarkan sebagai 'error'.

Pelaksana tidak boleh mengganti metode ini, melainkan menerapkan readable._destroy().
drop(number, Abortable)

Metode ini mengembalikan aliran baru dengan potongan batas pertama yang dijatuhkan dari awal.
emit(string | symbol, any[])
emit<E>(E, ReadableEventMap[E])

Secara sinkron memanggil masing-masing pendengar yang terdaftar untuk peristiwa bernama eventName, dalam urutan terdaftar, meneruskan argumen yang disediakan ke masing-masing.

Mengembalikan true jika peristiwa memiliki pendengar, false sebaliknya.

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

Metode ini mirip dengan Array.prototype.every dan memanggil fn pada setiap potongan dalam aliran untuk memeriksa apakah semua nilai pengembalian yang ditunggu adalah nilai kebenaran untuk fn. Setelah panggilan fn pada awaitnilai pengembalian chunk ed palsu, aliran dihancurkan dan janji dipenuhi dengan false. Jika semua panggilan fn pada potongan mengembalikan nilai yang benar, janji dipenuhi dengan true.
filter((data: any, options?: Abortable) => boolean | Promise<boolean>, ReadableOperatorOptions)

Metode ini memungkinkan pemfilteran aliran. Untuk setiap potongan dalam aliran, fungsi fn akan dipanggil dan jika mengembalikan nilai yang benar, potongan akan diteruskan ke aliran hasil. Jika fungsi fn mengembalikan promise - promise tersebut akan 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">)

Metode ini mirip dengan Array.prototype.find dan memanggil fn pada setiap potongan dalam aliran untuk menemukan potongan dengan nilai kebenaran untuk fn. Setelah nilai pengembalian panggilan fn yang ditunggu-tunggu adalah kebenaran, aliran dihancurkan dan janji dipenuhi dengan nilai yang fn mengembalikan nilai yang benar. Jika semua panggilan fn pada potongan mengembalikan nilai palsu, janji dipenuhi dengan undefined.
flatMap((data: any, options?: Abortable) => any, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

Metode ini mengembalikan aliran baru dengan menerapkan callback yang diberikan ke setiap bagian aliran dan kemudian meratakan hasilnya.

Dimungkinkan untuk mengembalikan aliran atau iterable atau asinkron lain yang dapat diulang dari fn dan aliran hasil akan digabungkan (diratakan) ke dalam aliran yang dikembalikan.
forEach((data: any, options?: Abortable) => void | Promise<void>, Pick<ReadableOperatorOptions, "concurrency" | "signal">)

Metode ini memungkinkan iterasi aliran. Untuk setiap potongan dalam aliran, fungsi fn akan dipanggil. Jika fungsi fn mengembalikan promise - promise tersebut akan ed await.

Metode ini berbeda dari for await...of loop karena dapat secara opsional memproses potongan secara bersamaan. Selain itu, iterasi forEach hanya dapat dihentikan dengan melewati signal opsi dan membatalkan AbortController terkait sementara for await...of dapat dihentikan dengan break atau return. Dalam kedua kasus tersebut, aliran akan dihancurkan.

Metode ini berbeda dengan mendengarkan 'data' peristiwa karena menggunakan readable peristiwa dalam mesin yang mendasarinya dan dapat membatasi jumlah panggilan fn bersamaan.
isPaused()

Metode ini readable.isPaused() mengembalikan status operasi saat ini dari .Readable Ini digunakan terutama oleh mekanisme yang mendasari metode ini readable.pipe() . Dalam kebanyakan kasus umum, tidak akan ada alasan untuk menggunakan metode ini secara langsung.

const readable = new stream.Readable();

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

Iterator yang dibuat dengan metode ini memberi pengguna opsi untuk membatalkan penghancuran aliran jika for await...of loop keluar oleh return, break, atau throw, atau jika iterator harus menghancurkan aliran jika aliran memancarkan kesalahan selama iterasi.
listenerCount(string | symbol, (args: any[]) => void)
listenerCount<E>(E, (args: ReadableEventMap[E]) => void)

Mengembalikan jumlah pendengar yang mendengarkan peristiwa bernama eventName. Jika listener disediakan, itu akan mengembalikan berapa kali pendengar ditemukan dalam daftar pendengar acara.
listeners(string | symbol)
listeners<E>(E)

Mengembalikan salinan array pendengar untuk peristiwa bernama 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)

Metode ini memungkinkan pemetaan melalui aliran. Fungsi fn akan dipanggil untuk setiap potongan dalam aliran. Jika fungsi fn mengembalikan promise - promise tersebut akan ed awaitsebelum diteruskan ke aliran hasil.
off(string | symbol, (args: any[]) => void)
off<E>(E, (args: ReadableEventMap[E]) => void)

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

Menambahkan fungsi listener ke akhir array pendengar untuk peristiwa bernama eventName. Tidak ada pemeriksaan yang dilakukan untuk melihat apakah listener telah ditambahkan. Beberapa panggilan yang melewati kombinasi eventName dan listener yang sama akan mengakibatkan listener ditambahkan, dan dipanggil, beberapa kali.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

Secara default, pendengar peristiwa dipanggil dalam urutan ditambahkan. Metode emitter.prependListener() dapat digunakan sebagai alternatif untuk menambahkan pendengar peristiwa ke awal array pendengar.

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)

Menambahkan fungsi satu kali untuk peristiwa bernama . Lain kali eventName dipicu, pendengar ini dihapus lalu dipanggil.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

Secara default, pendengar peristiwa dipanggil dalam urutan ditambahkan. Metode emitter.prependOnceListener() dapat digunakan sebagai alternatif untuk menambahkan pendengar peristiwa ke awal array pendengar.

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

Metode ini readable.pause() akan menyebabkan aliran dalam mode mengalir berhenti memancarkan 'data' peristiwa, beralih dari mode mengalir. Data apa pun yang tersedia akan tetap berada di buffer internal.

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

Metode ini readable.pause() tidak berpengaruh jika ada 'readable' pendengar peristiwa.
prependListener(string | symbol, (args: any[]) => void)
prependListener<E>(E, (args: ReadableEventMap[E]) => void)

Menambahkan fungsi ke awal array pendengar untuk peristiwa bernama . Tidak ada pemeriksaan yang dilakukan untuk melihat apakah listener telah ditambahkan. Beberapa panggilan yang melewati kombinasi eventName dan listener yang sama akan mengakibatkan listener ditambahkan, dan dipanggil, beberapa kali.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.
prependOnceListener(string | symbol, (args: any[]) => void)
prependOnceListener<E>(E, (args: ReadableEventMap[E]) => void)

Menambahkan fungsi satu kali untuk peristiwa bernama ke awal array pendengar. Lain kali eventName dipicu, pendengar ini dihapus, lalu dipanggil.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.
push(any, BufferEncoding)
rawListeners(string | symbol)
rawListeners<E>(E)

Mengembalikan salinan array pendengar untuk peristiwa bernama eventName, termasuk pembungkus apa pun (seperti yang dibuat oleh .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)

Metode ini readable.read() membaca data dari buffer internal dan mengembalikannya. Jika tidak ada data yang tersedia untuk dibaca, dikembalikan. null Secara default, data dikembalikan sebagai Buffer objek kecuali pengkodean telah ditentukan menggunakan metode atau readable.setEncoding() aliran beroperasi dalam mode objek.

Argumen opsional size menentukan jumlah byte tertentu untuk dibaca. Jika size byte tidak tersedia untuk dibaca, null akan dikembalikan kecuali aliran telah berakhir, dalam hal ini semua data yang tersisa di buffer internal akan dikembalikan.

Jika size argumen tidak ditentukan, semua data yang terkandung dalam buffer internal akan dikembalikan.

Argumen size harus kurang dari atau sama dengan 1 GiB.

Metode ini readable.read() hanya boleh dipanggil pada Readable aliran yang beroperasi dalam mode dijeda. Dalam mode mengalir, readable.read() dipanggil secara otomatis sampai buffer internal benar-benar terkuras.

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

Setiap panggilan ke readable.read() mengembalikan potongan data, atau null. Potongan-potongan itu tidak tergabung. Perulangan while diperlukan untuk menggunakan semua data yang saat ini berada di buffer. Saat membaca file .read() besar dapat mengembalikan null, setelah mengkonsumsi semua konten buffer sejauh ini, tetapi masih ada lebih banyak data yang akan datang belum buffer. Dalam hal ini, peristiwa baru 'readable' akan dipancarkan ketika ada lebih banyak data di buffer. Akhirnya 'end' event akan dipancarkan ketika tidak ada lagi data yang akan datang.

Oleh karena itu, untuk membaca seluruh konten file dari readable, perlu untuk mengumpulkan potongan di beberapa 'readable' peristiwa:

const chunks = [];

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

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

Aliran Readable dalam mode objek akan selalu mengembalikan satu item dari panggilan ke readable.read(size), terlepas dari nilai size argumen.

Jika metode mengembalikan readable.read() sebagian data, 'data' peristiwa juga akan dipancarkan.

Panggilan yang dibaca setelah 'end' peristiwa dipancarkan akan kembali null. Tidak ada kesalahan runtime yang akan diangkat.
reduce<T>((previous: any, data: any, options?: Abortable) => T)

Metode ini memanggil fn pada setiap potongan aliran secara berurutan, meneruskannya hasil dari perhitungan pada elemen sebelumnya. Ini mengembalikan janji untuk nilai akhir pengurangan.

Jika tidak ada nilai awal yang disediakan, potongan pertama aliran digunakan sebagai nilai awal. Jika aliran kosong, janji ditolak dengan properti dengan TypeErrorERR_INVALID_ARGS kode.

Fungsi peredam mengulangi elemen aliran demi elemen yang berarti tidak ada parameter konkurensi atau paralelisme. Untuk melakukan pengurangan secara bersamaan, Anda dapat mengekstrak fungsi asinkron ke readable.map metode.
reduce<T>((previous: T, data: any, options?: Abortable) => T, T, Abortable)
removeAllListeners(string | symbol)
removeAllListeners<E>(E)

Menghapus semua pendengar, atau pendengar dari eventNameyang ditentukan.

Adalah praktik buruk untuk menghapus listener yang ditambahkan di tempat lain dalam kode, terutama ketika instans EventEmitter dibuat oleh beberapa komponen atau modul lain (misalnya soket atau aliran file).

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.
removeListener(string | symbol, (args: any[]) => void)
removeListener<E>(E, (args: ReadableEventMap[E]) => void)

Menghapus listener yang ditentukan dari array pendengar untuk peristiwa bernama eventName.

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

removeListener() akan menghapus, paling banyak, satu instans pendengar dari array pendengar. Jika ada pendengar tunggal yang telah ditambahkan beberapa kali ke array pendengar untuk eventNameyang ditentukan, maka removeListener() harus dipanggil beberapa kali untuk menghapus setiap instans.

Setelah peristiwa dipancarkan, semua pendengar yang melekat padanya pada saat pemancaran dipanggil secara berurutan. Ini menyiratkan bahwa panggilan apa pun removeListener()removeAllListeners() memancarkan dan sebelum pemroses terakhir menyelesaikan eksekusi tidak akan menghapusnya dari emit() sedang berlangsung. Peristiwa berikutnya berulah seperti yang diharapkan.

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

Karena pemroses dikelola menggunakan array internal, memanggil ini akan mengubah indeks posisi pemroses mana pun yang terdaftar setelah pemroses dihapus. Ini tidak akan berdampak pada urutan di mana pendengar dipanggil, tetapi itu berarti bahwa salinan array pendengar seperti yang dikembalikan oleh metode emitter.listeners() perlu dibuat ulang.

Ketika satu fungsi telah ditambahkan sebagai handler beberapa kali untuk satu peristiwa (seperti dalam contoh di bawah), removeListener() akan menghapus instans yang terakhir ditambahkan. Dalam contoh pendengar once('ping') dihapus:

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.
resume()

Metode ini readable.resume() menyebabkan aliran yang dijeda Readable secara eksplisit untuk melanjutkan peristiwa pemancaran 'data' , mengalihkan aliran ke mode mengalir.

Metode ini readable.resume() dapat digunakan untuk sepenuhnya mengonsumsi data dari aliran tanpa benar-benar memproses data tersebut:

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

Metode ini readable.resume() tidak berpengaruh jika ada 'readable' pendengar peristiwa.
setEncoding(BufferEncoding)

Metode ini readable.setEncoding() mengatur pengkodean karakter untuk data yang dibaca dari Readable aliran.

Secara default, tidak ada pengkodean yang ditetapkan dan data aliran akan dikembalikan sebagai Buffer objek. Mengatur pengodean menyebabkan data aliran dikembalikan sebagai string pengodean yang ditentukan, bukan sebagai Buffer objek. Misalnya, memanggil readable.setEncoding('utf8') akan menyebabkan data output ditafsirkan sebagai data UTF-8, dan diteruskan sebagai string. Memanggil readable.setEncoding('hex') akan menyebabkan data dikodekan dalam format string heksadesimal.

Aliran Readable akan menangani karakter multi-byte yang dikirimkan melalui aliran dengan benar yang akan didekodekan secara tidak benar jika hanya ditarik dari aliran sebagai Buffer objek.

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

Metode ini mirip dengan Array.prototype.some dan memanggil fn pada setiap potongan dalam aliran hingga nilai pengembalian yang ditunggu-tunggu adalah true (atau nilai kebenaran apa pun). Setelah panggilan fn pada awaitnilai pengembalian chunk ed benar, aliran dihancurkan dan janji dipenuhi dengan true. Jika tidak ada panggilan fn pada potongan yang mengembalikan nilai yang benar, janji itu dipenuhi dengan false.
take(number, Abortable)

Metode ini mengembalikan aliran baru dengan potongan batas pertama.
toArray(Abortable)

Metode ini memungkinkan dengan mudah mendapatkan konten aliran.

Karena metode ini membaca seluruh aliran ke dalam memori, metode ini meniadakan manfaat aliran. Ini ditujukan untuk interoperabilitas dan kenyamanan, bukan sebagai cara utama untuk mengonsumsi streaming.
unpipe(WritableStream)

Metode ini readable.unpipe() melepaskan aliran yang Writable sebelumnya terpasang menggunakan metode pipa .

Jika tidak destination ditentukan, maka semua pipa terlepas.

Jika ditentukan destination , tetapi tidak ada pipa yang disiapkan untuk itu, maka metode tersebut tidak melakukan apa-apa.

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)

Melewati chunk sebagai null sinyal akhir aliran (EOF) dan berperilaku sama dengan readable.push(null), setelah itu tidak ada lagi data yang dapat ditulis. Sinyal EOF diletakkan di ujung buffer dan data buffer apa pun akan tetap disiram.

Metode ini readable.unshift() mendorong sebagian data kembali ke buffer internal. Ini berguna dalam situasi tertentu di mana aliran dikonsumsi oleh kode yang perlu "membatalkan konsumsi" sejumlah data yang telah ditarik secara optimis dari sumbernya, sehingga data dapat diteruskan ke pihak lain.

Metode ini stream.unshift(chunk) tidak dapat dipanggil setelah 'end' peristiwa dipancarkan atau kesalahan runtime akan dilemparkan.

Pengembang yang sering menggunakan stream.unshift() harus mempertimbangkan untuk beralih ke penggunaan Transform aliran sebagai gantinya. Lihat bagian untuk API for stream implementers informasi lebih lanjut.

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

Tidak seperti push, stream.unshift(chunk) tidak akan mengakhiri proses membaca dengan mengatur ulang status pembacaan internal aliran. Hal ini dapat menyebabkan hasil yang tidak terduga jika readable.unshift() dipanggil selama pembacaan (yaitu dari dalam implementasi _read pada aliran kustom). Mengikuti panggilan ke readable.unshift() dengan dorongan langsung akan mengatur ulang status pembacaan dengan tepat, namun yang terbaik adalah menghindari panggilan readable.unshift() saat dalam proses melakukan pembacaan.
wrap(ReadableStream)

Sebelum Node.js 0.10, streaming tidak mengimplementasikan seluruh node:stream API modul seperti yang ditentukan saat ini. (Lihat Compatibility untuk informasi lebih lanjut.)

Saat menggunakan pustaka Node.js lama yang memancarkan 'data' peristiwa dan memiliki metode jeda yang hanya bersifat pemberitahuan, metode tersebut readable.wrap() dapat digunakan untuk membuat Readable aliran yang menggunakan aliran lama sebagai sumber datanya.

Jarang perlu digunakan readable.wrap() tetapi metode ini telah disediakan sebagai kenyamanan untuk berinteraksi dengan aplikasi dan perpustakaan Node.js lama.

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

Menghubungi readable.destroy() dan AbortError mengembalikan janji yang terpenuhi ketika siaran selesai.
[asyncIterator]()
[toAsyncStreamable]()

Saat --experimental-stream-iter bendera diaktifkan, Readable streaming mengimplementasikan Stream.toAsyncStreamable protokol, memungkinkan konsumsi yang efisien oleh stream/iter API.

Ini menyediakan iterator asinkron batch yang menguras buffer internal aliran ke dalam Uint8Array[] batch, mengamortisasi overhead Promise per potongan dari jalur standar Symbol.asyncIterator . Untuk aliran mode byte, potongan dihasilkan secara langsung sebagai Buffer instans (yang merupakan Uint8Array subkelas). Untuk aliran mode objek atau yang dikodekan, setiap potongan dinormalisasi sebelum Uint8Array pengelompokan.

Iterator yang dikembalikan ditandai sebagai sumber yang divalidasi, sehingga from() melewatinya tanpa normalisasi tambahan.

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'

Tanpa --experimental-stream-iter bendera, memanggil metode ini melempar ERR_STREAM_ITER_MISSING_FLAG.

Metode yang Diwarisi

eventNames()

Mengembalikan array yang mencantumkan peristiwa di mana pemancar telah mendaftarkan pendengar.

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

Mengembalikan nilai pendengar maks saat ini untuk EventEmitter yang diatur oleh emitter.setMaxListeners(n) atau default ke events.defaultMaxListeners.
pipe<T>(T, PipeOptions)
setMaxListeners(number)

Secara default EventEmitterakan mencetak peringatan jika lebih dari 10 pendengar ditambahkan untuk peristiwa tertentu. Ini adalah default yang berguna yang membantu menemukan kebocoran memori. Metode emitter.setMaxListeners() memungkinkan batas untuk dimodifikasi untuk instans EventEmitter spesifik ini. Nilai dapat diatur ke Infinity (atau 0) untuk menunjukkan jumlah listener yang tidak terbatas.

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.
[captureRejectionSymbol](Error, string | symbol, any[])

Metode ini Symbol.for('nodejs.rejection') dipanggil jika penolakan janji terjadi saat memancarkan peristiwa dan captureRejections diaktifkan pada emitor. Dimungkinkan untuk digunakan events.captureRejectionSymbol sebagai pengganti 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.
  }
}

Detail Properti

closed

Adalah true setelah 'close' telah dipancarkan.

closed: boolean

Nilai Properti

boolean

destroyed

Adalah true setelah readable.destroy() telah dipanggil.

destroyed: boolean

Nilai Properti

boolean

errored

Mengembalikan kesalahan jika aliran telah dihancurkan dengan kesalahan.

errored: Error | null

Nilai Properti

Error | null

readable

Apakah true jika aman untuk memanggil baca, yang berarti aliran belum dihancurkan atau dipancarkan 'error' atau 'end'.

readable: boolean

Nilai Properti

boolean

readableAborted

Mengembalikan apakah aliran dihancurkan atau error sebelum dipancarkan 'end'.

readableAborted: boolean

Nilai Properti

boolean

readableDidRead

Mengembalikan apakah 'data' telah dipancarkan.

readableDidRead: boolean

Nilai Properti

boolean

readableEncoding

Pengambil untuk properti encoding aliran tertentu Readable . Properti encoding dapat diatur menggunakan metode setEncoding .

readableEncoding: BufferEncoding | null

Nilai Properti

BufferEncoding | null

readableEnded

Menjadi true ketika 'end' peristiwa dipancarkan.

readableEnded: boolean

Nilai Properti

boolean

readableFlowing

Properti ini mencerminkan status Readable aliran saat ini seperti yang dijelaskan di bagian Tiga status .

readableFlowing: boolean | null

Nilai Properti

boolean | null

readableHighWaterMark

Mengembalikan nilai lulus highWaterMark saat membuat ini Readable.

readableHighWaterMark: number

Nilai Properti

number

readableLength

Properti ini berisi jumlah byte (atau objek) dalam antrean yang siap dibaca. Nilai tersebut memberikan data introspeksi mengenai status highWaterMark.

readableLength: number

Nilai Properti

number

readableObjectMode

Pengambil untuk properti objectMode aliran tertentu Readable .

readableObjectMode: boolean

Nilai Properti

boolean

Detail Metode

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Alias untuk emitter.on(eventName, listener).

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

Parameter

eventName

E

listener

(args: ReadableEventMap[E]) => void

Mengembalikan

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

Metode ini juga memungkinkan untuk AbortSignal disediakan, yang akan menghancurkan aliran yang disusun saat dibatalkan.

Lihat stream.compose(...streams) untuk informasi lebih lanjut.

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

Parameter

stream

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

options

Abortable

Mengembalikan

Duplex

aliran yang disusun dengan aliran stream.

destroy(Error)

Hancurkan alirannya. Secara opsional memancarkan peristiwa 'error' , dan memancarkan 'close' peristiwa (kecuali diatur emitClose ke false). Setelah panggilan ini, aliran yang dapat dibaca akan melepaskan sumber daya internal apa pun dan panggilan berikutnya akan push() diabaikan.

Setelah destroy() dipanggil, panggilan lebih lanjut akan menjadi no-op dan tidak ada kesalahan lebih lanjut kecuali dari _destroy() yang dapat dipancarkan sebagai 'error'.

Pelaksana tidak boleh mengganti metode ini, melainkan menerapkan readable._destroy().

function destroy(error?: Error): this

Parameter

error

Error

Kesalahan yang akan diteruskan sebagai payload dalam 'error' peristiwa

Mengembalikan

this

drop(number, Abortable)

Metode ini mengembalikan aliran baru dengan potongan batas pertama yang dijatuhkan dari awal.

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

Parameter

limit

number

jumlah potongan yang akan dihilangkan dari yang dapat dibaca.

options

Abortable

Mengembalikan

Readable

aliran dengan potongan batas yang dijatuhkan sejak awal.

emit(string | symbol, any[]) 

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

Parameter

eventName

string | symbol

args

any[]

Mengembalikan

boolean

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

Secara sinkron memanggil masing-masing pendengar yang terdaftar untuk peristiwa bernama eventName, dalam urutan terdaftar, meneruskan argumen yang disediakan ke masing-masing.

Mengembalikan true jika peristiwa memiliki pendengar, false sebaliknya.

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

Parameter

eventName

E

args

ReadableEventMap[E]

Mengembalikan

boolean

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

Metode ini mirip dengan Array.prototype.every dan memanggil fn pada setiap potongan dalam aliran untuk memeriksa apakah semua nilai pengembalian yang ditunggu adalah nilai kebenaran untuk fn. Setelah panggilan fn pada awaitnilai pengembalian chunk ed palsu, aliran dihancurkan dan janji dipenuhi dengan false. Jika semua panggilan fn pada potongan mengembalikan nilai yang benar, janji dipenuhi dengan true.

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

Parameter

fn

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

fungsi untuk memanggil setiap bagian aliran. Asinkron atau tidak.

options

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

Mengembalikan

Promise<boolean>

Janji yang mengevaluasi jika trueFN mengembalikan nilai yang benar untuk setiap potongan.

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

Metode ini memungkinkan pemfilteran aliran. Untuk setiap potongan dalam aliran, fungsi fn akan dipanggil dan jika mengembalikan nilai yang benar, potongan akan diteruskan ke aliran hasil. Jika fungsi fn mengembalikan promise - promise tersebut akan ed await.

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

Parameter

fn

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

fungsi untuk memfilter potongan dari aliran. Asinkron atau tidak.

options

ReadableOperatorOptions

Mengembalikan

Readable

aliran yang disaring dengan predikat 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>

Parameter

fn

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

options

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

Mengembalikan

Promise<any>

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

Metode ini mirip dengan Array.prototype.find dan memanggil fn pada setiap potongan dalam aliran untuk menemukan potongan dengan nilai kebenaran untuk fn. Setelah nilai pengembalian panggilan fn yang ditunggu-tunggu adalah kebenaran, aliran dihancurkan dan janji dipenuhi dengan nilai yang fn mengembalikan nilai yang benar. Jika semua panggilan fn pada potongan mengembalikan nilai palsu, janji dipenuhi dengan undefined.

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

Parameter

fn

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

fungsi untuk memanggil setiap bagian aliran. Asinkron atau tidak.

options

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

Mengembalikan

Promise<T | undefined>

Janji yang mengevaluasi ke potongan pertama yang dievaluasi FN dengan nilai kebenaran, atau undefined jika tidak ada elemen yang ditemukan.

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

Metode ini mengembalikan aliran baru dengan menerapkan callback yang diberikan ke setiap bagian aliran dan kemudian meratakan hasilnya.

Dimungkinkan untuk mengembalikan aliran atau iterable atau asinkron lain yang dapat diulang dari fn dan aliran hasil akan digabungkan (diratakan) ke dalam aliran yang dikembalikan.

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

Parameter

fn

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

Fungsi untuk memetakan setiap potongan dalam aliran. Mungkin asinkron. Mungkin aliran atau generator.

options

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

Mengembalikan

Readable

aliran yang dipetakan datar dengan fungsi fn.

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

Metode ini memungkinkan iterasi aliran. Untuk setiap potongan dalam aliran, fungsi fn akan dipanggil. Jika fungsi fn mengembalikan promise - promise tersebut akan ed await.

Metode ini berbeda dari for await...of loop karena dapat secara opsional memproses potongan secara bersamaan. Selain itu, iterasi forEach hanya dapat dihentikan dengan melewati signal opsi dan membatalkan AbortController terkait sementara for await...of dapat dihentikan dengan break atau return. Dalam kedua kasus tersebut, aliran akan dihancurkan.

Metode ini berbeda dengan mendengarkan 'data' peristiwa karena menggunakan readable peristiwa dalam mesin yang mendasarinya dan dapat membatasi jumlah panggilan fn bersamaan.

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

Parameter

fn

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

fungsi untuk memanggil setiap bagian aliran. Asinkron atau tidak.

options

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

Mengembalikan

Promise<void>

janji ketika aliran selesai.

isPaused()

Metode ini readable.isPaused() mengembalikan status operasi saat ini dari .Readable Ini digunakan terutama oleh mekanisme yang mendasari metode ini readable.pipe() . Dalam kebanyakan kasus umum, tidak akan ada alasan untuk menggunakan metode ini secara langsung.

const readable = new stream.Readable();

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

function isPaused(): boolean

Mengembalikan

boolean

iterator(ReadableIteratorOptions)

Iterator yang dibuat dengan metode ini memberi pengguna opsi untuk membatalkan penghancuran aliran jika for await...of loop keluar oleh return, break, atau throw, atau jika iterator harus menghancurkan aliran jika aliran memancarkan kesalahan selama iterasi.

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

Parameter

options

ReadableIteratorOptions

Mengembalikan

AsyncIterator<any>

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

number

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

Mengembalikan jumlah pendengar yang mendengarkan peristiwa bernama eventName. Jika listener disediakan, itu akan mengembalikan berapa kali pendengar ditemukan dalam daftar pendengar acara.

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

Parameter

eventName

E

Nama peristiwa yang sedang didengarkan

listener

(args: ReadableEventMap[E]) => void

Fungsi penanganan aktivitas

Mengembalikan

number

listeners(string | symbol) 

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

Parameter

eventName

string | symbol

Mengembalikan

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

listeners<E>(E)

Mengembalikan salinan array pendengar untuk peristiwa bernama 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[]

Parameter

eventName

E

Mengembalikan

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

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

Metode ini memungkinkan pemetaan melalui aliran. Fungsi fn akan dipanggil untuk setiap potongan dalam aliran. Jika fungsi fn mengembalikan promise - promise tersebut akan ed awaitsebelum diteruskan ke aliran hasil.

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

Parameter

fn

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

Fungsi untuk memetakan setiap potongan dalam aliran. Asinkron atau tidak.

options

ReadableOperatorOptions

Mengembalikan

Readable

aliran yang dipetakan dengan fungsi fn.

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Alias untuk emitter.removeListener().

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

Parameter

eventName

E

listener

(args: ReadableEventMap[E]) => void

Mengembalikan

this

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Menambahkan fungsi listener ke akhir array pendengar untuk peristiwa bernama eventName. Tidak ada pemeriksaan yang dilakukan untuk melihat apakah listener telah ditambahkan. Beberapa panggilan yang melewati kombinasi eventName dan listener yang sama akan mengakibatkan listener ditambahkan, dan dipanggil, beberapa kali.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

Secara default, pendengar peristiwa dipanggil dalam urutan ditambahkan. Metode emitter.prependListener() dapat digunakan sebagai alternatif untuk menambahkan pendengar peristiwa ke awal array pendengar.

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

Parameter

eventName

E

Nama acara.

listener

(args: ReadableEventMap[E]) => void

Fungsi panggilan balik

Mengembalikan

this

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Menambahkan fungsi satu kali untuk peristiwa bernama . Lain kali eventName dipicu, pendengar ini dihapus lalu dipanggil.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

Secara default, pendengar peristiwa dipanggil dalam urutan ditambahkan. Metode emitter.prependOnceListener() dapat digunakan sebagai alternatif untuk menambahkan pendengar peristiwa ke awal array pendengar.

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

Parameter

eventName

E

Nama acara.

listener

(args: ReadableEventMap[E]) => void

Fungsi panggilan balik

Mengembalikan

this

pause()

Metode ini readable.pause() akan menyebabkan aliran dalam mode mengalir berhenti memancarkan 'data' peristiwa, beralih dari mode mengalir. Data apa pun yang tersedia akan tetap berada di buffer internal.

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

Metode ini readable.pause() tidak berpengaruh jika ada 'readable' pendengar peristiwa.

function pause(): this

Mengembalikan

this

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Menambahkan fungsi ke awal array pendengar untuk peristiwa bernama . Tidak ada pemeriksaan yang dilakukan untuk melihat apakah listener telah ditambahkan. Beberapa panggilan yang melewati kombinasi eventName dan listener yang sama akan mengakibatkan listener ditambahkan, dan dipanggil, beberapa kali.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

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

Parameter

eventName

E

Nama acara.

listener

(args: ReadableEventMap[E]) => void

Fungsi panggilan balik

Mengembalikan

this

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Menambahkan fungsi satu kali untuk peristiwa bernama ke awal array pendengar. Lain kali eventName dipicu, pendengar ini dihapus, lalu dipanggil.

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

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

Parameter

eventName

E

Nama acara.

listener

(args: ReadableEventMap[E]) => void

Fungsi panggilan balik

Mengembalikan

this

push(any, BufferEncoding) 

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

Parameter

chunk

any

encoding

BufferEncoding

Mengembalikan

boolean

rawListeners(string | symbol) 

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

Parameter

eventName

string | symbol

Mengembalikan

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

rawListeners<E>(E)

Mengembalikan salinan array pendengar untuk peristiwa bernama eventName, termasuk pembungkus apa pun (seperti yang dibuat oleh .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[]

Parameter

eventName

E

Mengembalikan

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

read(number)

Metode ini readable.read() membaca data dari buffer internal dan mengembalikannya. Jika tidak ada data yang tersedia untuk dibaca, dikembalikan. null Secara default, data dikembalikan sebagai Buffer objek kecuali pengkodean telah ditentukan menggunakan metode atau readable.setEncoding() aliran beroperasi dalam mode objek.

Argumen opsional size menentukan jumlah byte tertentu untuk dibaca. Jika size byte tidak tersedia untuk dibaca, null akan dikembalikan kecuali aliran telah berakhir, dalam hal ini semua data yang tersisa di buffer internal akan dikembalikan.

Jika size argumen tidak ditentukan, semua data yang terkandung dalam buffer internal akan dikembalikan.

Argumen size harus kurang dari atau sama dengan 1 GiB.

Metode ini readable.read() hanya boleh dipanggil pada Readable aliran yang beroperasi dalam mode dijeda. Dalam mode mengalir, readable.read() dipanggil secara otomatis sampai buffer internal benar-benar terkuras.

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

Setiap panggilan ke readable.read() mengembalikan potongan data, atau null. Potongan-potongan itu tidak tergabung. Perulangan while diperlukan untuk menggunakan semua data yang saat ini berada di buffer. Saat membaca file .read() besar dapat mengembalikan null, setelah mengkonsumsi semua konten buffer sejauh ini, tetapi masih ada lebih banyak data yang akan datang belum buffer. Dalam hal ini, peristiwa baru 'readable' akan dipancarkan ketika ada lebih banyak data di buffer. Akhirnya 'end' event akan dipancarkan ketika tidak ada lagi data yang akan datang.

Oleh karena itu, untuk membaca seluruh konten file dari readable, perlu untuk mengumpulkan potongan di beberapa 'readable' peristiwa:

const chunks = [];

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

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

Aliran Readable dalam mode objek akan selalu mengembalikan satu item dari panggilan ke readable.read(size), terlepas dari nilai size argumen.

Jika metode mengembalikan readable.read() sebagian data, 'data' peristiwa juga akan dipancarkan.

Panggilan yang dibaca setelah 'end' peristiwa dipancarkan akan kembali null. Tidak ada kesalahan runtime yang akan diangkat.

function read(size?: number): any

Parameter

size

number

Argumen opsional untuk menentukan berapa banyak data yang akan dibaca.

Mengembalikan

any

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

Metode ini memanggil fn pada setiap potongan aliran secara berurutan, meneruskannya hasil dari perhitungan pada elemen sebelumnya. Ini mengembalikan janji untuk nilai akhir pengurangan.

Jika tidak ada nilai awal yang disediakan, potongan pertama aliran digunakan sebagai nilai awal. Jika aliran kosong, janji ditolak dengan properti dengan TypeErrorERR_INVALID_ARGS kode.

Fungsi peredam mengulangi elemen aliran demi elemen yang berarti tidak ada parameter konkurensi atau paralelisme. Untuk melakukan pengurangan secara bersamaan, Anda dapat mengekstrak fungsi asinkron ke readable.map metode.

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

Parameter

fn

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

fungsi peredam untuk memanggil setiap potongan dalam aliran. Asinkron atau tidak.

Mengembalikan

Promise<T>

janji untuk nilai akhir pengurangan.

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>

Parameter

fn

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

initial

T

options

Abortable

Mengembalikan

Promise<T>

removeAllListeners(string | symbol) 

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

Parameter

eventName

string | symbol

Mengembalikan

this

removeAllListeners<E>(E)

Menghapus semua pendengar, atau pendengar dari eventNameyang ditentukan.

Adalah praktik buruk untuk menghapus listener yang ditambahkan di tempat lain dalam kode, terutama ketika instans EventEmitter dibuat oleh beberapa komponen atau modul lain (misalnya soket atau aliran file).

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

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

Parameter

eventName

E

Mengembalikan

this

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

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

Parameter

eventName

string | symbol

listener

(args: any[]) => void

Mengembalikan

this

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

Menghapus listener yang ditentukan dari array pendengar untuk peristiwa bernama eventName.

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

removeListener() akan menghapus, paling banyak, satu instans pendengar dari array pendengar. Jika ada pendengar tunggal yang telah ditambahkan beberapa kali ke array pendengar untuk eventNameyang ditentukan, maka removeListener() harus dipanggil beberapa kali untuk menghapus setiap instans.

Setelah peristiwa dipancarkan, semua pendengar yang melekat padanya pada saat pemancaran dipanggil secara berurutan. Ini menyiratkan bahwa panggilan apa pun removeListener()removeAllListeners() memancarkan dan sebelum pemroses terakhir menyelesaikan eksekusi tidak akan menghapusnya dari emit() sedang berlangsung. Peristiwa berikutnya berulah seperti yang diharapkan.

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

Karena pemroses dikelola menggunakan array internal, memanggil ini akan mengubah indeks posisi pemroses mana pun yang terdaftar setelah pemroses dihapus. Ini tidak akan berdampak pada urutan di mana pendengar dipanggil, tetapi itu berarti bahwa salinan array pendengar seperti yang dikembalikan oleh metode emitter.listeners() perlu dibuat ulang.

Ketika satu fungsi telah ditambahkan sebagai handler beberapa kali untuk satu peristiwa (seperti dalam contoh di bawah), removeListener() akan menghapus instans yang terakhir ditambahkan. Dalam contoh pendengar once('ping') dihapus:

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

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

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

Parameter

eventName

E

listener

(args: ReadableEventMap[E]) => void

Mengembalikan

this

resume()

Metode ini readable.resume() menyebabkan aliran yang dijeda Readable secara eksplisit untuk melanjutkan peristiwa pemancaran 'data' , mengalihkan aliran ke mode mengalir.

Metode ini readable.resume() dapat digunakan untuk sepenuhnya mengonsumsi data dari aliran tanpa benar-benar memproses data tersebut:

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

Metode ini readable.resume() tidak berpengaruh jika ada 'readable' pendengar peristiwa.

function resume(): this

Mengembalikan

this

setEncoding(BufferEncoding)

Metode ini readable.setEncoding() mengatur pengkodean karakter untuk data yang dibaca dari Readable aliran.

Secara default, tidak ada pengkodean yang ditetapkan dan data aliran akan dikembalikan sebagai Buffer objek. Mengatur pengodean menyebabkan data aliran dikembalikan sebagai string pengodean yang ditentukan, bukan sebagai Buffer objek. Misalnya, memanggil readable.setEncoding('utf8') akan menyebabkan data output ditafsirkan sebagai data UTF-8, dan diteruskan sebagai string. Memanggil readable.setEncoding('hex') akan menyebabkan data dikodekan dalam format string heksadesimal.

Aliran Readable akan menangani karakter multi-byte yang dikirimkan melalui aliran dengan benar yang akan didekodekan secara tidak benar jika hanya ditarik dari aliran sebagai Buffer objek.

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

Parameter

encoding

BufferEncoding

Pengodean yang akan digunakan.

Mengembalikan

this

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

Metode ini mirip dengan Array.prototype.some dan memanggil fn pada setiap potongan dalam aliran hingga nilai pengembalian yang ditunggu-tunggu adalah true (atau nilai kebenaran apa pun). Setelah panggilan fn pada awaitnilai pengembalian chunk ed benar, aliran dihancurkan dan janji dipenuhi dengan true. Jika tidak ada panggilan fn pada potongan yang mengembalikan nilai yang benar, janji itu dipenuhi dengan false.

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

Parameter

fn

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

fungsi untuk memanggil setiap bagian aliran. Asinkron atau tidak.

options

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

Mengembalikan

Promise<boolean>

Janji yang mengevaluasi jika trueFN mengembalikan nilai kebenaran untuk setidaknya satu potongan.

take(number, Abortable)

Metode ini mengembalikan aliran baru dengan potongan batas pertama.

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

Parameter

limit

number

jumlah potongan yang akan diambil dari yang dapat dibaca.

options

Abortable

Mengembalikan

Readable

aliran dengan potongan batas yang diambil.

toArray(Abortable)

Metode ini memungkinkan dengan mudah mendapatkan konten aliran.

Karena metode ini membaca seluruh aliran ke dalam memori, metode ini meniadakan manfaat aliran. Ini ditujukan untuk interoperabilitas dan kenyamanan, bukan sebagai cara utama untuk mengonsumsi streaming.

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

Parameter

options

Abortable

Mengembalikan

Promise<any[]>

Janji yang berisi array dengan konten aliran.

unpipe(WritableStream)

Metode ini readable.unpipe() melepaskan aliran yang Writable sebelumnya terpasang menggunakan metode pipa .

Jika tidak destination ditentukan, maka semua pipa terlepas.

Jika ditentukan destination , tetapi tidak ada pipa yang disiapkan untuk itu, maka metode tersebut tidak melakukan apa-apa.

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

Parameter

destination

WritableStream

Aliran khusus opsional untuk membuka pipa

Mengembalikan

this

unshift(any, BufferEncoding)

Melewati chunk sebagai null sinyal akhir aliran (EOF) dan berperilaku sama dengan readable.push(null), setelah itu tidak ada lagi data yang dapat ditulis. Sinyal EOF diletakkan di ujung buffer dan data buffer apa pun akan tetap disiram.

Metode ini readable.unshift() mendorong sebagian data kembali ke buffer internal. Ini berguna dalam situasi tertentu di mana aliran dikonsumsi oleh kode yang perlu "membatalkan konsumsi" sejumlah data yang telah ditarik secara optimis dari sumbernya, sehingga data dapat diteruskan ke pihak lain.

Metode ini stream.unshift(chunk) tidak dapat dipanggil setelah 'end' peristiwa dipancarkan atau kesalahan runtime akan dilemparkan.

Pengembang yang sering menggunakan stream.unshift() harus mempertimbangkan untuk beralih ke penggunaan Transform aliran sebagai gantinya. Lihat bagian untuk API for stream implementers informasi lebih lanjut.

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

Tidak seperti push, stream.unshift(chunk) tidak akan mengakhiri proses membaca dengan mengatur ulang status pembacaan internal aliran. Hal ini dapat menyebabkan hasil yang tidak terduga jika readable.unshift() dipanggil selama pembacaan (yaitu dari dalam implementasi _read pada aliran kustom). Mengikuti panggilan ke readable.unshift() dengan dorongan langsung akan mengatur ulang status pembacaan dengan tepat, namun yang terbaik adalah menghindari panggilan readable.unshift() saat dalam proses melakukan pembacaan.

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

Parameter

chunk

any

Potongan data untuk dihapus ke antrean baca. Untuk aliran yang tidak beroperasi dalam mode objek, chunk harus berupa {string}, {Buffer}, {TypedArray}, {DataView} atau null. Untuk aliran mode objek, chunk mungkin berupa nilai JavaScript apa pun.

encoding

BufferEncoding

Pengkodean potongan string. Harus berupa pengkodean yang valid Buffer , seperti 'utf8' atau 'ascii'.

wrap(ReadableStream)

Sebelum Node.js 0.10, streaming tidak mengimplementasikan seluruh node:stream API modul seperti yang ditentukan saat ini. (Lihat Compatibility untuk informasi lebih lanjut.)

Saat menggunakan pustaka Node.js lama yang memancarkan 'data' peristiwa dan memiliki metode jeda yang hanya bersifat pemberitahuan, metode tersebut readable.wrap() dapat digunakan untuk membuat Readable aliran yang menggunakan aliran lama sebagai sumber datanya.

Jarang perlu digunakan readable.wrap() tetapi metode ini telah disediakan sebagai kenyamanan untuk berinteraksi dengan aplikasi dan perpustakaan Node.js lama.

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

Parameter

stream

ReadableStream

Aliran "gaya lama" yang dapat dibaca

Mengembalikan

this

[asyncDispose]()

Menghubungi readable.destroy() dan AbortError mengembalikan janji yang terpenuhi ketika siaran selesai.

function [asyncDispose](): Promise<void>

Mengembalikan

Promise<void>

[asyncIterator]() 

function [asyncIterator](): AsyncIterator<any>

Mengembalikan

AsyncIterator<any>

AsyncIterator untuk mengkonsumsi aliran sepenuhnya.

[toAsyncStreamable]()

Saat --experimental-stream-iter bendera diaktifkan, Readable streaming mengimplementasikan Stream.toAsyncStreamable protokol, memungkinkan konsumsi yang efisien oleh stream/iter API.

Ini menyediakan iterator asinkron batch yang menguras buffer internal aliran ke dalam Uint8Array[] batch, mengamortisasi overhead Promise per potongan dari jalur standar Symbol.asyncIterator . Untuk aliran mode byte, potongan dihasilkan secara langsung sebagai Buffer instans (yang merupakan Uint8Array subkelas). Untuk aliran mode objek atau yang dikodekan, setiap potongan dinormalisasi sebelum Uint8Array pengelompokan.

Iterator yang dikembalikan ditandai sebagai sumber yang divalidasi, sehingga from() melewatinya tanpa normalisasi tambahan.

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'

Tanpa --experimental-stream-iter bendera, memanggil metode ini melempar ERR_STREAM_ITER_MISSING_FLAG.

function [toAsyncStreamable](): ByteReadableStream

Mengembalikan

ByteReadableStream

Detail Metode yang Diwarisi

eventNames()

Mengembalikan array yang mencantumkan peristiwa di mana pemancar telah mendaftarkan pendengar.

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

Mengembalikan

(string | symbol)[]

Diwarisi dari Stream.eventNames

getMaxListeners()

Mengembalikan nilai pendengar maks saat ini untuk EventEmitter yang diatur oleh emitter.setMaxListeners(n) atau default ke events.defaultMaxListeners.

function getMaxListeners(): number

Mengembalikan

number

Diwarisi dari Stream.getMaxPendengar

pipe<T>(T, PipeOptions) 

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

Parameter

destination

T

options

PipeOptions

Mengembalikan

T

Diwarisi dari Stream.pipe

setMaxListeners(number)

Secara default EventEmitterakan mencetak peringatan jika lebih dari 10 pendengar ditambahkan untuk peristiwa tertentu. Ini adalah default yang berguna yang membantu menemukan kebocoran memori. Metode emitter.setMaxListeners() memungkinkan batas untuk dimodifikasi untuk instans EventEmitter spesifik ini. Nilai dapat diatur ke Infinity (atau 0) untuk menunjukkan jumlah listener yang tidak terbatas.

Mengembalikan referensi ke EventEmitter, sehingga panggilan dapat ditautkan.

function setMaxListeners(n: number): this

Parameter

n

number

Mengembalikan

this

Diwarisi dari Stream.setMaxPendengar

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

Metode ini Symbol.for('nodejs.rejection') dipanggil jika penolakan janji terjadi saat memancarkan peristiwa dan captureRejections diaktifkan pada emitor. Dimungkinkan untuk digunakan events.captureRejectionSymbol sebagai pengganti 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[])

Parameter

error

Error

event

string | symbol

args

any[]

Diwarisi dari Stream. [captureRejectionSymbol]