Leiten Sie einen Stream an s3.upload ()

Ich verwende derzeit ein node.js-Plugin namens s3-upload-stream, um sehr große Dateien an Amazon S3 zu streamen. Es verwendet die mehrteilige API und funktioniert größtenteils sehr gut.

Dieses Modul zeigt jedoch sein Alter an und ich musste bereits Änderungen daran vornehmen (der Autor hat es ebenfalls abgelehnt). Heute bin ich auf ein anderes Problem mit Amazon gestoßen, und ich würde gerne die Empfehlung des Autors annehmen und anfangen, das offizielle aws-sdk zu verwenden, um meine Uploads durchzuführen.

ABER.

Das offizielle SDK scheint Piping to nicht zu unterstützen s3.upload(). Die Natur von s3.upload ist, dass Sie den lesbaren Stream als Argument an den S3-Konstruktor übergeben müssen.

Ich habe mehr als 120 Benutzercodemodule, die verschiedene Dateiverarbeitungen durchführen, und sie sind unabhängig vom endgültigen Ziel ihrer Ausgabe. Der Motor gibt ihnen einen ableitbaren, beschreibbaren Ausgabestream und sie leiten ihn weiter. Ich kann ihnen kein AWS.S3Objekt geben und sie bitten, es aufzurufen upload(), ohne allen Modulen Code hinzuzufügen. Der Grund, den ich benutzte, s3-upload-streamwar, dass es Rohrleitungen unterstützte.

Gibt es eine Möglichkeit, aws-sdk zu s3.upload()etwas zu machen , zu dem ich den Stream leiten kann?

Schließen Sie die S3- upload()Funktion mit dem stream.PassThrough()Stream node.js ein .

Hier ist ein Beispiel:

inputStream
  .pipe(uploadFromStream(s3));

function uploadFromStream(s3) {
  var pass = new stream.PassThrough();

  var params = {Bucket: BUCKET, Key: KEY, Body: pass};
  s3.upload(params, function(err, data) {
    console.log(err, data);
  });

  return pass;
}

Eine etwas späte Antwort, es könnte hoffentlich jemand anderem helfen. Sie können sowohl den beschreibbaren Stream als auch das Versprechen zurückgeben, sodass Sie nach Abschluss des Uploads Antwortdaten erhalten.

const AWS = require('aws-sdk');
const stream = require('stream');

const uploadStream = ({ Bucket, Key }) => {
  const s3 = new AWS.S3();
  const pass = new stream.PassThrough();
  return {
    writeStream: pass,
    promise: s3.upload({ Bucket, Key, Body: pass }).promise(),
  };
}

Und Sie können die Funktion wie folgt verwenden:

const { writeStream, promise } = uploadStream({Bucket: 'yourbucket', Key: 'yourfile.mp4'});
const readStream = fs.createReadStream('/path/to/yourfile.mp4');

const pipeline = readStream.pipe(writeStream);

Jetzt können Sie entweder das Versprechen überprüfen:

promise.then(() => {
  console.log('upload completed successfully');
}).catch((err) => {
  console.log('upload failed.', err.message);
});

Oder als stream.pipe()Rückgabe stream.Writable, das Ziel (writeStream-Variable oben), das eine Kette von Pipes zulässt, können wir auch seine Ereignisse verwenden:

 pipeline.on('close', () => {
   console.log('upload successful');
 });
 pipeline.on('error', (err) => {
   console.log('upload failed', err.message)
 });

In der akzeptierten Antwort endet die Funktion, bevor der Upload abgeschlossen ist, und ist daher falsch. Der folgende Code leitet einen lesbaren Stream korrekt weiter.

Referenz hochladen

async function uploadReadableStream(stream) {
  const params = {Bucket: bucket, Key: key, Body: stream};
  return s3.upload(params).promise();
}

async function upload() {
  const readable = getSomeReadableStream();
  const results = await uploadReadableStream(readable);
  console.log('upload complete', results);
}

Sie können auch noch einen Schritt weiter gehen und Fortschrittsinformationen wie folgt ausgeben ManagedUpload:

const manager = s3.upload(params);
manager.on('httpUploadProgress', (progress) => {
  console.log('progress', progress) // { loaded: 4915, total: 192915, part: 1, key: 'foo.jpg' }
});

ManagedUpload-Referenz

Eine Liste der verfügbaren Ereignisse

Typ-Skript-Lösung: In
diesem Beispiel wird Folgendes verwendet:

import * as AWS from "aws-sdk";
import * as fsExtra from "fs-extra";
import * as zlib from "zlib";
import * as stream from "stream";

Und asynchrone Funktion:

public async saveFile(filePath: string, s3Bucket: AWS.S3, key: string, bucketName: string): Promise<boolean> { 

         const uploadStream = (S3: AWS.S3, Bucket: string, Key: string) => {
            const passT = new stream.PassThrough();
            return {
              writeStream: passT,
              promise: S3.upload({ Bucket, Key, Body: passT }).promise(),
            };
          };
        const { writeStream, promise } = uploadStream(s3Bucket, bucketName, key);
        fsExtra.createReadStream(filePath).pipe(writeStream);     //  NOTE: Addition You can compress to zip by  .pipe(zlib.createGzip()).pipe(writeStream)
        let output = true;
        await promise.catch((reason)=> { output = false; console.log(reason);});
        return output;
}

Nennen Sie diese Methode irgendwo wie:

let result = await saveFileToS3(testFilePath, someS3Bucket, someKey, someBucketName);

