如何在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类构造函数中，您可以传递read函数，但是为什么这是必需的呢？我只是想连续将大块推入流中，然后将其通过管道传送到另一个进程

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
    next();
  },
  read: ( size ) {
    // Add new data to be read by the `write()` function above, before being piped to the next stream readable.
    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 );
}