從儲存到串流：將 MongoDB 資料直接交付給用戶

第 1 步：MongoDB 遊標

以下是我們設定遊標的方法（重複使用您的程式碼片段）：

const cursor =
    userObject?.data?.serviceProviderName === 'ZYRO'
        ? zyroTransactionModel.find(query).cursor()
        : finoTransactionModel.find(query).cursor();

console.log("Cursor created successfully");

第 2 步：設定 ZIP 檔案
使用 yazl 庫將 CSV 資料串流傳輸到 ZIP 檔案：

const yazl = require('yazl');
const zipfile = new yazl.ZipFile();

reply.raw.writeHead(200, {
    "Content-Type": "application/zip",
    "Content-Disposition": "attachment; filename=transactions.zip",
});

zipfile.outputStream.pipe(reply.raw);

const cleanup = async () => {
    console.log("Cleaning up resources...");
    zipfile.end(); // Finalize ZIP
    await cursor.close();
};
reply.raw.on("close", cleanup);
reply.raw.on("error", cleanup);

第 3 步：建立動態 CSV 流
動態產生 CSV 資料並將其傳輸到 ZIP 檔案：

const createNewCSVStream = (headers) => {
    const csvStream = new Readable({ read() {} });
    csvStream.push(headers.join(",") + "\n"); // Add headers
    return csvStream;
};

const filteredHeaders = getHeaders(transactionDownloadFields, userObject?.state?.auth?.role);
const currentCSVStream = createNewCSVStream(filteredHeaders);

zipfile.addReadStream(currentCSVStream, "transactions_part_1.csv");

第 4 步：將 MongoDB 資料串流傳輸到 CSV
將資料從 MongoDB 直接串流傳輸到 CSV：

cursor.on('data', (doc) => {
    const csvRow = filteredHeaders.map(header => doc[header.key] || '').join(',');
    currentCSVStream.push(csvRow + '\n'); // Write row
});

cursor.on('end', () => {
    currentCSVStream.push(null); // End the stream
    zipfile.end(); // Finalize the ZIP
});

第 5 步：處理來自 MongoDB 遊標的資料
從 MongoDB 遊標串流文檔，根據需要轉換它們，並將行動態寫入 CSV 流：

try {
    for await (const doc of cursor) {
        if (clientDisconnected) {
            console.log("Client disconnected. Stopping processing...");
            break;
        }

        streamedCount++;
        rowCount++;

        let row = "";
        const filteredHeaders = getHeaders(
            transactionDownloadFields,
            userObject?.state?.auth?.role
        );

        for (let i = 0; i  f.value === field.value);

            // Get the value from the transaction document
            let value = getValueFromTransaction(doc, field.value);

            // Apply transformation if the field has a transform function
            if (originalField?.transform) {
                value = originalField.transform(value);
            }

            // Enclose the value in double quotes
            value = value !== undefined ? `"${value}"` : '"N/A"';
            row += (i > 0 ? "," : "") + value;
        }
        row += "\n";
        currentCSVStream.push(row);

        // Check if the row count has reached the threshold for the current CSV file
        if (rowCount >= MAX_ROWS_PER_FILE) {
            console.log(`Threshold reached for file ${fileIndex - 1}. Starting new file...`);
            currentCSVStream.push(null); // End the current CSV stream
            currentCSVStream = createNewCSVStream(); // Start a new stream
            rowCount = 0; // Reset the row count
        }
    }

    // Finalize the current CSV stream if it has data
    if (currentCSVStream) {
        currentCSVStream.push(null);
    }

    // Finalize the ZIP file
    zipfile.end();
    console.log(`Successfully streamed ${streamedCount} rows across ${fileIndex - 1} files.`);
} catch (error) {
    console.error("Error during processing:", error);
    if (!headersSent) reply.status(500).send({ error: "Failed to generate ZIP file" });
} finally {
    // Cleanup: Close the MongoDB cursor
    await cursor.close().catch((err) => console.error("Error closing cursor:", err));
}

總結

文件迭代使用 for wait...of:

有效率地從 MongoDB 遊標中逐一流式記錄文件。
無需將所有資料載入到記憶體即可即時處理。

• 動態 CSV 行產生：

透過迭代filteredHeaders動態建構每一行。
如果在 transactionDownloadFields 中定義，則使用轉換函數套用轉換。
行閾值與檔案分割：

根據門檻 (MAX_ROWS_PER_FILE) 監控行數。
結束目前 CSV 流，並在達到閾值時啟動新的 CSV 流。

• 錯誤處理：

如果處理過程中出現問題，則記錄並傳送錯誤回應。
透過關閉finally區塊中的MongoDB遊標來確保正確的清理。

• 完成流程：

推入 null 以終止目前 CSV 流。
處理完所有行後，完成 ZIP 檔案。

以上是從儲存到串流：將 MongoDB 資料直接交付給用戶的詳細內容。更多資訊請關注PHP中文網其他相關文章！