如何在 nodejs 中创建双工流？

Question

我像这样创建一个新的双工流

const Duplex = require('stream').Duplex;
let myStream = new Duplex()

通过 Websocket 我收到块/缓冲区，每次通过 Websocket 有新块进入时，我都会像这样将其添加到流中：

myStream.push(buffer)

然后我将流传输到另一个进程（本例中为 ffmpeg）

myStream.pipe(process.stdout); 这会导致错误NodeError: The _read() method is not implemented 我理解但我不明白为什么以及如何实现它。我还看到，在 Duplex 类构造函数中，您可以传递一个读取函数，但为什么这是必要的？我只想不断将块推送到流中，然后将其通过管道传输到另一个进程

Answer 1

nodejs Duplex 流要求实现者同时指定写入和读取方法：

import stream from 'stream';

const duplex = new stream.Duplex({
  write: (chunk, encoding, next) {
    // Do something with the chunk and then call next() to indicate 
    // that the chunk has been processed. The write() fn will handle
    // data piped into this duplex stream. After the write() has
    // finished, the data will be processed by the read() below.
    next();
  },
  read: ( size ) {
    // Add new data to be read by streams piped from this duplex
    this.push( "some data" )
  }
})

有关流的官方 nodejs 文档可在此处获得：API for Stream Implementers

websocket 场景
上面描述的 websocket 示例可能应该使用 Readable 而不是双工流。双工流在存储转发或处理转发场景中很有用。但是，听起来 websocket 示例中的流仅用于将数据从 websocket 移动到流接口。这可以使用 Readable 来实现：

import stream from 'stream';

const onSocketConnection = ( socket ) => {
    const readable = new stream.Readable({
      // The read logic is omitted since the data is pushed to the socket
      // outside of the script's control. However, the read() function 
      // must be defined.
      read(){}
    });

    socket.on('message', ( data ) => {
        // Push the data on the readable queue
        readable.push( data );
    });

    readable.pipe( ffmpeg );
}