readable.destroy() 不同时发出 'close' 和 'error' 事件 Node.js

Question

我写了一些碰巧有效的代码，但我认为 readable.destroy()（例如，src.destroy()，在我的示例中）应该发出 close 和 error 事件...

这里有一个简单的例子来说明我的困惑：

const fs = require('fs');
const src = fs.createReadStream('streaming-example.js');
const dst = fs.createWriteStream('streaming-example.txt');
src.pipe(dst);
src.on('readable', () => {
  let chunk;
  while (null !== (chunk = src.read())) {
    /**
    * This should cause 'error' and 'close' events to emit
    * @see https://nodejs.org/dist/latest-v11.x/docs/api/stream.html#stream_readable_destroy_error
    */
    src.destroy();
  }
});
src.on('close', () => console.log(`'close' event emitted`));
src.on('end', () => console.log(`'end' event emitted`));
src.on('error', (err) => console.log(`'error' event emitted`));

这是该程序的示例运行：

$ node streaming-example.js 
'close' event emitted
$

（并且，它恰好也完成写入名为 streaming-example.txt 的新文件）

如果不清楚，我预计会发出 close 和 error 事件，进而触发相应的回调。但是，似乎只发出了 close 事件。

error 事件发射发生了什么？

Answer 1

事实证明，对 node.js 代码库进行深入研究可以阐明这种混淆。

正在查看 node/lib/_stream_readable.js, we see that the destroy function is defined in node/lib/internal/streams/destroy.js。在这里，我们直接被告知（通过简单的 javascript）error 事件只会在恰好将一个真值传递给 destroy 函数时触发（从语义上讲，这将是调用 destroy 的相同客户端代码）。

例如，如果我们通过更改

简单地修改上面的示例代码

readable.destroy();

成为

readable.destroy(true); // or, more semantically correct, some Error value

我们得到以下输出：

$ node streaming-example.js 
'error' event emitted with err: true
$

但是，现在我们失去了 close 事件。那么...刚刚发生了什么？

再次查看 node/lib/internal/streams/destroy.js，我们注意到以下特殊情况逻辑：

const readableDestroyed = this._readableState &&
  this._readableState.destroyed;
const writableDestroyed = this._writableState &&
  this._writableState.destroyed;

if (readableDestroyed || writableDestroyed) {
  if (cb) {
    cb(err);
  } else if (err &&
              (!this._writableState || !this._writableState.errorEmitted)) {
    process.nextTick(emitErrorNT, this, err);
  }
  return this;
}

// We set destroyed to true before firing error callbacks in order
// to make it re-entrance safe in case destroy() is called within callbacks

if (this._readableState) {
  this._readableState.destroyed = true;
}

// If this is a duplex stream mark the writable part as destroyed as well
if (this._writableState) {
  this._writableState.destroyed = true;
}

error 已正确发出，但 close 似乎未发出，这一事实向我们暗示我们正在处理双工流。我们可以只查找那是什么，但为了简单起见，让我们坚持使用计算机告诉我们的内容。用这个 while-loop

替换原来的 while-loop

while (null !== (chunk = src.read())) {
  console.log('before-destroy, src:', JSON.stringify(src));
  src.destroy(true);  // should cause 'close' and 'error' events to emit
  console.log('after-destroy, src:', JSON.stringify(src));       
}

我们得到以下输出：

