Readable interface

パッケージ:: @azure/storage-common

Extends: Stream

プロパティ

closed	`true`が放出された後の`'close'`です。
destroyed	`true`が呼ばれた後に`readable.destroy()`ですか?
errored	ストリームがエラーで破壊された場合はエラーを返します。
readable	`true`、ストリームが破壊されていないか、または`'error'`に送信されていないかを「`'end'`」と呼んでも安全かどうかです。
readableAborted	ストリームが `'end'`を出す前に破壊されたかエラーかを返します。
readableDidRead	`'data'`が発信されたかどうかを返します。
readableEncoding	与えられた`encoding`流のプロパティ`Readable`ゲッター。 `encoding`プロパティはsetEncodingメソッドで設定できます。
readableEnded	イベントが`true`発生すると`'end'`になります。
readableFlowing	この性質は、`Readable`のセクションで説明されている流の現在の状態を反映しています。
readableHighWaterMark	この`highWaterMark`作成時に渡された`Readable`の値を返します。
readableLength	このプロパティは、キュー内で読み取れるバイト(またはオブジェクト)の数を含みます。この値は、 `highWaterMark`の状況に関する内省データを提供します。
readableObjectMode	与えられた`objectMode`流のプロパティ`Readable`ゲッター。

メソッド

addListener(string \| symbol, (args: any[]) => void)
addListener<E>(E, (args: ReadableEventMap[E]) => void)	`emitter.on(eventName, listener)` の別名。
compose(WritableStream \| WritableStream \| TransformStream \| (source: any) => void, Abortable)	`import { Readable } from 'node:stream'; async function* splitToWords(source) { for await (const chunk of source) { const words = String(chunk).split(' '); for (const word of words) { yield word; } } } const wordsStream = Readable.from(['text passed through', 'composed stream']).compose(splitToWords); const words = await wordsStream.toArray(); console.log(words); // prints ['text', 'passed', 'through', 'composed', 'stream']` `readable.compose(s)` は `stream.compose(readable, s)` と等価です。この方法では、 `AbortSignal` が提供され、中止時に合成されたストリームが破壊されます。詳細については、`stream.compose(...streams)` を参照してください。
destroy(Error)	ストリームを破棄します。オプションで `'error'` イベントを発信し、 `'close'` イベントを発信することもできます( `emitClose` が `false`に設定されていない場合)。この呼び出しの後、読み取り可能なストリームは内部リソースを解放し、その後の `push()` 呼び出しは無視されます。 `destroy()`が呼び出されると、それ以上の通話は no-op となり、`_destroy()`からの誤りは`'error'`として発信できません。実装者はこの手法を上書きせず、代わりに `readable._destroy()`を実装すべきです。
drop(number, Abortable)	このメソッドは、最初の制限チャンクを最初にドロップした新しいストリームを返します。
emit(string \| symbol, any[])
emit<E>(E, ReadableEventMap[E])	`eventName`という名前のイベントに登録された各リスナーを、登録された順序で同期的に呼び出し、指定された引数を各リスナーに渡します。イベントにリスナーがある場合は `true` を返します。それ以外の場合 `false` 返します。 import { EventEmitter } from 'node:events'; const myEmitter = new EventEmitter(); // First listener myEmitter.on('event', function firstListener() { console.log('Helloooo! first listener'); }); // Second listener myEmitter.on('event', function secondListener(arg1, arg2) { console.log(`event with parameters ${arg1}, ${arg2} in second listener`); }); // Third listener myEmitter.on('event', function thirdListener(...args) { const parameters = args.join(', '); console.log(`event with parameters ${parameters} in third listener`); }); console.log(myEmitter.listeners('event')); myEmitter.emit('event', 1, 2, 3, 4, 5); // Prints: // [ // [Function: firstListener], // [Function: secondListener], // [Function: thirdListener] // ] // Helloooo! first listener // event with parameters 1, 2 in second listener // event with parameters 1, 2, 3, 4, 5 in third listener
every((data: any, options?: Abortable) => boolean \| Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	この方法は `Array.prototype.every` に似ており、ストリーム内の各チャンクで fn を呼び出して、待ちたい返り値が fnの真値であるかを確認します。チャンク返り値に対する`await`呼び出しがフェイルシーになると、ストリームは破棄され、約束は達成`false`されます。チャンクに対するすべての fn 呼び出しが真値(truthy)を返す場合、その約束は `true`で満たされます。
filter((data: any, options?: Abortable) => boolean \| Promise<boolean>, ReadableOperatorOptions)	この方法はストリームのフィルタリングを可能にします。ストリーム内の各チャンクごとに fn 関数が呼び出され、もしtruey値を返すと、そのチャンクは結果ストリームに渡されます。もしfn関数が約束を返すなら、その約束は`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">)	この方法は `Array.prototype.find` に似ており、ストリーム内の各チャンクで fn を呼び出し、 fnの真値を持つチャンクを探します。 fn呼び出しの待ち望まれた返り値が真実値になると、ストリームは破棄され、約束はfnが真理値を返した値で満たされます。チャンクに対するすべての fn 呼び出しが偽の値を返す場合、その約束は `undefined`で満たされます。
flatMap((data: any, options?: Abortable) => any, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	この方法は、各チャンクにコールバックを適用し、その結果をフラット化することで新しいストリームを返します。 fnからストリームや他の反復可能または非同期反復可能を返すことも可能であり、その結果のストリームは返されたストリームに統合(フラット化)されます。
forEach((data: any, options?: Abortable) => void \| Promise<void>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	この方法はストリームの反復処理を可能にします。ストリーム内の各チャンクごとに fn 関数が呼び出されます。もしfn関数が約束を返すなら、その約束は`await`されます。この方法は `for await...of` ループとは異なり、選択的にチャンクを同時に処理できます。さらに、 `forEach` の反復は `signal` オプションを通過し、関連するAbortControllerを中止することでのみ停止できますが、 `for await...of` は `break` や `return`で停止できます。いずれにせよ、その川は破壊されます。この方法は、 `'data'` イベントを聞くのとは異なり、基盤となる機械で `readable` イベントを利用し、同時実行の fn コール数を制限できます。
isPaused()	`readable.isPaused()`メソッドは`Readable`の現在の動作状態を返します。これは主に `readable.pipe()` 法の基盤となるメカニズムによって利用されます。ほとんどの場合、この方法を直接使う理由はありません。 `const readable = new stream.Readable(); readable.isPaused(); // === false readable.pause(); readable.isPaused(); // === true readable.resume(); readable.isPaused(); // === false`
iterator(ReadableIteratorOptions)	この方法で作成されたイテレーターは、 `for await...of` ループが `return`、 `break`、または `throw`で終了した場合にストリームの破壊をキャンセルするか、反復中にストリームがエラーを出した場合に破壊するオプションをユーザーに与えます。
listenerCount(string \| symbol, (args: any[]) => void)
listenerCount<E>(E, (args: ReadableEventMap[E]) => void)	`eventName`という名前のイベントをリッスンしているリスナーの数を返します。 `listener` 指定すると、イベントのリスナーの一覧でリスナーが見つかった回数が返されます。
listeners(string \| symbol)
listeners<E>(E)	`eventName`という名前のイベントのリスナーの配列のコピーを返します。 `server.on('connection', (stream) => { console.log('someone connected!'); }); console.log(util.inspect(server.listeners('connection'))); // Prints: [ [Function] ]`
map((data: any, options?: Abortable) => any, ReadableOperatorOptions)	この方法により、ストリーム上のマッピングが可能になります。 fn関数はストリーム内のすべてのチャンクごとに呼び出されます。 fn関数が約束を返すと、その約束は結果ストリームに渡される前に`await`されます。
off(string \| symbol, (args: any[]) => void)
off<E>(E, (args: ReadableEventMap[E]) => void)	`emitter.removeListener()` の別名。
on(string \| symbol, (args: any[]) => void)
on<E>(E, (args: ReadableEventMap[E]) => void)	`listener`という名前のイベントのリスナー配列の末尾に `eventName` 関数を追加します。 `listener` が既に追加されているかどうかを確認するチェックは行われません。 `eventName` と `listener` の同じ組み合わせを渡す複数の呼び出しでは、`listener` が複数回追加され、呼び出されます。 `server.on('connection', (stream) => { console.log('someone connected!'); });` 呼び出しをチェーンできるように、`EventEmitter`への参照を返します。既定では、イベントリスナーは追加された順序で呼び出されます。 `emitter.prependListener()` メソッドは、リスナー配列の先頭にイベントリスナーを追加する代わりに使用できます。 `import { EventEmitter } from 'node:events'; const myEE = new EventEmitter(); myEE.on('foo', () => console.log('a')); myEE.prependListener('foo', () => console.log('b')); myEE.emit('foo'); // Prints: // b // a`
once(string \| symbol, (args: any[]) => void)
once<E>(E, (args: ReadableEventMap[E]) => void)	という名前のイベント `listenereventName` 関数を追加します。次回 `eventName` がトリガーされると、このリスナーが削除され、呼び出されます。 `server.once('connection', (stream) => { console.log('Ah, we have our first user!'); });` 呼び出しをチェーンできるように、`EventEmitter`への参照を返します。既定では、イベントリスナーは追加された順序で呼び出されます。 `emitter.prependOnceListener()` メソッドは、リスナー配列の先頭にイベントリスナーを追加する代わりに使用できます。 `import { EventEmitter } from 'node:events'; const myEE = new EventEmitter(); myEE.once('foo', () => console.log('a')); myEE.prependOnceListener('foo', () => console.log('b')); myEE.emit('foo'); // Prints: // b // a`
pause()	`readable.pause()`方式では、フローモードのストリームが`'data'`イベントの発生を停止し、フローモードから切り替わります。利用可能になったデータは内部バッファに残ります。 const readable = getReadableStreamSomehow(); readable.on('data', (chunk) => { console.log(`Received ${chunk.length} bytes of data.`); readable.pause(); console.log('There will be no additional data for 1 second.'); setTimeout(() => { console.log('Now data will start flowing again.'); readable.resume(); }, 1000); }); `readable.pause()`メソッドは、`'readable'`イベントリスナーがいる場合は影響しません。
prependListener(string \| symbol, (args: any[]) => void)
prependListener<E>(E, (args: ReadableEventMap[E]) => void)	関数を、という名前のイベントのリスナー配列の先頭に追加します。 `listener` が既に追加されているかどうかを確認するチェックは行われません。 `eventName` と `listener` の同じ組み合わせを渡す複数の呼び出しでは、`listener` が複数回追加され、呼び出されます。 `server.prependListener('connection', (stream) => { console.log('someone connected!'); });` 呼び出しをチェーンできるように、`EventEmitter`への参照を返します。
prependOnceListener(string \| symbol, (args: any[]) => void)
prependOnceListener<E>(E, (args: ReadableEventMap[E]) => void)	という名前のイベントの 1 回限りの関数をリスナー配列の先頭に追加します。次に `eventName` がトリガーされると、このリスナーが削除され、呼び出されます。 `server.prependOnceListener('connection', (stream) => { console.log('Ah, we have our first user!'); });` 呼び出しをチェーンできるように、`EventEmitter`への参照を返します。
push(any, BufferEncoding)
rawListeners(string \| symbol)
rawListeners<E>(E)	ラッパー (`eventName`によって作成されたものなど) を含む、`.once()`という名前のイベントのリスナーの配列のコピーを返します。 import { EventEmitter } from 'node:events'; const emitter = new EventEmitter(); emitter.once('log', () => console.log('log once')); // Returns a new Array with a function `onceWrapper` which has a property // `listener` which contains the original listener bound above const listeners = emitter.rawListeners('log'); const logFnWrapper = listeners[0]; // Logs "log once" to the console and does not unbind the `once` event logFnWrapper.listener(); // Logs "log once" to the console and removes the listener logFnWrapper(); emitter.on('log', () => console.log('log persistently')); // Will return a new Array with a single function bound by `.on()` above const newListeners = emitter.rawListeners('log'); // Logs "log persistently" twice newListeners[0](); emitter.emit('log');
read(number)	`readable.read()`メソッドは内部バッファからデータを読み出し、返します。読み取れるデータがない場合は `null` が返されます。デフォルトでは、`Buffer`メソッドでエンコーディングが指定されていないか、ストリームがオブジェクトモードで動作している場合を除き、データは`readable.setEncoding()`オブジェクトとして返されます。オプションの `size` 引数は、読み取るべき特定のバイト数を指定します。 `size`バイトが読み取れない場合は、ストリームが終了`null`が返されます。終了した場合、内部バッファに残っているすべてのデータが返されます。 `size`引数が指定されていない場合、内部バッファに含まれるすべてのデータが返されます。 `size`の引数は1 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`が戻ります。ランタイムエラーは発生しません。
reduce<T>((previous: any, data: any, options?: Abortable) => T)	このメソッドはストリームの各チャンクで順番に fn を呼び出し、前の要素の計算結果をそのチャンクに渡します。リ約の最終値に対する約束を返します。初期値が提供されない場合は、ストリームの最初のチャンクが初期値として使用されます。ストリームが空の場合、`TypeError`コードの性質を持つ`ERR_INVALID_ARGS`で約束が拒否されます。リデューサー関数はストリームを要素ごとに反復するため、並行パラメータや並列処理は存在しません。リダクションを同時に行うには、非同期関数を `readable.map` メソッドに抽出できます。
reduce<T>((previous: T, data: any, options?: Abortable) => T, T, Abortable)
removeAllListeners(string \| symbol)
removeAllListeners<E>(E)	すべてのリスナー、または指定した `eventName`のリスナーを削除します。特に、`EventEmitter` インスタンスが他のコンポーネントまたはモジュール (ソケットやファイルストリームなど) によって作成された場合に、コード内の他の場所に追加されたリスナーを削除するのは不適切な方法です。呼び出しをチェーンできるように、`EventEmitter`への参照を返します。
removeListener(string \| symbol, (args: any[]) => void)
removeListener<E>(E, (args: ReadableEventMap[E]) => void)	`listener`という名前のイベントのリスナー配列から、指定した `eventName` を削除します。 `const callback = (stream) => { console.log('someone connected!'); }; server.on('connection', callback); // ... server.removeListener('connection', callback);` `removeListener()` は、リスナー配列からリスナーのインスタンスを最大で 1 つ削除します。指定した `eventName`のリスナー配列に 1 つのリスナーが複数回追加されている場合は、各インスタンスを削除するために `removeListener()` を複数回呼び出す必要があります。イベントが生成されると、出力時にそれにアタッチされているすべてのリスナーが順番に呼び出されます。これは、発信`removeListener()`かつ最後のリスナーが実行を終える`removeAllListeners()`またはコールが行っても、進行中の`emit()`から除外されないことを意味します。後続のイベントは想定どおりに動作します。 import { EventEmitter } from 'node:events'; class MyEmitter extends EventEmitter {} const myEmitter = new MyEmitter(); const callbackA = () => { console.log('A'); myEmitter.removeListener('event', callbackB); }; const callbackB = () => { console.log('B'); }; myEmitter.on('event', callbackA); myEmitter.on('event', callbackB); // callbackA removes listener callbackB but it will still be called. // Internal listener array at time of emit [callbackA, callbackB] myEmitter.emit('event'); // Prints: // A // B // callbackB is now removed. // Internal listener array [callbackA] myEmitter.emit('event'); // Prints: // A リスナーは内部配列で管理されるため、これを呼び出すと、リスナーが削除された後に登録されたリスナーのポジションインデックスが変更されます。これはリスナーが呼び出される順序には影響しませんが、`emitter.listeners()` メソッドによって返されるリスナー配列のコピーを再作成する必要があることを意味します。 (次の例のように) 1 つのイベントに対して 1 つの関数がハンドラーとして複数回追加された場合、`removeListener()` は最後に追加されたインスタンスを削除します。この例では、`once('ping')` リスナーは削除されます。 `import { EventEmitter } from 'node:events'; const ee = new EventEmitter(); function pong() { console.log('pong'); } ee.on('ping', pong); ee.once('ping', pong); ee.removeListener('ping', pong); ee.emit('ping'); ee.emit('ping');` 呼び出しをチェーンできるように、`EventEmitter`への参照を返します。
resume()	`readable.resume()`方式は、明示的に一時停止された`Readable`ストリームを`'data'`イベントの発生を再開させ、ストリームをフローングモードに切り替わります。 `readable.resume()`メソッドは、実際にデータを処理せずにストリームからのデータを完全に消費するために使用できます: `getReadableStreamSomehow() .resume() .on('end', () => { console.log('Reached the end, but did not read anything.'); });` `readable.resume()`メソッドは、`'readable'`イベントリスナーがいる場合は影響しません。
setEncoding(BufferEncoding)	`readable.setEncoding()`メソッドは、`Readable`ストリームから読み取るデータの文字エンコーディングを設定します。デフォルトではエンコーディングは割り当てられず、ストリームデータは `Buffer` オブジェクトとして返されます。エンコーディングを設定すると、ストリームデータは指定されたエンコーディングの文字列として返され、 `Buffer` オブジェクトとしては戻されません。例えば、 `readable.setEncoding('utf8')` を呼び出すと、出力データがUTF-8データとして解釈され、文字列として渡されます。 `readable.setEncoding('hex')`を呼び出すと、データは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); });`
some((data: any, options?: Abortable) => boolean \| Promise<boolean>, Pick<ReadableOperatorOptions, "concurrency" \| "signal">)	この方法は`Array.prototype.some`に似ており、待ち望まれた返り値(または任意の真値)がされるまでストリーム内の各チャンクで`true`を呼び出します。チャンク返却値に対する`await`呼び出しが真理になると、ストリームは破棄され、約束は`true`で満たされます。チャンクに対する fn 呼び出しが真実値を返さなければ、約束は `false`で満たされます。
take(number, Abortable)	このメソッドは最初のリミットチャンクを持つ新しいストリームを返します。
toArray(Abortable)	この方法により、ストリームの内容を簡単に取得できます。この方法はストリーム全体をメモリに読み込むため、ストリームの利点を相殺します。これは相互運用性と利便性を重視しており、ストリームを消費する主な方法ではありません。
unpipe(WritableStream)	`readable.unpipe()`法は`Writable`パイプ法で接続されていた流を切り離します。 `destination`が指定されていなければ、すべてのパイプは分離されます。 `destination`が指定されていてもパイプが設定されていなければ、メソッドは何もしません。 `import fs from 'node:fs'; const readable = getReadableStreamSomehow(); const writable = fs.createWriteStream('file.txt'); // All the data from readable goes into 'file.txt', // but only for the first second. readable.pipe(writable); setTimeout(() => { console.log('Stop writing to file.txt.'); readable.unpipe(writable); console.log('Manually close the file stream.'); writable.end(); }, 1000);`
unshift(any, BufferEncoding)	`chunk`を`null`として渡すことはストリームの終了(EOF)を知らせ、`readable.push(null)`と同じ動作をします。その後はデータが書き込めなくなります。 EOF信号はバッファの端に配置され、バッファされたデータは依然としてフラッシュされます。 `readable.unshift()`メソッドはデータのチャンクを内部バッファに戻します。これは、ストリームがソースから楽観的に取り出したデータを「アンコンソング」する必要があるコードによって消費されている特定の状況で有用です。これにより、そのデータを他の第三者に渡すことができます。 `stream.unshift(chunk)`イベントが発生した後に`'end'`メソッドを呼び出すことはできず、実行時エラーが発生します。 `stream.unshift()`をよく使う開発者は、代わりに`Transform`ストリームの使用に切り替えることを検討すべきです。詳細は `API for stream implementers` セクションをご覧ください。 // Pull off a header delimited by \n\n. // Use unshift() if we get too much. // Call the callback with (error, header, stream). import { StringDecoder } from 'node:string_decoder'; function parseHeader(stream, callback) { stream.on('error', callback); stream.on('readable', onReadable); const decoder = new StringDecoder('utf8'); let header = ''; function onReadable() { let chunk; while (null !== (chunk = stream.read())) { const str = decoder.write(chunk); if (str.includes('\n\n')) { // Found the header boundary. const split = str.split(/\n\n/); header += split.shift(); const remaining = split.join('\n\n'); const buf = Buffer.from(remaining, 'utf8'); stream.removeListener('error', callback); // Remove the 'readable' listener before unshifting. stream.removeListener('readable', onReadable); if (buf.length) stream.unshift(buf); // Now the body of the message can be read from the stream. callback(null, header, stream); return; } // Still reading the header. header += str; } } } プッシュとは異なり、`stream.unshift(chunk)`はストリームの内部読み取り状態をリセットして読み取りプロセスを終了しません。これにより、 `readable.unshift()` が読み取り中(すなわちカスタムストリーム上の _read 実装内から)呼び出されると予期せぬ結果が生じる可能性があります。 `readable.unshift()`呼び出しの後に即時プッシュを行うことで読み取り状態が適切にリセットされますが、読み取りの過程で`readable.unshift()`を呼び出すのは避けるのが最善です。
wrap(ReadableStream)	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. });`
[asyncDispose]()	`AbortError`で`readable.destroy()`に電話をかけ、配信終了時にその約束を返します。
[asyncIterator]()
[toAsyncStreamable]()	`--experimental-stream-iter`フラグが有効になると、`Readable`ストリームは`Stream.toAsyncStreamable`プロトコルを実装し、`stream/iter`APIによる効率的な利用を可能にします。これにより、ストリームの内部バッファを `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`が投げ出されます。

継承されたメソッド

eventNames()	エミッターがリスナーを登録したイベントを一覧表示する配列を返します。 `import { EventEmitter } from 'node:events'; const myEE = new EventEmitter(); myEE.on('foo', () => {}); myEE.on('bar', () => {}); const sym = Symbol('symbol'); myEE.on(sym, () => {}); console.log(myEE.eventNames()); // Prints: [ 'foo', 'bar', Symbol(symbol) ]`
getMaxListeners()	`EventEmitter` の現在の最大リスナー値を返します。この値は、`emitter.setMaxListeners(n)` によって設定されるか、既定で `events.defaultMaxListeners`に設定されます。
pipe<T>(T, PipeOptions)
setMaxListeners(number)	既定では、`EventEmitter`s は、特定のイベントに対して `10` を超えるリスナーが追加された場合に警告を出力します。これは、メモリリークの検出に役立つ便利な既定値です。 `emitter.setMaxListeners()` メソッドを使用すると、この特定の `EventEmitter` インスタンスの制限を変更できます。この値を `Infinity` (または `0`) に設定すると、リスナーの数に制限はありません。呼び出しをチェーンできるように、`EventEmitter`への参照を返します。
[captureRejectionSymbol](Error, string \| symbol, any[])	`Symbol.for('nodejs.rejection')`メソッドは、イベントを発生させる際に約束拒否が発生し、エミッターで`captureRejections`が有効化された場合に呼び出されます。 `events.captureRejectionSymbol`の代わりに `Symbol.for('nodejs.rejection')` を使うことも可能です。 `import { EventEmitter, captureRejectionSymbol } from 'node:events'; class MyClass extends EventEmitter { constructor() { super({ captureRejections: true }); } [captureRejectionSymbol](err, event, ...args) { console.log('rejection happened for', event, 'with', err, ...args); this.destroy(err); } destroy(err) { // Tear the resource down here. } }`

