Readable class

new Readable(options?: ReadableOptions<Readable>)

參數

它 true 已經被發射出來 'close' 。

closed: boolean

屬性值

它 true 已經被叫去 readable.destroy() 了。

destroyed: boolean

屬性值

如果串流被破壞並出現錯誤，則會回傳錯誤。

errored: null | Error

屬性值

是true是否可以安全地呼叫讀取，表示串流尚未被銷毀或發射'error''end'。

readable: boolean

屬性值

回傳 是否在發射 'end'前流被摧毀或錯誤。

readableAborted: boolean

屬性值

回傳是否 'data' 已發射。

readableDidRead: boolean

屬性值

給定encoding溪流的性質Readable的集積子。 encoding此屬性可使用 setEncoding 方法設定。

readableEncoding: null | BufferEncoding

屬性值

當事件被發射時，會變成true'end'。

readableEnded: boolean

屬性值

此特性反映溪流當前狀態 Readable ，如三 態 章節所述。

readableFlowing: null | boolean

屬性值

在建立此 highWaterMark時回傳 的 Readable 已通過值。

readableHighWaterMark: number

屬性值

此屬性包含佇列中準備讀取的位元組（或物件）數量。 該值提供關於 狀態 highWaterMark的內省資料。

readableLength: number

屬性值

給定objectMode溪流的性質Readable的集積子。

readableObjectMode: boolean

屬性值

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

參數

傳回

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

參數

傳回

import { Readable } from 'node:stream';

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

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

const wordsStream = Readable.from(['text passed through', 'composed stream']).compose(splitToWords);
const words = await wordsStream.toArray();

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

readable.compose(s) 等價於 stream.compose(readable, s)。

此方法也允許提供 ， AbortSignal 當中止時會銷毀已組合的串流。

如需相關資訊，請參閱 stream.compose(...streams) 。

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

參數

傳回

一個由 stream流 組成的流。

終結數據流。 可選擇性地 發射事件 'error' ，並 發射事件 'close' （除非 emitClose 設定為 false）。 呼叫結束後，可讀串流會釋放所有內部資源，後續的呼叫 push() 將被忽略。

一旦 destroy() 被呼叫，任何後續呼叫將為 no-op 且除了 外 _destroy() 不會再發出錯誤，為 'error'。

實作者不應覆寫此方法，而是實作 readable._destroy()。

function destroy(error?: Error): Readable

參數

傳回

此方法回傳一個新串流，並從一開始就丟棄了第一個 限制 區塊。

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

參數

傳回

一個從一開始就有 限制 區塊的串流。

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

參數

傳回

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

參數

傳回

此方法類似 Array.prototype.every 於 ，且在串流中每個區塊呼叫 fn ，以檢查所有等待的回傳值是否都是 fn 的真實值。 一旦 fn 呼叫區塊 awaited 返回值為錯誤，串流即被銷毀，承諾以 完成。false 如果所有對區塊的 fn 呼叫都回傳真值，則承諾在 中被實現。true

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

參數

傳回

一個承諾，評估 truefn 是否對每個區塊回傳真值。

此方法可對串流進行過濾。 對於串流中的每個區塊，都會呼叫 fn 函式，如果回傳 truey 值，該區塊會傳給結果串流。 如果 fn 函數回傳一個承諾，那個承諾就是 awaited。

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

參數

傳回

一個以 fn 為謂詞過濾的串流。

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

參數

傳回

此方法類似 Array.prototype.find 於 ，並呼叫串流中每個區塊的 fn ，以尋找具有 fn 真值的區塊。 一旦 fn 呼叫的等待回傳值為真值，串流即被銷毀，承諾以 fn 回傳真值的值實現。 若區塊上所有 fn 呼叫均回傳假值，則承諾以 實現。undefined

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

參數

傳回

一個 promise 對第一個 fn 以 truey 值的 ch，或 undefined 若未找到元素，則計算該 promise。

此方法透過對串流的每個區塊套用回調，然後將結果平整，回傳一個新的串流。

