节点JS |如何将图像从套接字服务器发送到其他服务器

Question

我正在做一个新项目，我想问你一个我面临的问题。

我有一个运行 socket.io 模块的网络服务器。该服务器侦听其端口（3012）并使用套接字将图像流式传输到客户端。

我的主服务器使用不同的端口 (4049)。该服务器的前端部分包括一个空白容器。

我想找到一种方法从 Socket 服务器向我的主服务器发送流式图像，我的主服务器每次都作为新客户端进行监听。

谢谢

Answer 1

您需要做的是每次在客户端上的图像Readable stream 上触发.on('data') 事件时将块发送到套接字服务器，然后当您收到块时将它们写入Writeable Stream on websocket 服务器端。

有几点需要注意：

• 您需要检测服务器上的 EOF（检查特定文件类型的 EOF 字节），或从客户端发出某种标头。示例。

const EOF = Buffer.alloc(6);

// Client Side

client.sendBytes(EOF); // on end

// Server
if(chunk.slice(-6).compare(EOF) === 0)
  console.log('File EOF close write stream');

• 如果您同时读取多个图像，则需要为每个块添加一个标识符，以便在服务器端正确写入。标识符应始终具有相同的长度，以便您可以在服务器端正确分割缓冲区。

const imageOne = fs.createReadStream('./image-1.jpg');
const imageTwo = fs.createReadStream('./image-2.jpg');

// This will be mixed, and you'll end up with a broken image
imageOne.on('data', chunk => client.sendBytes(chunk)); // add identifier
imageTwo.on('data', chunk => client.sendBytes(chunk)); // add identifier

以下是使用websocket 包的示例。

服务器

/* rest of implementation */
wsServer.on('request', function(request) {
  const connection = request.accept(null, request.origin);

  const JPEG_EOF = Buffer.from([0xFF, 0xD9]);
  let stream = null;

  connection.on('message', function(message) {
    if (message.type === 'binary') {

      if(!stream) {
        stream = fs.createWriteStream(generateImageName())
         // this could be any Writable Stream
         // not necessarily a file stream.
         // It can be an HTTP request for example.
      }

      // Check if it's the end
      if(JPEG_EOF.compare(message.binaryData) === 0) {
        console.log('done');
        stream.end(message.binaryData);
        stream = null;
        return;
      }

      // You will need to implement a back pressure mechanism
      stream.write(message.binaryData)    
    }
  });
});

客户

/** ... **/
client.on('connect', function(connection) {

  fs.createReadStream('./some-image.jpg')
    .on('data', chunk => {
        connection.sendBytes(chunk);
    });

});
/** ... **/

上面的例子只处理jpeg图像，因为它直接检查jpeg的最后2个字节，你可以实现其他文件类型的逻辑。

在示例中，我假设每次连接一次只能流式传输 1 张图像，否则会混淆。

现在您需要为.write 实现一个背压机制，这意味着您必须检查返回值并等待drain 事件。 稍后当我有更多时间时，我将提交一个带有自定义 Readable stream 的示例，正​​确处理背压

更新

使用下面的 sn-p，由于实现了 Readable 流，我们可以使用 .pipe 来处理背压。

const { Readable } = require('stream');

class ImageStream extends Readable {

  constructor() {
    super();

    this.chunks = [];
    this.EOF = Buffer.from([0xFF, 0xD9]);
  }

  add(chunk) {
    this.chunks.push(chunk);    

    if(this.isPaused()) {
      this.resume();

      // Need to call _read if instead of this.push('') you return without calling .push
      // this._read(); 
    }
  }

  _read() {

    const chunk = this.chunks.shift();

    if(!chunk) { // nothing to push, pause the stream until more data is added
      this.pause(); 
      return this.push(''); // check: https://nodejs.org/api/stream.html#stream_readable_push
      // If you return without pushing
      // you need to call _read again after resume
    }

    this.push(chunk);

    // If the last 2 bytes are not sent in the same chunk
    // This won't work, you can implement some logic if that can happen
    // It's a really edge case.
    const last = chunk.slice(-2);
    if(this.EOF.compare(last) == 0)
      this.push(null); // Image done, end the stream.

  }
}

/* ... */
wsServer.on('request', function(request) {
  const connection = request.accept(null, request.origin);

  let stream = null;

  connection.on('message', function(message) {
    if (message.type === 'binary') {

      if(!stream) {
        stream = new ImageStream();
        stream.pipe(fs.createWriteStream(generateImageName()));
        // stream.pipe(request(/* ... */));
        stream.on('end', () => {
          stream = null; // done
        });
      }

      stream.add(message.binaryData);
    }
  });

  connection.on('close', function(connection) {
    // close user connection
  });
});