プロパティの詳細

closed

trueが放出された後の'close'です。

closed: boolean

プロパティ値

boolean

destroyed

trueが呼ばれた後にreadable.destroy()ですか?

destroyed: boolean

プロパティ値

boolean

errored

ストリームがエラーで破壊された場合はエラーを返します。

errored: Error | null

プロパティ値

Error | null

readable

true、ストリームが破壊されていないか、または'error'に送信されていないかを「'end'」と呼んでも安全かどうかです。

readable: boolean

プロパティ値

boolean

readableAborted

ストリームが 'end'を出す前に破壊されたかエラーかを返します。

readableAborted: boolean

プロパティ値

boolean

readableDidRead

'data'が発信されたかどうかを返します。

readableDidRead: boolean

プロパティ値

boolean

readableEncoding

与えられたencoding流のプロパティReadableゲッター。 encodingプロパティはsetEncodingメソッドで設定できます。

readableEncoding: BufferEncoding | null

プロパティ値

BufferEncoding | null

readableEnded

イベントがtrue発生すると'end'になります。

readableEnded: boolean

プロパティ値

boolean

readableFlowing

この性質は、Readableのセクションで説明されている流の現在の状態を反映しています。

readableFlowing: boolean | null

プロパティ値

boolean | null

readableHighWaterMark

