Доступ к сырому потоку файлов из-за огромной загрузки файлов

Question

Я создаю приложение, которое загружает некоторые файлы и отправляет их прямо на S3. Я бы предпочел, чтобы на моем сервере даже не было файла tmp, поэтому я использую модуль Knox и хотел бы взять необработанный поток из Formidable и отправить его через Knox на S3. Я сделал нечто подобное, используя Knox для загрузки файла с помощью этого кода:

knox.downloads.get(widget.download).on('response',function(sres){
    res.writeHead(200, {
        'Content-Type':'application/zip',
        'Content-Length': sres.headers['content-length'],
        'Content-Disposition':'attachment; filename=' + widget.download
    });
    util.pump(sres, res);
}).end();

Теперь я хотел бы сделать что-то похожее в противоположном направлении (загрузка файла из браузера на S3).

Пока что я написал обработчик событий для захвата каждого фрагмента данных из файла по мере его загрузки:

var form = new formidable.IncomingForm();
form.onPart = function(part){
    if(!part.filename){
        form.handlePart(part);
    }else{
        if(part.name == 'download'){
            // Upload to download bucket
            controller.putDownload(part);
        }else{
            // Upload to the image bucket
            controller.putImage(part);
        }
        //res.send(sys.inspect(part));
    }
}
form.parse(req, function(err, fields, files){
    if(err){
        res.json(err);
    }else{
        res.send(sys.inspect({fields:fields, files:files}), {'content-type':'text/plain'});
        //controller.createWidget(res,fields,files);            
    }
});


controller.putDownload = function(part){
    part.addListener('data', function(buffer){
        knox.download.putStream(data,part.filename, function(err,s3res){
            if(err)throwError(err);
            else{
                console.log(s3res);
            }
        });
    })
    knox.downloads.putStream(part, part.filename, function(err,s3res){

        if(err)throwError(err);
        else{
            console.log(s3res);
        }
    });
}

Но событие данных только дает мне буфер. Так можно ли захватить сам поток и отправить его на S3?

Answer 1

То, что вы хотите сделать, это переопределить метод Form.onPart:

IncomingForm.prototype.onPart = function(part) {
  // this method can be overwritten by the user
  this.handlePart(part);
};

Поведение Formidable по умолчанию заключается в записи части в файл. Вы этого не хотите. Вы хотите обработать события 'part' для записи в загрузку knox. Начните с этого:

form.onPart = function(part) {
    if (!part.filename) {
        // let formidable handle all non-file parts
        form.handlePart(part);
        return;
    }

Затем откройте запрос knox и самостоятельно обработайте события необработанной детали:

part.on('data', function(data) {
    req.write(data);
});
part.on('end', function() {
    req.end();
});
part.on('error', function(err) {
    // handle this too
});

В качестве бонуса, если req.write(data) возвращает false, это означает, что буфер отправки заполнен. Вы должны приостановить Грозный парсер. Когда вы получите событие drain из потока Knox, вы должны возобновить Formidable.

Answer 2

Используйте multiparty . Он поддерживает этот вид потоковой передачи, как вы хотите. У него даже есть пример потоковой передачи непосредственно на s3: https://github.com/superjoe30/node-multiparty/blob/master/examples/s3.js

Answer 3

В промежуточном программном обеспечении Express я использую formidable вместе с PassThrough для потоковой передачи файла в S3 (в моем случае, в Minio, совместимый с S3 через Minio SDK; и я считаю, что это работает и для AWS S3)с тем же Minio SDK)

Вот пример кода.

const formidable = require('formidable')
const { PassThrough } = require('stream')

const form = new formidable.IncomingForm()
const pass = new PassThrough()

const fileMeta = {}
form.onPart = part => {
  if (!part.filename) {
    form.handlePart(part)
    return
  }
  fileMeta.name = part.filename
  fileMeta.type = part.mime
  part.on('data', function (buffer) {
    pass.write(buffer)
  })
  part.on('end', function () {
    pass.end()
  })
}
form.parse(req, err => {
  if (err) {
    req.minio = { error: err }
    next()
  } else {
    handlePostStream(req, next, fileMeta, pass)
  }
})

И handlePostStream выглядит как показано ниже:

const uuidv1 = require('uuid/v1')

const handlePostStream = async (req, next, fileMeta, fileStream) => {
  let filename = uuidv1()

  try {
    const metaData = {
      'content-type': fileMeta.type,
      'file-name': Buffer.from(fileMeta.name).toString('base64')
    }

    const minioClient = /* Get Minio Client*/
    await minioClient.putObject(MINIO_BUCKET, filename, fileStream, metaData)

    req.minio = { post: { filename: `${filename}` } }
  } catch (error) {
    req.minio = { error }
  }
  next()
}

Вы можетенайдите исходный код на GitHub и его модульные тесты .

Answer 4

Невозможно перехватить поток, потому что данные должны быть переведены с помощью Formidable.buffer, которое вам дано - это содержимое файла в виде фрагментов buffer.length: это может быть проблемой, потому что, глядя на документы Formidable, выясняется, что до тех пор, пока файл не будет полностью загружен, он не сможет достоверно сообщить размер файла и файл Knox *Метод 1003 * может понадобиться.

Никогда раньше не использовал Knox таким образом, но вам может повезти с чем-то вроде этого:

controller.putDownload = function(part){
    var req = knox.download.put(part.filename, {
      'Content-Type': 'text/plain'
    });
    part.addListener('data', function(buffer){
    req.write(buffer);
    });
    req.on('response', function(res){
       // error checking
    });
    req.end();
}

Немного неуверен в битах проверки ответа, но.... посмотрим, сможете ли вы придать этому форму.Кроме того, Потоковая передача октета из запроса на S3 с помощью knox на node.js также имеет запись, которая может быть вам полезна.