stream: construct

[ronag]

Provide a standardized way of asynchronously creating and initializing resources before performing any work.

Some streams need to first asynchronously create resources before they can perform any work. Currently this is implemented in the different stream implementations which both makes things partly duplicated, more difficult, more error prone and often incorrect to some degree (e.g. 'open' and 'ready' are emitted after 'close').

This PR provides a standardized way of asynchronously constructing streams and handles the "pending" state that occurs until construction has either completed or failed.

This will allow further simplification and improved consistency for various stream implementations such as fs and net stream.

Passes the graceful-fs test suite.

This will make it possible to easily implement more complex stream such as e.g. fs streams:

const { Writable } = require('stream');
const fs = require('fs')

class WriteStream extends Writable {
  constructor (options) {
    options.autoDestroy = true;
    super(options);
  }
  _construct({ filename }, callback) {
    this.filename = filename;
    this.fd = null;
    fs.open(this.filename, (fd, err) => {
      if (err) {
        callback(err);
      } else {
        this.fd = fd;
        callback();
      }
    });
  }
  _write(chunk, encoding, callback) {
    fs.write(this.fd, chunk, callback);
  }
  _destroy(err, callback) {
    if (this.fd) {
      fs.close(this.fd, (er) => callback(er || err));
    } else {
      callback(err);
    }
  }
}

const { Readable } = require('stream');
const fs = require('fs')

class ReadStream extends Readable {
  constructor (options) {
    options.autoDestroy = true;
    super(options);
  }
  _construct({ filename }, callback) {
    this.filename = filename;
    this.fd = null;
    fs.open(this.filename, (fd, err) => {
      if (err) {
        callback(err);
      } else {
        this.fd = fd;
        callback();
      }
    });
  }
  _read(n) {
    const buf = Buffer.alloc(n);
    fs.read(this.fd, buf, 0, n, null, (err, bytesRead) => {
      if (err) {
        this.destroy(err);
      } else {
        this.push(bytesRead > 0 ? buf : null);
      }
    });
  }
  _destroy(err, callback) {
    if (this.fd) {
      fs.close(this.fd, (er) => callback(er || err));
    } else {
      callback(err);
    }
  }
}

Furthermore it makes it easier to e.g. add transforms into pipeline by inlining initialization:

const { Duplex } = require('stream');
const fs = require('fs')

stream.pipeline(
  fs.createReadStream('object.json')
    .setEncoding('utf-8'),
  new Duplex({
    construct (options, callback) {
      this.data = '';
      callback();
    },
    transform (chunk, encoding, callback) {
      this.data += chunk;
      callback();
    },
    flush(callback) {
      try {
        // Make sure is valid json.
        JSON.parse(this.data);
        this.push(this.data);
      } catch (err) {
        callback(err);
      }
    }
  }),
  fs.createWriteStream('valid-object.json')
);

Semver

I think this should be semver-major.

• fs: Emit some previously synchronous errors asynchronously. See updated test. Bug fix.
• fs: open() is removed but it still works if monkey-patched.
• fs: Don't emit 'close' twice if emitClose: true.
• fs: stream.fd is nulled during destroy instead of final. See updated test.

Otherwise, this should be pretty non breaking as far as I can tell. graceful-fs tests pass and there are tests for compat.

The changes are mostly opt-in through the construct method, except for the previously listed updates to fs streams.

Checklist

• make -j4 test (UNIX), or vcbuild test (Windows) passes
• tests and/or benchmarks are included
• documentation is changed or added
• commit message follows commit guidelines

Refs: #29314, #23133

NOTE TO SELF: After merge look into:

• Remove emitClose and make 'close' behaviour align with streams.
• Deprecate close(cb) in favour of end(cb). Make close(cb) call end(cb).
• https://github.com/nxtedition/node/pull/new/stream-sync-final

[ronag]

@Fishrock123 I just started looking a bit into Bob streams and I'm curious about your take on this.

[mscdex]