このhighWaterMark作成時に渡されたReadableの値を返します。

readableHighWaterMark: number

プロパティ値

number

readableLength

このプロパティは、キュー内で読み取れるバイト(またはオブジェクト)の数を含みます。この値は、 highWaterMarkの状況に関する内省データを提供します。

readableLength: number

プロパティ値

number

readableObjectMode

与えられたobjectMode流のプロパティReadableゲッター。

readableObjectMode: boolean

プロパティ値

boolean

メソッドの詳細

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

emitter.on(eventName, listener) の別名。

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

パラメーター

eventName: E

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

返品

this

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

import { Readable } from 'node:stream';

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

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

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

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

readable.compose(s) は stream.compose(readable, s) と等価です。

この方法では、 AbortSignal が提供され、中止時に合成されたストリームが破壊されます。

詳細については、stream.compose(...streams) を参照してください。

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

パラメーター

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

options: Abortable

返品

Duplex

この流れは streamと構成されています。

destroy(Error)

ストリームを破棄します。オプションで 'error' イベントを発信し、 'close' イベントを発信することもできます( emitClose が falseに設定されていない場合)。この呼び出しの後、読み取り可能なストリームは内部リソースを解放し、その後の push() 呼び出しは無視されます。

destroy()が呼び出されると、それ以上の通話は no-op となり、_destroy()からの誤りは'error'として発信できません。

