Promise kombinieren Stream

Question

Ich möchte große Datei Zeile für Zeile lesen und fügen Sie die Daten in DB speichern. Mein functon gibt ein Promise zurück, in dem es erstellt wurde, und löste es, wenn Ereignis aufgerufen wurde, aber ich möchte es nicht wirklich, weil es in stream.on('data')Promise.map() auf jeder Zeile produziert, und ich möchte sicher sein, dass alle Einfügeoperationen abgeschlossen sind vor resolve() aufgerufen. Wie kann ich in diesem Fall die richtige Kette herstellen?

var loadFromFile = (options) => new Promise(function (resolve, reject) { 
     let stream = fs.createReadStream(options.filePath, { 
      flags: 'r', 
     }); 

     stream.on('data', (chunk) => { 
      /* process data chunk to string Array */ 
      Promise.map(lines, (line) => { 
      /* process and insert each line here */ 
      }) 
      .catch((err)=>{ 
       reject(err); 
      }); 
      }); 

      stream.on('end',() => { 
       if (someBuisnessLogicCheckHere) { 
       reject('Invalid data was found in file'); 
       } 
       resolve(); // Here I am not sure, that all inserts for each chunk in Promise.map was completed 
      }); 
    }); 

Answer 1

Wenn Sie sicher sein wollen Sie für das Versprechen nicht, bis alle der Versprechungen in der Mapping-Operation zu beheben haben beschlossen, warten, dass Promise.map zurückkehrt; siehe Kommentar:

var loadFromFile = (options) => new Promise(function(resolve, reject) { 
    let stream = fs.createReadStream(options.filePath, { 
     flags: 'r', 
    }); 
    let promises = [];         // Array of promises we'll wait for 

    stream.on('data', (chunk) => { 
     promises.push(        // Remember this promise 
      Promise.map(lines, (line) => { 
       /* process and insert each line here */ 
      }) 
      .catch((err) => { 
       reject(err); 
      }) 
     ); 
    }); 

    stream.on('end',() => { 
     if (someBuisnessLogicCheckHere) { 
      reject('Invalid data was found in file'); 
     } 
     Promise.all(promises).then(() => { resolve()}); // Wait for it before resolving 
    }); 
}); 

Bitte beachte, dass ich nicht nur Promise.all(promises).then(resolve); tun, denn das wird eine Reihe von Auflösungen resolve geben, die Ihr ursprünglicher Code nicht mit dem Anrufer nicht geteilt.

Wenn es eine Möglichkeit gibt, dass das Array ziemlich groß wird, während es mit den meisten bereits aufgelösten Versprechungen gefüllt wird, könnten Sie sie proaktiv entfernen, wenn sie auflösen, und nur auf diejenigen warten, die am Ende übrig sind.

Answer 2

Wenn Sie die Datei mit automatisierten Mechanismen der Pause/Fortsetzen in der Pipe-Methode verarbeiten müssen, sehen Sie sich scramjet.

Ihr Beispiel würde viel klarer sehen:

var loadFromFile = (options) => 
    fs.createReadStream(options.filePath, { 
     flags: 'r', 
    }) 
    .split(/\r?\n/) 
     // split by line by regex 
    .parse((line) => aPromiseReturningFunction(line)) 
     // will wait until promises are resolved 
    .map((data) => doSomeAsyncAndReturnPromise(data))  
     // will wait as above 
    .accumulate((acc, resolved) => acc.push(resolved), []) 
    .then((arr) => { 
     if (someBuisnessLogicCheckHere) { 
       return Promise.reject('Invalid data was found in file'); 
     } 
     // Here all your items are saved (i.e. all Promises resolved) 
    }); 
});