Node FTP 从一台服务器下载文件并上传到另一台服务器

Question

我已经四处寻找了一段时间，this 是我在互联网上找到的与我的问题相关的唯一资源。我正在尝试从一台 ftp 服务器下载文件，然后将它们上传到另一台 ftp 服务器，使用 Promise 一个一个地上传，而无需在此过程中将文件保存在本地。

首先，我从ftp 模块递归调用client.List() 以获取我需要从源ftp 服务器下载的文件路径数组。这很好用。

getRecursively(client, path) {
    var _this = this;
    let downloadList = [];
    let paths = [];
    let promise = new Promise((resolve, reject) => {
        client.list(path, function(err, list) {
            async function loop() {
                for (var i = 0; i < list.length; i++) {
                    if (list[i].type == 'd') {
                        let _list = await _this.getRecursively(client, path + '/' + list[i].name)
                        downloadList = downloadList.concat(_list);
                    } else {
                        if ( list[i].name.match(/\.(jpg|jpeg)$/i) ) {
                            downloadList.push({path: path, name: list[i].name});
                        }
                    }
                }
                console.log("One complete");
                resolve(downloadList);

            }

            loop();
        })
    })
    return promise;

}

接下来，我将遍历文件路径列表并发送使用es6-promise-pool 模块限制的承诺，所以现在它的并发限制设置为 10。

这是每个承诺的样子：

getAndInsert(file) {
    let _this = this;
    let promise = new Promise((resolve, reject) => {
        let c = new Client();
        c.on('ready', () => {
            let d = new Client();
            d.on('ready', () => {
                c.get(file.path + '/' + file.name, function(err, stream) {
                    if (err) {console.log(err); console.log("FILE NAME: " + file.name)}
                    d.put(stream.pipe(passThrough()), '/images/' + file.name, function() {
                        _this.uploadCount += 1;
                        _this.uploadedImages.push(file.name)
                        console.log(_this.uploadCount + '/' + _this._list.length + " uploaded.")
                        c.end();
                        d.end();
                        resolve(true);
                    });

                });

            })

            d.on('error', (err) => {
                if (err) console.log(err);
                _this.onCompleteCallback();
            })

            d.connect(destinationFTP);
        })

        c.on('error', (err) => {
            if (err) console.log(err);
            _this.onCompleteCallback();
        })

        c.connect(sourceFTP);

    })
    return promise;
}

每个承诺都与源和目标 ftp 服务器建立自己的连接。当我调用d.put(stream.pipe(passThrough()) 时，我也在使用stream 模块的Transform 对象。这是那个函数。

    const passThrough = () => {
       var passthrough = new Transform();
       passthrough._transform = function(data, encoding, done) {
           this.push(data);
           done();
       };
       return passthrough;
   }

最后，这是触发 Promise 的主要代码。

*buildPromises(list) {
    for (let i = 0; i < list.length; i++) {
        yield this.getAndInsert(list[i]);
    }
}

let iterator = _this.buildPromises(list);
var pool = new PromisePool(iterator, 10);
pool.start()
    .then(function(){
        console.log("Finished")
    }).catch((err) => {
        console.log(err);
        console.log("error processing pool promise");
    })

这将通过并很好地构建列表，但是当我发送承诺时，我收到以下错误：

Error: write after end
at writeAfterEnd (_stream_writable.js:236:12)
at Transform.Writable.write (_stream_writable.js:287:5)
at Socket.ondata (_stream_readable.js:639:20)
at emitOne (events.js:116:13)
at Socket.emit (events.js:211:7)
at Socket.Readable.read (_stream_readable.js:475:10)
at flow (_stream_readable.js:846:34)
at Transform.<anonymous> (_stream_readable.js:707:7)
at emitNone (events.js:106:13)

它可能会像 5 一样通过然后出错，有时甚至更多，但它似乎非常一致。我也注意到有时我会收到类似的错误，说“文件已在使用中”，但我上传的每个文件都有一个唯一的名称。任何帮助表示赞赏，如果您需要更多信息，我会尽力提供更多信息。谢谢。

Answer 1

所以我找到了解决方案。在我的getAndInsert() 函数中：

c.get(file.path + '/' + file.name, function(err, stream) {
    if (err) {console.log(err); console.log("FILE NAME: " + file.name)}
    d.put(stream.pipe(passThrough()), '/images/' + file.name, function() {
       _this.uploadCount += 1;
       _this.uploadedImages.push(file.name)
       console.log(_this.uploadCount + '/' + _this._list.length + " uploaded.")
       c.end();
       d.end();
       resolve(true);
    });
});

问题出在stream.pipe(passThrough())。好像我在stream 已经结束之后才开始写。这就是解决我的问题的方法：

let chunks = [];
stream.on('data', (chunk) => {
     chunks.push(chunk);
})
stream.on('end', () => {
   d.put(Buffer.concat(chunks), '/images/' + file.name, function() {
      _this.uploadCount += 1;
      _this.uploadedImages.push(file.name)
      console.log(_this.uploadCount + '/' + _this._list.length + " uploaded.")
      c.end();
      d.end();
      resolve(true);
   });
})

当流中的新数据可用时，推送到名为chunks 的数组。流完成后，调用.put并传入Buffer.concat(chunks)

希望这可以帮助任何遇到类似问题的人。