実装者はこの手法を上書きせず、代わりに readable._destroy()を実装すべきです。

function destroy(error?: Error): this

パラメーター

error: Error

このエラーは 'error' 事象でペイロードとして渡されます

返品

this

drop(number, Abortable)

このメソッドは、最初の制限チャンクを最初にドロップした新しいストリームを返します。

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

パラメーター

limit: number

読み取り可能なチャンクの数を落とす。

options: Abortable

返品

Readable

最初から制限チャンクがドロップされたストリーム。

emit(string | symbol, any[])

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

パラメーター

eventName: string | symbol

args: any[]

返品

boolean

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

eventNameという名前のイベントに登録された各リスナーを、登録された順序で同期的に呼び出し、指定された引数を各リスナーに渡します。

イベントにリスナーがある場合は true を返します。それ以外の場合 false 返します。

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

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

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

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

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

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

パラメーター

eventName: E

args: ReadableEventMap[E]

返品

boolean

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

この方法は Array.prototype.every に似ており、ストリーム内の各チャンクで fn を呼び出して、待ちたい返り値が fnの真値であるかを確認します。チャンク返り値に対するawait呼び出しがフェイルシーになると、ストリームは破棄され、約束は達成falseされます。チャンクに対するすべての fn 呼び出しが真値(truthy)を返す場合、その約束は trueで満たされます。

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

