獲得有關流管道的不完整資料以進行快速回應-有解無憂

需要將資料庫表轉換為 csv 報告。
如果我立即用一個查詢卸載整個平板電腦，那么應用程式會因為記憶體耗盡而崩潰。我決定以 100 行為一組從表中查詢資料，將每一行轉換為報表的一行，然后將其寫入一個通過快速回應管道傳輸的流。

這一切幾乎是這樣發生的：

資料庫查詢

const select100Users = (maxUserCreationDateStr) => {
   return db.query(`
      SELECT * FROM users WHERE created_at < to_timestamp(${maxUserCreationDateStr})
      ORDER BY created_at DESC LIMIT 100`);
}

流初始化

const { PassThrough } = require('stream');
const getUserReportStream = () => {
   const stream = new PassThrough();
   writeUserReport(stream).catch((e) => stream.emit('error', e));
   return stream;
};

用快速回應管道傳輸流

app.get('/report', (req, res) => {
   const stream = getUserReportStream();
   res.setHeader('Content-Type', 'application/vnd.ms-excel');
   res.setHeader(`Content-Disposition', 'attachment; filename="${ filename }"`);

   stream.pipe(res);
});

最后我如何將資料寫入流

const writeUserReport(stream) => {
   let maxUserCreationDateGlobal = Math.trunc(Date.now() / 1000);
   let flag = true;

   stream.write(USER_REPORT_HEADER);

   while (flag) {
      const rows100 = await select100Users(maxUserCreationDateGlobal);
      console.log(rows100.length);

      if (rows100.length === 0) {
         flag = false;
      } else {
         let maxUserCreationDate = maxUserCreationDateGlobal;

         const users100 = await Promise.all(
            rows100.map((r) => {
               const created_at = r.created_at;
               const createdAt = new Date(created_at);

               if (created_at && createdAt.toString() !== 'Invalid Date') {
                  const createdAtNumber = Math.trunc(createdAt.valueOf() / 1000);
                  maxUserCreationDate = Math.min(maxUserCreationDate, createdAtNumber);
               }

               return mapUser(r); // returns a promise
            })
         );

         users100.forEach((u) => stream.write(generateCsvRowFromUser(u)));
         maxUserCreationDateGlobal = maxUserCreationDate;

         if (rows100.length < 100) {
            flag = false;
            console.log('***');
         }
      }
   }

   console.log('end');
   stream.end();
};

結果，我在控制臺中看到了這個輸出：

100 // 100
100 // 200
100 // 300
100 // 400
100 // 500
87  // 587
***
end

但是在下載的檔案中，我得到了 401 行（第一行帶有 USER_REPORT_HEADER）。感覺就像stream.end()在從流中讀取所有值之前關閉流。

我嘗試以類似的方式使用 rxjs 中的 BehaviorSubject 而不是 PassThrough - 結果是相同的..

我怎樣才能等待從我在那里寫的所有資料的流中讀取？
或者也許有人可以推薦一種替代方法來解決這個問題。

uj5u.com熱心網友回復：

stream.write期望您將回呼作為第二個（或第三個引數）傳遞，以了解寫操作何時完成。除非之前的寫操作完成，否則不能再次呼叫 write。

所以一般來說，我建議讓整個函式異步，每次呼叫stream.write時都將它包裝成 Promise 之類的

await new Promise((resolve, reject) => stream.write(data, (error) => {
   if (error) {
      reject(error);
      return;
   }
   resolve();
});

顯然，將其提取到某種方法是有意義的。

編輯：另外，我認為這不是實際問題。我假設您的 http 連接只是在所有獲取完成之前超時，因此一旦達到超時期限，服務器最終將關閉流。

轉載請註明出處，本文鏈接：https://www.uj5u.com/caozuo/485268.html

標籤：节点.js 表示溪流

上一篇：Access-Control-Expose-Headers是如何作業的？

下一篇：在網路負載中隱藏密鑰