可以從 fn 回傳一個串流或其他可迭代或非同步迭代，結果串流會合併（扁平化）到回傳的串流中。

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

參數

傳回

一個以 函數 fn 平映的串流。

此方法允許迭代串流。 串流中的每個區塊都會被呼叫 fn 函式。 如果 fn 函數回傳一個承諾，那個承諾就是 awaited。

此方法與 for await...of 迴圈不同，因為它可選擇性地同時處理區塊。 此外，forEach迭代只能透過通過signal選項並中止相關的 AbortController 來停止，而 for await...of 可以以 或 break停止return。 無論哪種情況，溪流都會被摧毀。

此方法與監聽 'data' 事件不同，因為它利用 readable 底層機械中的事件，並可限制同時 fn 呼叫的數量。

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

參數

傳回

等直播結束後才會給他一個承諾。

一種用來從迭代器中建立可讀串流的工具方法。

static function from(iterable: Iterable<any> | AsyncIterable<any>, options?: ReadableOptions<Readable>): Readable

參數

傳回

一種從網路Readable中建立 的工具ReadableStream方法 。

static function fromWeb(readableStream: ReadableStream<any>, options?: Pick<ReadableOptions<Readable>, "encoding" | "highWaterMark" | "objectMode" | "signal">): Readable

參數

傳回

回傳串流是否被讀取或取消。

static function isDisturbed(stream: ReadableStream<any> | ReadableStream): boolean

參數

傳回

該 readable.isPaused() 方法回傳 的當前運作狀態 Readable。 這主要由方法 readable.pipe() 背後的機制所利用。 在大多數情況下，沒有理由直接使用這種方法。

const readable = new stream.Readable();

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

function isPaused(): boolean

傳回

此方法建立的迭代器讓使用者選擇在迴圈被 for await...of、 return、 或 break、 退出時取消銷毀throw串流，或若迭代器在迭代過程中產生錯誤，則刪除該串流。

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

此方法允許對串流進行映射。 fn 函式會對串流中的每個區塊被呼叫。 如果 fn 函式回傳承諾，該承諾會在傳遞給結果串流前先被 awaited 處理。

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

參數

傳回

一個以 函數 fn 映射的溪流。

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

此 readable.pause() 方法會使處於流動模式的溪流停止發出 'data' 事件，切換出流動模式。 任何可用的資料都會保留在內部緩衝區中。

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

如果存在readable.pause()事件監聽器，這個'readable'方法就沒有效果。

function pause(): Readable

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

此 readable.read() 方法會從內部緩衝區讀取資料並回傳資料。 若無可讀取資料， null 則會回傳。 預設情況下，資料會以物件形式回傳 Buffer ，除非已用該 readable.setEncoding() 方法指定編碼或串流以物件模式運作。

可選 size 參數指定要讀取的特定位元組數。 若 size 無法讀取位元組， null 則會回傳 ，除非 串流已結束，屆時內部緩衝區中剩餘的資料將全部回傳。

若未指定參數 size ，內部緩衝區內所有資料將被回傳。

size參數必須小於或等於 1 GiB。

readable.read()此方法僅應在Readable暫停模式下運作的串流中呼叫。 在流動模式下， readable.read() 會自動呼叫，直到內部緩衝器完全放空。

const readable = getReadableStreamSomehow();

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

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

每次呼叫 回 readable.read() 傳一塊資料，或 null。 區塊並非串接。 使用緩衝區中所有目前資料是必須迴 while 圈的必要條件。 讀取大型檔案 .read() 時，可能會回傳 null，已經用盡所有緩衝內容，但仍有更多資料尚未緩衝。 此時當緩衝區資料增加時，會發出新 'readable' 事件。 最後 'end' ，當沒有更多資料時，事件才會被發射。

因此，要從 讀取 readable檔案的全部內容，必須在多個 'readable' 事件中收集區塊：

const chunks = [];

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

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