パラメーター

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

ストリームの各チャンクで呼び出す関数です。非同期かどうかは別として。

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

返品

Promise<boolean>

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

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

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

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

パラメーター

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

ストリームからチャンクをフィルタリングする関数です。非同期かどうかは別として。

options: ReadableOperatorOptions

返品

Readable

述語 fn でフィルタリングされたストリームです。

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

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

パラメーター

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

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

返品

Promise<any>

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

この方法は Array.prototype.find に似ており、ストリーム内の各チャンクで fn を呼び出し、 fnの真値を持つチャンクを探します。 fn呼び出しの待ち望まれた返り値が真実値になると、ストリームは破棄され、約束はfnが真理値を返した値で満たされます。チャンクに対するすべての fn 呼び出しが偽の値を返す場合、その約束は undefinedで満たされます。

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

パラメーター

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

ストリームの各チャンクで呼び出す関数です。非同期かどうかは別として。

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

返品

Promise<T | undefined>

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

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

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

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

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

パラメーター

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

ストリーム内のすべてのチャンクをマッピングする関数です。同期が悪いかもしれません。小川や発電機かもしれません。

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

返品

Readable

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

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

この方法はストリームの反復処理を可能にします。ストリーム内の各チャンクごとに fn 関数が呼び出されます。 もしfn関数が約束を返すなら、その約束は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, "concurrency" | "signal">): Promise<void>

