NodeJS中的流没有发出'end'、'close'和'finish'，而是只发出其中的一些事件。

Question

我现在正在玩nodeJS中的streams，我问自己，如何才能让我正在玩的所有不同的streams都能发出'end'、'finish'和'close'事件。

运行下面的代码。

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 

有谁知道怎么做才正确？或者更好的说......什么是最佳实践？

Answer 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