$ node streaming-example.js 
before-destroy, src: {"_readableState":{"objectMode":false,"highWaterMark":65536,"buffer":{"head":null,"tail":null,"length":0},"length":0,"pipes":{"_writableState":{"objectMode":false,"highWaterMark":16384,"finalCalled":false,"needDrain":false,"ending":false,"ended":false,"finished":false,"destroyed":false,"decodeStrings":true,"defaultEncoding":"utf8","length":633,"writing":true,"corked":0,"sync":false,"bufferProcessing":false,"writelen":633,"bufferedRequest":null,"lastBufferedRequest":null,"pendingcb":1,"prefinished":false,"errorEmitted":false,"emitClose":false,"autoDestroy":false,"bufferedRequestCount":0,"corkedRequestsFree":{"next":null,"entry":null}},"writable":true,"_events":{},"_eventsCount":5,"path":"streaming-example.txt","fd":24,"flags":"w","mode":438,"autoClose":true,"bytesWritten":0,"closed":false},"pipesCount":1,"flowing":false,"ended":false,"endEmitted":false,"reading":true,"sync":false,"needReadable":true,"emittedReadable":false,"readableListening":true,"resumeScheduled":false,"paused":false,"emitClose":false,"autoDestroy":false,"destroyed":false,"defaultEncoding":"utf8","awaitDrain":0,"readingMore":true,"decoder":null,"encoding":null},"readable":true,"_events":{"end":[null,null,null]},"_eventsCount":5,"path":"streaming-example.js","fd":23,"flags":"r","mode":438,"end":null,"autoClose":true,"bytesRead":633,"closed":false}
after-destroy, src: {"_readableState":{"objectMode":false,"highWaterMark":65536,"buffer":{"head":null,"tail":null,"length":0},"length":0,"pipes":{"_writableState":{"objectMode":false,"highWaterMark":16384,"finalCalled":false,"needDrain":false,"ending":false,"ended":false,"finished":false,"destroyed":false,"decodeStrings":true,"defaultEncoding":"utf8","length":633,"writing":true,"corked":0,"sync":false,"bufferProcessing":false,"writelen":633,"bufferedRequest":null,"lastBufferedRequest":null,"pendingcb":1,"prefinished":false,"errorEmitted":false,"emitClose":false,"autoDestroy":false,"bufferedRequestCount":0,"corkedRequestsFree":{"next":null,"entry":null}},"writable":true,"_events":{},"_eventsCount":5,"path":"streaming-example.txt","fd":24,"flags":"w","mode":438,"autoClose":true,"bytesWritten":0,"closed":false},"pipesCount":1,"flowing":false,"ended":false,"endEmitted":false,"reading":true,"sync":false,"needReadable":true,"emittedReadable":false,"readableListening":true,"resumeScheduled":false,"paused":false,"emitClose":false,"autoDestroy":false,"destroyed":true,"defaultEncoding":"utf8","awaitDrain":0,"readingMore":true,"decoder":null,"encoding":null},"readable":true,"_events":{"end":[null,null,null]},"_eventsCount":5,"path":"streaming-example.js","fd":null,"flags":"r","mode":438,"end":null,"autoClose":true,"bytesRead":633,"closed":false}
'error' event emitted
$

这告诉我们，我们可能正在处理双工流，或者至少这向我们解释了为什么只发出 error 事件（因为，特别是只看 after-destroy 输出，this._readableState 和 this._writeableState 都是真值，因此 destroy 函数将局部变量 readableDestroyed 和 writeableDestroyed 设置为真，我们从console.log this._writableState.errorEmitted 是 false，所以 process.nextTick(emitErrorNT, this, err); 在退出 destroy 函数之前执行。

问题现已得到充分回答。

另外，很高兴知道 duplex 流与其他类型的流之间的区别。为此，this portion of the node.js documentation 的快速参考是一个开始。

那么，当 close 和 error 事件都被发出时（即，当我们不处理 duplex 流时？下面的代码和执行就是这样做的，如下所示：

const readable = process.stdin;
const writable = process.stdout;
readable.setEncoding('utf8');
readable.on('readable', () => {
  let chunk;
  while ((chunk = readable.read()) !== null) {
    writable.write(`data: ${chunk}`);
  }
  readable.destroy(true);
});
readable.on('close', () => console.log(`'close' event emitted`));
readable.on('error', (err) => console.log(`'error' event emitted with err:`, err));

执行此脚本以及一些 I/O（键入 asdf，然后按 return/enter 键），提供以下输出：

$ node streaming-example2.js 
asdf
data: asdf
'error' event emitted with err: true
'close' event emitted
$

readable.destroy() 不同时发出 'close' 和 'error' 事件 Node.js

readable.destroy() not emitting both 'close' and 'error' events Node.js

javascript

node.js

events

streaming

dom-events