Keine der Antworten hat bei mir funktioniert, weil ich wollte:

• Pipe in s3.upload()
• Leiten Sie das Ergebnis von s3.upload()in einen anderen Stream

Die akzeptierte Antwort macht das letztere nicht. Die anderen verlassen sich auf die Versprechen-API, deren Arbeit bei der Arbeit mit Stream-Pipes umständlich ist.

Dies ist meine Änderung der akzeptierten Antwort.

const s3 = new S3();

function writeToS3({Key, Bucket}) {
  const Body = new stream.PassThrough();

  s3.upload({
    Body,
    Key,
    Bucket: process.env.adpBucket
  })
   .on('httpUploadProgress', progress => {
       console.log('progress', progress);
   })
   .send((err, data) => {
     if (err) {
       Body.destroy(err);
     } else {
       console.log(`File uploaded and available at ${data.Location}`);
       Body.destroy();
     }
  });

  return Body;
}

const pipeline = myReadableStream.pipe(writeToS3({Key, Bucket});

pipeline.on('close', () => {
  // upload finished, do something else
})
pipeline.on('error', () => {
  // upload wasn't successful. Handle it
})

In der oben akzeptierten Antwort ist Folgendes zu beachten: Sie müssen den Durchlauf in der Funktion zurückgeben, wenn Sie Pipe wie verwenden.

fs.createReadStream(<filePath>).pipe(anyUploadFunction())

function anyUploadFunction () { 
 let pass = new stream.PassThrough();
 return pass // <- Returning this pass is important for the stream to understand where it needs to write to.
}

Andernfalls wird stillschweigend mit dem nächsten fortgefahren, ohne dass ein Fehler ausgegeben wird, oder es wird ein Fehler ausgegeben, der davon TypeError: dest.on is not a functionabhängt, wie Sie die Funktion geschrieben haben

Für diejenigen, die sich beschweren, dass die, wenn sie die s3-API-Upload-Funktion und eine Null-Byte-Datei verwenden, auf s3 (@ Radar155 und @gabo) landen - ich hatte auch dieses Problem.

Erstellen Sie einen zweiten PassThrough-Stream und leiten Sie einfach alle Daten vom ersten zum zweiten weiter und übergeben Sie den Verweis auf diese Sekunde an s3. Sie können dies auf verschiedene Arten tun - möglicherweise besteht eine schmutzige Methode darin, auf das Ereignis "Daten" im ersten Stream zu warten und dann dieselben Daten in den zweiten Stream zu schreiben - ähnlich wie für das Ereignis "Ende" - einfach aufzurufen die Endfunktion im zweiten Stream. Ich habe keine Ahnung, ob dies ein Fehler in der aws-API, der Version des Knotens oder einem anderen Problem ist - aber es hat das Problem für mich umgangen.

So könnte es aussehen:

var PassThroughStream = require('stream').PassThrough;
var srcStream = new PassThroughStream();

var rstream = fs.createReadStream('Learning/stocktest.json');
var sameStream = rstream.pipe(srcStream);
// interesting note: (srcStream == sameStream) at this point
var destStream = new PassThroughStream();
// call your s3.upload function here - passing in the destStream as the Body parameter
srcStream.on('data', function (chunk) {
    destStream.write(chunk);
});

srcStream.on('end', function () {
    dataStream.end();
});

Wenn es jemandem hilft, konnte ich erfolgreich vom Client auf s3 streamen:

https://gist.github.com/mattlockyer/532291b6194f6d9ca40cb82564db9d2a

Der serverseitige Code geht davon aus, dass reqes sich um ein Stream-Objekt handelt. In meinem Fall wurde er vom Client mit in den Headern festgelegten Dateiinformationen gesendet.

const fileUploadStream = (req, res) => {
  //get "body" args from header
  const { id, fn } = JSON.parse(req.get('body'));
  const Key = id + '/' + fn; //upload to s3 folder "id" with filename === fn
  const params = {
    Key,
    Bucket: bucketName, //set somewhere
    Body: req, //req is a stream
  };
  s3.upload(params, (err, data) => {
    if (err) {
      res.send('Error Uploading Data: ' + JSON.stringify(err) + '\n' + JSON.stringify(err.stack));
    } else {
      res.send(Key);
    }
  });
};

Ja, es verstößt gegen die Konvention, aber wenn man sich das Wesentliche ansieht, ist es viel sauberer als alles andere, was ich mit Multer, Busboy usw. gefunden habe.

+1 für Pragmatismus und danke an @SalehenRahman für seine Hilfe.

Ich verwende KnexJS und hatte ein Problem mit der Streaming-API. Ich habe es endlich behoben, hoffentlich hilft das Folgende jemandem.

const knexStream = knex.select('*').from('my_table').stream();
const passThroughStream = new stream.PassThrough();

knexStream.on('data', (chunk) => passThroughStream.write(JSON.stringify(chunk) + '\n'));
knexStream.on('end', () => passThroughStream.end());

const uploadResult = await s3
  .upload({
    Bucket: 'my-bucket',
    Key: 'stream-test.txt',
    Body: passThroughStream
  })
  .promise();

Wenn Sie die Größe des Streams kennen, können Sie den Stream mit minio-js wie folgt hochladen:

  s3Client.putObject('my-bucketname', 'my-objectname.ogg', stream, size, 'audio/ogg', function(e) {
    if (e) {
      return console.log(e)
    }
    console.log("Successfully uploaded the stream")
  })