パラメーター

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

ストリームの各チャンクで呼び出す関数です。非同期かどうかは別として。

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

返品

Promise<void>

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

isPaused()

readable.isPaused()メソッドはReadableの現在の動作状態を返します。これは主に readable.pipe() 法の基盤となるメカニズムによって利用されます。ほとんどの場合、この方法を直接使う理由はありません。

const readable = new stream.Readable();

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

function isPaused(): boolean

返品

boolean

iterator(ReadableIteratorOptions)

この方法で作成されたイテレーターは、 for await...of ループが return、 break、または throwで終了した場合にストリームの破壊をキャンセルするか、反復中にストリームがエラーを出した場合に破壊するオプションをユーザーに与えます。

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

パラメーター

options: ReadableIteratorOptions

返品

AsyncIterator<any>

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

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

パラメーター

eventName: string | symbol

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

返品

number

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

eventNameという名前のイベントをリッスンしているリスナーの数を返します。 listener 指定すると、イベントのリスナーの一覧でリスナーが見つかった回数が返されます。

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

パラメーター

eventName: E

リッスンするイベントの名前

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

イベントハンドラー関数

返品

number

listeners(string | symbol)

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

パラメーター

eventName: string | symbol

返品

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

listeners<E>(E)

eventNameという名前のイベントのリスナーの配列のコピーを返します。

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

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

パラメーター

eventName: E

返品

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

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

この方法により、ストリーム上のマッピングが可能になります。 fn関数はストリーム内のすべてのチャンクごとに呼び出されます。 fn関数が約束を返すと、その約束は結果ストリームに渡される前にawaitされます。

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

パラメーター

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

ストリーム内のすべてのチャンクをマッピングする関数です。非同期かどうかは別として。

options: ReadableOperatorOptions

返品

Readable

関数 fnで写像されたストリーム。

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

emitter.removeListener() の別名。

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

パラメーター

eventName: E

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

返品

this

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

listenerという名前のイベントのリスナー配列の末尾に eventName 関数を追加します。 listener が既に追加されているかどうかを確認するチェックは行われません。 eventName と listener の同じ組み合わせを渡す複数の呼び出しでは、listener が複数回追加され、呼び出されます。

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

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

既定では、イベントリスナーは追加された順序で呼び出されます。 emitter.prependListener() メソッドは、リスナー配列の先頭にイベントリスナーを追加する代わりに使用できます。

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

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

パラメーター

eventName: E

イベントの名前。

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

コールバック関数

返品

this

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

という名前のイベント listenereventName 関数を追加します。次回 eventName がトリガーされると、このリスナーが削除され、呼び出されます。

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

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

既定では、イベントリスナーは追加された順序で呼び出されます。 emitter.prependOnceListener() メソッドは、リスナー配列の先頭にイベントリスナーを追加する代わりに使用できます。

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

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

パラメーター

eventName: E

イベントの名前。

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

コールバック関数

返品

this

pause()

readable.pause()方式では、フローモードのストリームが'data'イベントの発生を停止し、フローモードから切り替わります。利用可能になったデータは内部バッファに残ります。

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

readable.pause()メソッドは、'readable'イベントリスナーがいる場合は影響しません。

function pause(): this

返品

this

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