物件模式中的Readable串流無論參數值readable.read(size)為何，都會從呼叫size中回傳單一項目。

若 readable.read() 方法回傳一塊資料，也會發出事件 'data' 。

事件發出後呼叫'end'將返回null。 不會產生執行時錯誤。

function read(size?: number): any

參數

傳回

此方法依序呼叫串流的每個區塊 fn ，並將前一個元素的計算結果傳遞給該區塊。 它回傳約約的最終值承諾。

若未提供 初始 值，則以串流的第一區塊作為初始值。 如果串流為空，承諾會以 TypeError a 拒絕，且帶有 代碼 ERR_INVALID_ARGS 屬性。

縮減函數逐元素迭代串流，表示不存在 並行 參數或平行性。 要同時執行減量，你可以提取方法的 readable.map 非同步函數。

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

參數

傳回

承諾最終還款的價值。

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

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

參數

傳回

此 readable.resume() 方法會使明確暫停 Readable 的串流重新 'data' 發出事件，切換到流模式。

此 readable.resume() 方法可用於完全消耗串流中的資料，而不必實際處理任何資料：

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

如果存在readable.resume()事件監聽器，這個'readable'方法就沒有效果。

function resume(): Readable

傳回

此 readable.setEncoding() 方法設定從串流讀取 Readable 資料的字元編碼。

預設情況下，不會指派編碼，串流資料會以物件形式回傳 Buffer 。 設定編碼會使串流資料以指定編碼的字串形式回傳，而非物件 Buffer 。 例如，呼叫 readable.setEncoding('utf8') 會使輸出資料被解讀為 UTF-8 資料，並以字串傳遞。 呼叫 readable.setEncoding('hex') 會使資料以十六進位字串格式編碼。

Readable串流會正確處理透過串流傳送的多位元組字元，否則若直接從串流中拉取物件Buffer，會被錯誤解碼。

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

function setEncoding(encoding: BufferEncoding): Readable

參數

傳回

此方法類似 Array.prototype.some 於 ，且在串流中每個區塊呼叫 fn ，直到等待的回傳值為（ true 或任何真值）。 一旦 fn 呼叫區塊 awaited 返回值為真值，串流即被銷毀，承諾由 實現。true 如果區塊上的 fn 呼叫都沒有回傳真值，則承諾在 中被實現。false

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

參數

傳回

一個評估 truefn 是否對至少一個區塊回傳真值的承諾。

此方法回傳一個包含首 批極限 區塊的新串流。

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

參數

傳回

一個有 限制 區塊的串流。

此方法能輕鬆取得串流內容。

由於此方法會將整個串流讀取到記憶體，因此抵銷了串流的優勢。 它的目的是為了互通性和便利性，而不是主要的串流消費方式。

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

參數

傳回

一個包含串流內容陣列的承諾。

一種從 ReadableStream建立網路Readable的工具方法 。

static function toWeb(streamReadable: ReadableStream, options?: ReadableToWebOptions): ReadableStream<any>

參數

傳回

此readable.unpipe()方法將先前用Writable法連接的水流分離。

若未指定， destination 則 所有 管線皆為分離。

如果 destination 被指定，但沒有為它設置管線，則該方法不會做任何事。

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

function unpipe(destination?: WritableStream): Readable

參數

傳回

通過 chunk 為 null 訊號會表示串流結束（EOF），行為與 相同 readable.push(null)，之後無法再寫入資料。 EOF 訊號會放在緩衝區末端，任何緩衝的資料仍會被清除。

此 readable.unshift() 方法會將一塊資料推回內部緩衝區。 這在某些情況下非常有用，例如串流被需要「恢復」部分從來源中樂觀拉取的資料，以便將資料傳遞給其他方。

事件發出後stream.unshift(chunk)無法呼叫該'end'方法，否則會拋出執行錯誤。

