NodeJS 中的流不会发出“结束”、“关闭”和“完成”，但只是其中一些事件答案

【问题标题】：Streams in NodeJS not emitting 'end', 'close' and 'finish but only some of those eventsNodeJS 中的流不会发出“结束”、“关闭”和“完成”，但只是其中一些事件
【发布时间】：2020-08-26 19:10:02
【问题描述】：

我现在正在使用 nodeJS 中的流，我在问自己，如何在我正在使用的所有不同流上发出“结束”、“完成”和“关闭”事件。

运行以下代码：

import * as stream from 'stream';

const clock = () => {
  const readStream = new stream.Readable({
    objectMode: true,
    read() {}
  })
  readStream.push({ time: new Date() });
  readStream.push(null);
  readStream.destroy();
  return readStream;
}

const xformer = () => {
  return new stream.Transform({
    objectMode: true,
    transform: (data, _, done) => {
      done(null, { ...data, transformed: true });
    }
  })
}

const renderer = () => {
  return new stream.Writable({
    objectMode: true,
    write: (data, _, done) => {
      console.log('<-', data)
      done();
    }
  })
}

setInterval(() => {
  console.log("\x1b[34m", "NEW STREAM", "\x1b[0m");
  clock()             // Readable stream
    .pipe(xformer())  // Transform stream
    .pipe(renderer()) // Writable stream
    .on('end', () => {
      console.log("\x1b[32m", "STREAM ENDED", "\x1b[0m");
    })
    .on('close', () => {
      console.log("\x1b[32m", "STREAM CLOSED", "\x1b[0m");
    })
    .on('finish', () => {
      console.log("\x1b[32m", "STREAM FINISHED", "\x1b[0m");
    })
}, 1000);

只给我以下输出：

 NEW STREAM 
<- { time: 2020-05-11T09:22:08.655Z, transformed: true }
 STREAM FINISHED 
 NEW STREAM 
<- { time: 2020-05-11T09:22:09.658Z, transformed: true }
 STREAM FINISHED 
 NEW STREAM 
<- { time: 2020-05-11T09:22:10.662Z, transformed: true }
 STREAM FINISHED 
 NEW STREAM 
<- { time: 2020-05-11T09:22:11.665Z, transformed: true }
 STREAM FINISHED 
 NEW STREAM 
<- { time: 2020-05-11T09:22:12.668Z, transformed: true }
 STREAM FINISHED 
 NEW STREAM 
<- { time: 2020-05-11T09:22:13.667Z, transformed: true }
 STREAM FINISHED 
 NEW STREAM 
<- { time: 2020-05-11T09:22:14.669Z, transformed: true }
 STREAM FINISHED

有谁知道如何正确地做到这一点？或者说得更好……什么是最佳实践？

【问题讨论】：

标签： node.js typescript events stream

【解决方案1】：

您需要单独监听每个流上的事件，作为一个单独的实体。

console.log("NEW STREAM");
// Readable stream
clock()             
  .on('end', () => {
    console.log("STREAM ENDED - READABLE");
  })
  .on('finish', () => {
    console.log("STREAM FINISHED - READABLE");
  })  
  .on('close', () => {
    console.log("STREAM CLOSED - READABLE");
  })
  .on('error', () => {
    console.log("STREAM ERROR - READABLE");
  })

   // Transform stream
  .pipe(xformer())  
  .on('end', () => {
    console.log("STREAM ENDED - TRANSFORM");
  })
  .on('finish', () => {
    console.log("STREAM FINISHED - TRANSFORM");
  })
  .on('close', () => {
    console.log("STREAM CLOSED - TRANSFORM");
  })
  .on('error', () => {
    console.log("STREAM ERROR - TRANSFORM");
  })

  // Writable stream
  .pipe(renderer()) 
  .on('end', () => {
    console.log("STREAM ENDED - WRITABLE");
  })
  .on('finish', () => {
    console.log("STREAM FINISHED - WRITABLE");
  })
  .on('close', () => {
    console.log("STREAM CLOSED - WRITABLE");
  })
  .on('error', () => {
    console.log("STREAM ERROR - WRITABLE");
  })

这应该给你这个......

NEW STREAM 
<- { time: 2020-05-13T10:37:54.267Z, transformed: true }
STREAM CLOSED - READABLE 
STREAM ENDED - READABLE 
STREAM FINISHED - TRANSFORM 
STREAM ENDED - TRANSFORM 
STREAM FINISHED - WRITABLE

或者，

stream.pipeline(
  clock(), xformer(), renderer(), (err) => {
    if (err) {
      console.log(err);
    } else {
      console.log("STREAM COMPLETED");
    }
  }
)

这会给...

<- { time: 2020-05-13T10:41:34.075Z, transformed: true }
 STREAM COMPLETED

【讨论】：