関数を、という名前のイベントのリスナー配列の先頭に追加します。 listener が既に追加されているかどうかを確認するチェックは行われません。 eventName と listener の同じ組み合わせを渡す複数の呼び出しでは、listener が複数回追加され、呼び出されます。

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

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

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

パラメーター

eventName: E

イベントの名前。

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

コールバック関数

返品

this

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

という名前のイベントの 1 回限りの関数をリスナー配列の先頭に追加します。次に eventName がトリガーされると、このリスナーが削除され、呼び出されます。

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

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

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

パラメーター

eventName: E

イベントの名前。

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

コールバック関数

返品

this

push(any, BufferEncoding)

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

パラメーター

chunk: any

encoding: BufferEncoding

返品

boolean

rawListeners(string | symbol)

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

パラメーター

eventName: string | symbol

返品

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

rawListeners<E>(E)

ラッパー (eventNameによって作成されたものなど) を含む、.once()という名前のイベントのリスナーの配列のコピーを返します。

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

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

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

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

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

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

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

パラメーター

eventName: E

返品

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

read(number)

readable.read()メソッドは内部バッファからデータを読み出し、返します。読み取れるデータがない場合は null が返されます。デフォルトでは、Bufferメソッドでエンコーディングが指定されていないか、ストリームがオブジェクトモードで動作している場合を除き、データはreadable.setEncoding()オブジェクトとして返されます。

オプションの size 引数は、読み取るべき特定のバイト数を指定します。 sizeバイトが読み取れない場合は、ストリームが終了nullが返されます。終了した場合、内部バッファに残っているすべてのデータが返されます。

size引数が指定されていない場合、内部バッファに含まれるすべてのデータが返されます。

sizeの引数は1 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

パラメーター

size: number

読み取るべきデータ量を指定するオプションの引数があります。

返品

any

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

このメソッドはストリームの各チャンクで順番に fn を呼び出し、前の要素の計算結果をそのチャンクに渡します。リ約の最終値に対する約束を返します。

初期値が提供されない場合は、ストリームの最初のチャンクが初期値として使用されます。ストリームが空の場合、TypeErrorコードの性質を持つERR_INVALID_ARGSで約束が拒否されます。

リデューサー関数はストリームを要素ごとに反復するため、並行パラメータや並列処理は存在しません。リダクションを同時に行うには、非同期関数を readable.map メソッドに抽出できます。

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

パラメーター

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

ストリーム内のすべてのチャンクを呼び出しるリデューサー関数です。非同期かどうかは別として。

返品

Promise<T>

削減の最終的な価値の約束です。

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

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

パラメーター

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

initial: T

options: Abortable

返品

Promise<T>

removeAllListeners(string | symbol)

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

パラメーター

eventName: string | symbol

返品

this

removeAllListeners<E>(E)

すべてのリスナー、または指定した eventNameのリスナーを削除します。

特に、EventEmitter インスタンスが他のコンポーネントまたはモジュール (ソケットやファイルストリームなど) によって作成された場合に、コード内の他の場所に追加されたリスナーを削除するのは不適切な方法です。

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

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

パラメーター

eventName: E

返品

this

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

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

パラメーター

eventName: string | symbol

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

返品

this

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

listenerという名前のイベントのリスナー配列から、指定した eventName を削除します。

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

removeListener() は、リスナー配列からリスナーのインスタンスを最大で 1 つ削除します。指定した eventNameのリスナー配列に 1 つのリスナーが複数回追加されている場合は、各インスタンスを削除するために removeListener() を複数回呼び出す必要があります。

イベントが生成されると、出力時にそれにアタッチされているすべてのリスナーが順番に呼び出されます。これは、発信removeListener()かつ最後のリスナーが実行を終えるremoveAllListeners()またはコールが行っても、進行中のemit()から除外されないことを意味します。後続のイベントは想定どおりに動作します。

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

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

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

myEmitter.on('event', callbackA);

myEmitter.on('event', callbackB);

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

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

リスナーは内部配列で管理されるため、これを呼び出すと、リスナーが削除 された後に 登録されたリスナーのポジションインデックスが変更されます。これはリスナーが呼び出される順序には影響しませんが、emitter.listeners() メソッドによって返されるリスナー配列のコピーを再作成する必要があることを意味します。

(次の例のように) 1 つのイベントに対して 1 つの関数がハンドラーとして複数回追加された場合、removeListener() は最後に追加されたインスタンスを削除します。この例では、once('ping') リスナーは削除されます。

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

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

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

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

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

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

パラメーター

eventName: E

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

返品

this

resume()

readable.resume()方式は、明示的に一時停止されたReadableストリームを'data'イベントの発生を再開させ、ストリームをフローングモードに切り替わります。

readable.resume()メソッドは、実際にデータを処理せずにストリームからのデータを完全に消費するために使用できます:

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

readable.resume()メソッドは、'readable'イベントリスナーがいる場合は影響しません。

function resume(): this

返品

this

setEncoding(BufferEncoding)

readable.setEncoding()メソッドは、Readableストリームから読み取るデータの文字エンコーディングを設定します。

デフォルトではエンコーディングは割り当てられず、ストリームデータは Buffer オブジェクトとして返されます。エンコーディングを設定すると、ストリームデータは指定されたエンコーディングの文字列として返され、 Buffer オブジェクトとしては戻されません。例えば、 readable.setEncoding('utf8') を呼び出すと、出力データがUTF-8データとして解釈され、文字列として渡されます。 readable.setEncoding('hex')を呼び出すと、データは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): this