經常使用 stream.unshift() 串流的開發者應該考慮改 Transform 用串流。 更多資訊請參閱本 API for stream implementers 節。

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
import { StringDecoder } from 'node:string_decoder';
function parseHeader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onReadable);
  const decoder = new StringDecoder('utf8');
  let header = '';
  function onReadable() {
    let chunk;
    while (null !== (chunk = stream.read())) {
      const str = decoder.write(chunk);
      if (str.includes('\n\n')) {
        // Found the header boundary.
        const split = str.split(/\n\n/);
        header += split.shift();
        const remaining = split.join('\n\n');
        const buf = Buffer.from(remaining, 'utf8');
        stream.removeListener('error', callback);
        // Remove the 'readable' listener before unshifting.
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // Now the body of the message can be read from the stream.
        callback(null, header, stream);
        return;
      }
      // Still reading the header.
      header += str;
    }
  }
}

與 推送不同，不會 stream.unshift(chunk) 透過重置串流的內部讀取狀態來終止讀取過程。 若 readable.unshift() 在讀取過程中（即自訂串流的 _read 實作中呼叫），可能會產生意想不到的結果。 在呼叫 to readable.unshift() 後立即 推送 ，會適當地重置讀取狀態，但最好在執行讀取過程中避免呼叫 readable.unshift() 。

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

參數

在 Node.js 0.10 之前，串流並未依目前定義實作完整的 node:stream 模組 API。 （更多資訊請參見 Compatibility 。）

當使用較舊的 Node.js 函式庫，該函式庫會發出 'data' 事件 且暫停方法 僅為顧問時， readable.wrap() 該方法可用來建立 Readable 以舊串流作為資料來源的串流。

雖然很少需要使用 readable.wrap() ，但此方法是為了方便與舊 Node.js 應用程式和函式庫互動而提供。

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

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

function wrap(stream: ReadableStream): Readable

參數

傳回

function [asyncIterator](): AsyncIterator<any, undefined, any>

傳回

AsyncIterator 完全消費整個直播。

當該 --experimental-stream-iter 標誌啟用時， Readable 串流會實作該 Stream.toAsyncStreamable 協定，使 API 能高效使用 stream/iter 。

這提供了一個批次非同步迭代器，將串流的內部緩衝區分成 Uint8Array[] 批次，並攤銷標準 Symbol.asyncIterator 路徑的每個區塊 Promise 開銷。 對於位元組模式串流，區塊會直接 Buffer 以實例形式產生（屬於 Uint8Array 子類別）。 對於物件模式或編碼串流，每個區塊在批次處理前會被正規化為 Uint8Array 。

回傳的迭代器會被標記為已驗證的來源，因此 from() 無需額外正規化即可通過。

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

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

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

若無該 --experimental-stream-iter 標記，呼叫此方法會拋 ERR_STREAM_ITER_MISSING_FLAG出 。

function [toAsyncStreamable](): ByteReadableStream

傳回

傳回數位，列出發出器已註冊接聽程式的事件。

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

傳回

繼承自 Stream.eventNames

傳回 EventEmitter 所設定之 emitter.setMaxListeners(n) 目前的最大接聽程式值，或預設為 events.defaultMaxListeners。

function getMaxListeners(): number

傳回

繼承自 Stream.getMaxListeners

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

參數

傳回

繼承自 Stream.pipe

根據預設，如果特定事件新增了超過 EventEmitter 接聽程式，10會列印警告。 這是有助於尋找記憶體流失的實用預設值。 emitter.setMaxListeners() 方法允許修改這個特定 EventEmitter 實例的限制。 值可以設定為 Infinity （或 0），表示不限數目的接聽程式。

傳回 EventEmitter的參考，以便可以鏈結呼叫。

function setMaxListeners(n: number): Readable

參數

傳回

繼承自 Stream.setMaxListeners

Symbol.for('nodejs.rejection')此方法用於在發射事件時發生承諾拒絕，且captureRejections在發射端啟用時使用。 可以用 events.captureRejectionSymbol 來代替 Symbol.for('nodejs.rejection')。

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

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

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

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

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

參數

繼承自 Stream.__@captureRejectionSymbol@180