Node.js 10の非同期ループとストリームAPI

今月、Node.jsの10番目のバージョンがリリースされます。このバージョンでは、非同期のfor-await-ofループの出現によって引き起こされるフロー（ readable-stream ）の動作の変更を待っています。 それが何であり、何を準備する必要があるかを見てみましょう。

For-await-ofコンストラクト。

はじめに、簡単な例で非同期ループがどのように機能するかを見てみましょう。 明確にするために、完了したプロミスを追加します。

const promises = [ Promise.resolve(1), Promise.resolve(2), Promise.resolve(3), ];
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

通常のループはpromises配列を通過し、値自体を返します。

 for (const value of promises) { console.log(value); } // > Promise({resolved: 1}) // > Promise({resolved: 2}) // > Promise({resolved: 3})
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

非同期ループは、promiseが解決するのを待ち、promiseが返す値を返します。

 for await (const value of promises) { console.log(value); } // > 1 // > 2 // > 3
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

Node.jsの以前のバージョンで非同期ループを機能させるには、 --harmony_async_iteration



フラグを使用します。

ReadableStreamおよびfor-await-of

ReadableStreamオブジェクトはSymbol.asyncIterator



プロパティを受け取りました。これにより、for-await-ofループに渡すこともできます。 fs.createReadableStream



例fs.createReadableStream



取り上げfs.createReadableStream



。

 const readStream = fs.createReadStream(file); const chunks = []; for await (const chunk of readStream) { chunks.push(chunk); } console.log(Buffer.concat(chunks));
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

例からわかるように、ここでon('data', ...



およびon('end', ...



の呼び出しを取り除き、コード自体がより視覚的で予測可能になり始めました。

非同期ジェネレーター

場合によっては、受信データの追加処理が必要になることがありますが、これには非同期ジェネレーターが使用されます。 正規表現ファイルで検索を実装できます。

 async function * search(needle, chunks) { let pos = 0; for await (const chunk of chunks) { let string = chunk.toString(); while (string.length) { const match = string.match(needle); if (! match) { pos += string.length; break; } yield { index: pos + match.index, value: match[0], }; string = string.slice(match.index + match[0].length); pos += match.index; } } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

何が起こったのか見てみましょう：

 const stream = fs.createReadStream(file); for await (const {index, value} of search(/(a|b)c/, stream)) { console.log('found "%s" at %s', value, index); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

同意します。非常に便利です。その場で文字列をオブジェクトに変換し、TransformStreamを使用して、2つの異なるストリームで発生する可能性のあるエラーをキャッチする方法を考える必要がありませんでした。

Unixのようなスレッドの例

ファイルを読み取るタスクは非常に一般的ですが、網羅的ではありません。 UNIXパイプラインのような出力のストリーム処理が必要な場合を見てみましょう。 これを行うには、非同期ジェネレーターを使用します。非同期ジェネレーターを使用して、 ls



結果をスキップします。

最初に子プロセスconst subproc = spawn('ls')



、次に標準出力を読み取ります。

 for await (const chunk of subproc.stdout) { // ... }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

また、stdoutはBufferオブジェクトの形式で出力を生成するため、最初に行うことは、Buffer型の出力をStringに出力するジェネレーターを追加することです。

 async function *toString(chunks) { for await (const chunk of chunks) { yield chunk.toString(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、出力を行ごとに分割する単純なジェネレーターを作成します。 createReadStreamから転送されるデータ部分の最大長は制限されていることを考慮することが重要です。つまり、行全体、非常に長い文字列、または複数の行を同時に受信できます。

 async function *chunksToLines(chunks) { let previous = ''; for await (const chunk of chunks) { previous += chunk; while (true) { const i = previous.indexOf('\n'); if (i < 0) { break; } yield previous.slice(0, i + 1); previous = previous.slice(i + 1); } } if (previous.length > 0) { yield previous; } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

見つかった各値にはまだ改行が含まれているため、ジェネレータを作成して、値が空白からぶら下がらないようにします。

 async function *trim(values) { for await (const value of values) { yield value.trim(); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最後のアクションは、コンソールへの行ごとの出力です。

 async function print(values) { for await (const value of values) { console.log(value); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

結果のコードを結合します。

 async function main() { const subproc = spawn('ls'); await print(trim(chunksToLines(toString(subproc.stdout)))); console.log('DONE'); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ご覧のとおり、コードは多少読みにくいことがわかりました。 さらにいくつかの呼び出しまたはパラメーターを追加したい場合は、結果としてポリッジを取得します。 回避してコードをより線形にするために、 pipe



関数を追加しましょう。

 function pipe(value, ...fns) { let result = value; for (const fn of fns) { result = fn(result); } return result; }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これで、呼び出しを次の形式に減らすことができます。

 async function main() { const subproc = spawn('ls'); await pipe( subproc.stdout, toString, chunksToLines, trim, print, ); console.log('DONE'); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

演算子|>

すぐに新しいパイプライン演算子をJS標準に含める必要があることに留意してください。これにより、 pipe



が現在行っているのと同じことができるようになります。

 async function main() { const subproc = spawn('ls'); await subproc.stdout |> toString |> chunksToLines |> trim |> print; console.log('DONE'); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

おわりに

ご覧のとおり、非同期ループとイテレータにより、言語はさらに表現力豊かで、理解しやすくなりました。 コールバックからの地獄はさらに過去へと進み、やがて孫を怖がらせる恐怖物語になります。 そして、ジェネレーターはJS階層の中でその位置を占めるようで、意図したとおりに使用されます。

この記事の基礎は、Axel Rauschmeier によるNode.jsでのネイティブな非同期反復の使用でした。

トピックを続ける

• イテラブルとイテレータ [EN]。
• 非同期反復 [EN]。