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の真値であるかを確認します。 チャンク返り値に対するawait呼び出しがフェイルシーになると、ストリームは破棄され、約束は達成falseされます。 チャンクに対するすべての fn 呼び出しが真値(truthy)を返す場合、その約束は trueで満たされます。

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

パラメーター

返品

fntrueがすべてのチャンクに対して真値(truthy)値を返すかどうかを評価する約束です。

この方法はストリームのフィルタリングを可能にします。 ストリーム内の各チャンクごとに fn 関数が呼び出され、もしtruey値を返すと、そのチャンクは結果ストリームに渡されます。 もしfn関数が約束を返すなら、その約束はawaitされます。

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>

パラメーター

返品

fnが真値で評価した最初のチャンクに対して、または要素が見つからなければundefinedに対して評価する約束。

この方法は、各チャンクにコールバックを適用し、その結果をフラット化することで新しいストリームを返します。

fnからストリームや他の反復可能または非同期反復可能を返すことも可能であり、その結果のストリームは返されたストリームに統合(フラット化)されます。

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

パラメーター

返品

関数 fnでフラットマップされたストリームです。

この方法はストリームの反復処理を可能にします。 ストリーム内の各チャンクごとに fn 関数が呼び出されます。 もしfn関数が約束を返すなら、その約束はawaitされます。

この方法は 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>

パラメーター

返品

配信が終わったらの約束だ。

イテレーターからReadable Streamsを作成するためのユーティリティ手法。

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関数が約束を返すと、その約束は結果ストリームに渡される前にawaitされます。

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コードの性質を持つ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')を呼び出すと、データは16進文字列形式でエンコードされます。

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に似ており、待ち望まれた返り値(または任意の真値)がされるまでストリーム内の各チャンクでtrueを呼び出します。 チャンク返却値に対するawait呼び出しが真理になると、ストリームは破棄され、約束はtrueで満たされます。 チャンクに対する fn 呼び出しが真実値を返さなければ、約束は falseで満たされます。

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

パラメーター

返品

fnが少なくとも1つのチャンクで真実true値(truthy)値を返すかどうかを評価する約束です。

このメソッドは最初の リミット チャンクを持つ新しいストリームを返します。

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 実装内から)呼び出されると予期せぬ結果が生じる可能性があります。 readable.unshift()呼び出しの後に即時プッシュを行うことで読み取り状態が適切にリセットされますが、読み取りの過程でreadable.unshift()を呼び出すのは避けるのが最善です。

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

パラメーター

Node.js 0.10以前は、ストリームズは現在の定義通りの全 node:stream モジュールAPIを実装していませんでした。 (詳細は Compatibility を参照してください。)

'data'イベントを発生し、アドバイザリーのみの一時停止メソッドを持つ古い Node.js ライブラリを使用する場合、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プロトコルを実装し、stream/iterAPIによる効率的な利用を可能にします。

これにより、ストリームの内部バッファを 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

既定では、EventEmitters は、特定のイベントに対して 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