パラメーター

encoding: BufferEncoding

使用するエンコード。

返品

this

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

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

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

パラメーター

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

ストリームの各チャンクで呼び出す関数です。非同期かどうかは別として。

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

返品

Promise<boolean>

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

take(number, Abortable)

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

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

パラメーター

limit: number

読み取れるものから取り出すチャンクの数。

options: Abortable

返品

Readable

制限チャンクが取られたストリーム。

toArray(Abortable)

この方法により、ストリームの内容を簡単に取得できます。

この方法はストリーム全体をメモリに読み込むため、ストリームの利点を相殺します。これは相互運用性と利便性を重視しており、ストリームを消費する主な方法ではありません。

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

パラメーター

options: Abortable

返品

Promise<any[]>

ストリームの内容を含む配列を含む約束です。

unpipe(WritableStream)

readable.unpipe()法はWritableパイプ法で接続されていた流を切り離します。

destinationが指定されていなければ、すべてのパイプは分離されます。

destinationが指定されていてもパイプが設定されていなければ、メソッドは何もしません。

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

function unpipe(destination?: WritableStream): this

パラメーター

destination: WritableStream

オプションで特定のストリームをアンパイプする

返品

this

unshift(any, BufferEncoding)

chunkをnullとして渡すことはストリームの終了(EOF)を知らせ、readable.push(null)と同じ動作をします。その後はデータが書き込めなくなります。 EOF信号はバッファの端に配置され、バッファされたデータは依然としてフラッシュされます。

readable.unshift()メソッドはデータのチャンクを内部バッファに戻します。これは、ストリームがソースから楽観的に取り出したデータを「アンコンソング」する必要があるコードによって消費されている特定の状況で有用です。これにより、そのデータを他の第三者に渡すことができます。

stream.unshift(chunk)イベントが発生した後に'end'メソッドを呼び出すことはできず、実行時エラーが発生します。

stream.unshift()をよく使う開発者は、代わりにTransformストリームの使用に切り替えることを検討すべきです。詳細は API for stream implementers セクションをご覧ください。

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

プッシュとは異なり、stream.unshift(chunk)はストリームの内部読み取り状態をリセットして読み取りプロセスを終了しません。これにより、 readable.unshift() が読み取り中(すなわちカスタムストリーム上の _read 実装内から)呼び出されると予期せぬ結果が生じる可能性があります。 readable.unshift()呼び出しの後に即時プッシュを行うことで読み取り状態が適切にリセットされますが、読み取りの過程でreadable.unshift()を呼び出すのは避けるのが最善です。

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

パラメーター

chunk: any

データの塊を読み出しキューにアンシフトする。オブジェクトモードで動作しないストリームの場合、 chunk {string}、{Buffer}、{TypedArray}、{DataView}、または null でなければなりません。オブジェクトモードストリームの場合、 chunk は任意のJavaScript値であっても構いません。

encoding: BufferEncoding

文字列チャンクのエンコーディング。有効な Buffer エンコーディングでなければなりません。例えば 'utf8' や 'ascii'などです。

wrap(ReadableStream)

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

'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): this

パラメーター

stream: ReadableStream

「旧様式」の読みやすいストリーム

返品

this

[asyncDispose]()

AbortErrorでreadable.destroy()に電話をかけ、配信終了時にその約束を返します。

function [asyncDispose](): Promise<void>

返品

Promise<void>

[asyncIterator]()

function [asyncIterator](): AsyncIterator<any>

返品

AsyncIterator<any>

AsyncIterator ストリームを完全に消費する。

[toAsyncStreamable]()

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

返品

ByteReadableStream

継承済みメソッドの詳細

eventNames()

エミッターがリスナーを登録したイベントを一覧表示する配列を返します。

import { EventEmitter } from 'node:events';

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

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

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

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

返品

(string | symbol)[]

継承者 Stream.eventNames(ストリーム・イベントネームズ)

getMaxListeners()

EventEmitter の現在の最大リスナー値を返します。この値は、emitter.setMaxListeners(n) によって設定されるか、既定で events.defaultMaxListenersに設定されます。

function getMaxListeners(): number

返品

number

継承者 Stream.getMaxListeners

pipe<T>(T, PipeOptions)

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

パラメーター

destination: T

options: PipeOptions

返品

継承者 Stream.pipe

setMaxListeners(number)

既定では、EventEmitters は、特定のイベントに対して 10 を超えるリスナーが追加された場合に警告を出力します。これは、メモリリークの検出に役立つ便利な既定値です。 emitter.setMaxListeners() メソッドを使用すると、この特定の EventEmitter インスタンスの制限を変更できます。この値を Infinity (または 0) に設定すると、リスナーの数に制限はありません。

呼び出しをチェーンできるように、EventEmitterへの参照を返します。

function setMaxListeners(n: number): this

パラメーター

n: number

返品

this

継承者 Stream.setMaxListeners

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

Symbol.for('nodejs.rejection')メソッドは、イベントを発生させる際に約束拒否が発生し、エミッターでcaptureRejectionsが有効化された場合に呼び出されます。 events.captureRejectionSymbolの代わりに Symbol.for('nodejs.rejection') を使うことも可能です。

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

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

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

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

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

パラメーター

error: Error

event: string | symbol

args: any[]

継承ストリーム。[捕獲拒否シンボル]