通过异步迭代简化 Node.js 流

/**
 * @returns an asynchronous iterable
 */
async function* asyncGenerator(asyncIterable) {
  for await (const item of asyncIterable) { // input
    if (···) {
      yield '> ' + item; // output
    }
  }
}

/** @returns a Promise */
async function asyncFunction() {
  /*···*/
}

/** @returns an async iterable */
async function* asyncGeneratorFunction() {
  /*···*/
}

import * as stream from 'stream';

/**
 * Reads all the text in a readable stream
 * and returns it as a string,
 * via a Promise.
 * @param {stream.Readable} readable
 */
function readableToString(readable) {
  return new Promise((resolve, reject) => {
    let data = '';
    readable.on('data', function (chunk) {
      data += chunk;
    });
    readable.on('end', function () {
      resolve(data);
    });
    readable.on('error', function (err) {
      reject(err);
    });
  });
}

import * as fs from 'fs';

const readableStream = fs.createReadStream('tmp/test.txt', {
  encoding: 'utf8',
});

console(await readableToString(readableStream));
// 'This is a test!\n'

import * as stream from 'stream';

function* gen() {
  yield 'One line\n';
  yield 'Another line\n';
}
const readableStream = stream.Readable.from(gen(), { encoding: 'utf8' });
assert.equal(
  await readableToString(readableStream),
  'One line\nAnother line\n',
);

import { Readable } from 'stream';

const str = 'Some text!';
const readable = Readable.from(str, { encoding: 'utf8' });
assert.equal(await readableToString(readable), 'Some text!');

import * as fs from 'fs';

async function logChunks(readable) {
  for await (const chunk of readable) {
    console.log(chunk);
  }
}

const readable = fs.createReadStream('tmp/test.txt', { encoding: 'utf8' });
logChunks(readable);

// Output:
// 'This is a test!\n'

import { Readable } from 'stream';

async function readableToString2(readable) {
  let result = '';
  for await (const chunk of readable) {
    result += chunk;
  }
  return result;
}

const readable = Readable.from('Good morning!', { encoding: 'utf8' });
assert.equal(await readableToString2(readable), 'Good morning!');

import { Readable } from 'stream';

/**
 * @param chunkIterable An asynchronous or synchronous iterable
 * over “chunks” (arbitrary strings)
 * @returns An asynchronous iterable over “lines”
 * (strings with at most one newline that always appears at the end)
 */
async function* chunksToLines(chunkIterable) {
  let previous = '';
  for await (const chunk of chunkIterable) {
    previous += chunk;
    while (true) {
      const eolIndex = previous.indexOf('\n');
      if (eolIndex < 0) break;

      // line includes the EOL
      const line = previous.slice(0, eolIndex + 1);
      yield line;
      previous = previous.slice(eolIndex + 1);
    }
  }
  if (previous.length > 0) {
    yield previous;
  }
}

async function* numberLines(lineIterable) {
  let lineNumber = 1;
  for await (const line of lineIterable) {
    yield lineNumber + ' ' + line;
    lineNumber++;
  }
}

async function logLines(lineIterable) {
  for await (const line of lineIterable) {
    console.log(line);
  }
}

const chunks = Readable.from('Text with\nmultiple\nlines.\n', {
  encoding: 'utf8',
});
logLines(numberLines(chunksToLines(chunks)));

// Output:
// '1 Text with\n'
// '2 multiple\n'
// '3 lines.\n'

const writableStream = fs.createWriteStream('tmp/log.txt', {
  encoding: 'utf8',
});

import * as util from 'util';
import * as stream from 'stream';
import * as fs from 'fs';
import { once } from 'events';

const finished = util.promisify(stream.finished); // (A)

async function writeIterableToFile(iterable, filePath) {
  const writable = fs.createWriteStream(filePath, { encoding: 'utf8' });
  for await (const chunk of iterable) {
    if (!writable.write(chunk)) {
      // (B)
      // Handle backpressure
      await once(writable, 'drain');
    }
  }
  writable.end(); // (C)
  // Wait until done. Throws if there are errors.
  await finished(writable);
}

await writeIterableToFile(['One', ' line of text.\n'], 'tmp/log.txt');
assert.equal(
  fs.readFileSync('tmp/log.txt', { encoding: 'utf8' }),
  'One line of text.\n',
);

import * as stream from 'stream';
import * as fs from 'fs';
const pipeline = util.promisify(stream.pipeline);

async function writeIterableToFile(iterable, filePath) {
  const readable = stream.Readable.from(iterable, { encoding: 'utf8' });
  const writable = fs.createWriteStream(filePath);
  await pipeline(readable, writable); // (A)
}
await writeIterableToFile(['One', ' line of text.\n'], 'tmp/log.txt');
// ···

await pipeline(readable, writable);