node.js - 来自 SFTP 连接的 Node 异步 ReadStream

Question

因此，我正在创建一个类，最终想要创建一个方法，该方法获取 SFTP 服务器上的文件并生成可以通过管道传输到其他流/函数的读取流。我已经完成了大部分工作，只是我的 readStream 表现得很奇怪。相关代码如下:

const Client = require('ssh2').Client,
      Readable = require('stream').Readable,
      async = require('async');

/**
 * Class Definition stuff
 * ......
 */

getStream(get) {
    const self = this;
    const rs = new Readable;
    rs._read = function() {
        const read = this;
        self.conn.on('ready', function(){
            self.conn.sftp(function(err,sftp) {
                if(err) return err;

                sftp.open(get, 'r', function(err, fd){
                    sftp.fstat(fd, function(err, stats) {

                        let bufferSize = stats.size,
                            chunkSize = 512,//bytes
                            buffer = new Buffer(bufferSize),
                            bytesRead = 0;

                        async.whilst(
                            function () {
                                return bytesRead < bufferSize;
                            },
                            function (done) {
                                sftp.read(fd, buffer, bytesRead, chunkSize, bytesRead,
                                function (err, bytes, buff) {
                                    if (err) return done(err);
                                    // console.log(buff.toString('utf8'));
                                    read.push(buff);
                                    bytesRead += bytes;
                                    done();
                                });
                            },
                            function (err) {
                                if (err) console.log(err);
                                read.push(null);
                                sftp.close(fd);
                            }
                        );

                    });
                });
            });
        }).connect(self.connectionObj);
    }
    return rs;

}

在其他地方，我会这样调用这个方法:

let sftp = new SFTP(credentials);

sftp.getStream('/path/file.csv')
.pipe(toStuff);
.pipe(toOutput);

所以，长话短说。在 SFTP.read 操作期间，read.push(buff) 不断地不断推送文件的相同第一部分。但是，当我 console.log(buff) 时，它会正确地传输完整文件吗？

所以我摸不着头脑，想知道我对读取流做错了什么，它只是推送文件的开头，而不是继续到缓冲区的下一部分。

以下是 SSH2 SFTP 客户端的文档:https://github.com/mscdex/ssh2-streams/blob/master/SFTPStream.md

我用这个问题作为我上面写的内容的灵感:node.js fs.read() example

这是相似/相关的:Reading file from SFTP server using Node.js and SSH2

Answer 1

好吧，经历了很多麻烦之后，我意识到我犯了一些错误。首先，每次流准备好读取更多数据时都会调用 _read 函数，这意味着每次调用 _read 时都会启动 SFTP 连接。这也意味着 sftp.read() 函数每次都会重新启动，将起点重置回开头。

我需要一种方法来首先设置连接，然后读取和流式传输文件数据，所以我选择了库 noms 。如果有人感兴趣的话，这是最终代码:

getStream (get) {
    const self = this;

    let connection,
        fileData,
        buffer,
        totalBytes = 0,
        bytesRead = 0;

    return nom(
        // _read function
        function(size, next) {
            const read = this;

            // Check if we're done reading
            if(bytesRead === totalBytes) {
                connection.close(fileData);
                connection.end();
                self.conn.end();
                console.log('done');
                return read.push(null);
            }

            // Make sure we read the last bit of the file
            if ((bytesRead + size) > totalBytes) {
                size = (totalBytes - bytesRead);
            }

            // Read each chunk of the file
            connection.read(fileData, buffer, bytesRead, size, bytesRead,
                function (err, byteCount, buff, pos) {
                    // console.log(buff.toString('utf8'));
                    // console.log('reading');
                    bytesRead += byteCount;
                    read.push(buff);
                    next();
                }
            );
      },
      // Before Function
      function(start) {
          // setup the connection BEFORE we start _read
          self.conn.on('ready', function(){
              self.conn.sftp(function(err,sftp) {
                  if(err) return err;
                  sftp.open(get, 'r', function(err, fd){
                      sftp.fstat(fd, function(err, stats) {
                          connection = sftp;
                          fileData = fd;
                          totalBytes = stats.size;
                          buffer = new Buffer(totalBytes);
                          console.log('made connection');
                          start();
                      });
                  });
              });
          }).connect(self.connectionObj);
      })
}

始终寻求反馈。它的运行速度没有我希望的那么快，所以如果您有加快流速度的想法，请告诉我。