Can we use a different event name? I recognize fs.ReadStream/fs.WriteStream already use it, but by using this at the stream.Readable/stream.Writable level it can cause disruption in userland modules that may already use this event name (I'm speaking from experience here as ssh2-streams has used 'ready' since 2016 -- long before even fs started using it).

[ronag]

Can we use a different event name?

Sure, though I'm not sure what that could be, https://www.thesaurus.com/browse/ready?s=t.

• 'primed'?
• 'init'?

[ronag]

rebased

[ronag]

@addaleax

[ronag]

@mcollina is it an option to use Symbols as event "names" to avoid collisions?

e.g.

const r = new Readable();
r
 .on(Readable.READY, () => {})
 .on(Readable.DATA, (buf) => {})
 .on(Readable.ERROR, (err) => {});

Probably a too big of a change... just a possibly interesting idea...

[mscdex]

I'm not sure what would be a better name that reduces the risk of clashing with userland. Perhaps an added prefix of some kind might help with that, like 'stream-init' or similar. It might seem redundant but it would help differentiate from someone already emitting 'init', as I think that would be more common than 'stream-init'.

[mcollina]

I’ve tagged semver-major out of cautiousness because it’s a significant refactor.

[ronag]

@mcollina do you have any feedback or preference regarding @mscdex's concern?

[mcollina]

I agree with @mscdex concern, we need to find a different name.

[Trott]

This needs a rebase.

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/31547/

[mafintosh]

The async construction feature is def nice and tricky to impl atm.
Do I read this correctly that no event is added after construction now in the latest draft?
Name collisions with that event would be my main concern with userland streams so if no event then I'm soft +1

[ronag]

Do I read this correctly that no event is added after construction now in the latest draft?

We do add an event but it's a Symbol to avoid collisions. Only used internally to keep the implementation simple.

[nodejs-github-bot]

CI: https://ci.nodejs.org/job/node-test-pull-request/31565/ ✅

[ronag]

@addaleax @lpinca I assume you are still LGTM?

[ronag]

CITGM: https://ci.nodejs.org/view/Node.js-citgm/job/citgm-smoker/2386/
CITGM (master): https://ci.nodejs.org/view/Node.js-citgm/job/citgm-smoker/2387/

[addaleax]

Yes 👍

[ronag]

@mcollina I'm getting a suspicious CITGM failures on pino. Any insight? https://ci.nodejs.org/view/Node.js-citgm/job/citgm-smoker/2386/nodes=rhel7-s390x/testReport/(root)/citgm/pino_v6_3_0/ Is this a pino issue or issue with this PR?

[mcollina]

Pino is ok with the change. A test fail because of another deprecation.

[ronag]

Landed in 9949a2e...54b36e4

[rvagg]

arm-fanned has been offline for a while now and it's back on and should be working again but this PR is causing a perma-fail that wasn't caught because it was merged while they were offline.

17:42:04 not ok 583 parallel/test-fs-stream-construct
17:42:04   ---
17:42:04   duration_ms: 240.87
17:42:04   severity: fail
17:42:04   exitcode: -15
17:42:04   stack: |-
17:42:04     timeout
17:42:04   ...

Sorry @ronag, would you mind taking a look?

I haven't had a look at what this test does but a brief scan suggests that a timeout is probably odd so maybe the slower platforms are catching a genuine bug? If not, either use the platformTimeout() to adjust timings in this test for slower machines or maybe even skip it entirely if it can't be avoided.

[ronag]

I haven't had a look at what this test does but a brief scan suggests that a timeout is probably odd so maybe the slower platforms are catching a genuine bug? If not, either use the platformTimeout() to adjust timings in this test for slower machines or maybe even skip it entirely if it can't be avoided.

There is no timeout. It's probably the test not completing for whatever reason.

[ronag]

@rvagg Do you have a machine I can access for reproducing?

[gabrielschulhof]

@ronag you can request access to a machine in https://github.com/nodejs/build/issues/new