feat: Added streaming for fetching and aggregating MongoDB data for batch insertion.

Author: WillianAgostini

src/cluster.js

@@ -26,7 +26,6 @@ function initializeCluster({ backgroundTaskFile, clusterSize, onMessage }) {
         })
 
         child.on('message', (message) => {
-            if (message !== 'item-done') return
             onMessage(message)
         })
 

src/data-streaming.js

@@ -0,0 +1,12 @@
+import { StreamCache } from './stram-cache.js'
+import { getMongoConnection } from './db.js'
+
+const ITEMS_PER_PAGE = 4000
+
+const mongoDB = await getMongoConnection()
+const stream = mongoDB.students.find().stream()
+const cache = new StreamCache(stream, ITEMS_PER_PAGE)
+
+cache.stream().on('data', (data) => {
+    process.send(JSON.parse(data));
+});

src/index.js

@@ -1,27 +1,16 @@
 import { initialize } from "./cluster.js"
 import { getMongoConnection, getPostgresConnection } from './db.js'
 import cliProgress from 'cli-progress'
-import { setTimeout } from 'node:timers/promises'
 const mongoDB = await getMongoConnection()
 const postgresDB = await getPostgresConnection()
-const ITEMS_PER_PAGE = 4000
-const CLUSTER_SIZE = 99
+// const ITEMS_PER_PAGE = 4000
+const CLUSTER_SIZE = 8
 const TASK_FILE = new URL('./background-task.js', import.meta.url).pathname
+const DATA_STREAMING_FILE = new URL('./data-streaming.js', import.meta.url).pathname
 
 // console.log(`there was ${await postgresDB.students.count()} items on Postgres, deleting all...`)
 await postgresDB.students.deleteAll()
 
-async function* getAllPagedData(itemsPerPage, page = 0) {
-
-    const data = mongoDB.students.find().skip(page).limit(itemsPerPage)
-    const items = await data.toArray()
-    if (!items.length) return
-
-    yield items
-
-    yield* getAllPagedData(itemsPerPage, page += itemsPerPage)
-}
-
 const total = await mongoDB.students.countDocuments()
 // console.log(`total items on DB: ${total}`)
 
@@ -37,25 +26,29 @@ const cp = initialize(
         backgroundTaskFile: TASK_FILE,
         clusterSize: CLUSTER_SIZE,
         amountToBeProcessed: total,
-        async onMessage(message) {
-            progress.increment()
+        async onMessage(cumulativeProcessed) {
+            totalProcessed += cumulativeProcessed;
+            progress.update(totalProcessed);
 
-            if (++totalProcessed !== total) return
+            if (totalProcessed !== total) return
             // console.log(`all ${amountToBeProcessed} processed! Exiting...`)
             progress.stop()
             cp.killAll()
 
-            const insertedOnSQLite = await postgresDB.students.count()
-            console.log(`total on MongoDB ${total} and total on PostGres ${insertedOnSQLite}`)
-            console.log(`are the same? ${total === insertedOnSQLite ? 'yes' : 'no'}`)
+            const insertedOnSQLPostGres = await postgresDB.students.count()
+            console.log(`total on MongoDB ${total} and total on PostGres ${insertedOnSQLPostGres}`)
+            console.log(`are the same? ${total === insertedOnSQLPostGres ? 'yes' : 'no'}`)
             process.exit()
-
         }
     }
 )
-await setTimeout(1000)
-
-for await (const data of getAllPagedData(ITEMS_PER_PAGE)) {
-    cp.sendToChild(data)
-}
 
+initialize(
+    {
+        backgroundTaskFile: DATA_STREAMING_FILE,
+        clusterSize: 1,
+        async onMessage(message) {
+            cp.sendToChild(message)
+        }
+    }
+)

src/stram-cache.js

@@ -0,0 +1,33 @@
+import { Readable } from 'stream';
+
+export class StreamCache {
+    constructor(inputStream, cacheThreshold = 4000) {
+        this.cacheStream = new Readable({
+            read() { }
+        });
+        this.cache = [];
+        this.cacheThreshold = cacheThreshold;
+        inputStream.on('data', this._addDataToCache);
+        inputStream.on('end', () => {
+            this._emitCache();
+            this.cacheStream.emit('end');
+        });
+    }
+
+    _addDataToCache = (data) => {
+        this.cache.push(data);
+        if (this.cache.length >= this.cacheThreshold) {
+            this._emitCache();
+        }
+    }
+
+    _emitCache() {
+        if (!this.cache.length) return
+        this.cacheStream.push(JSON.stringify(this.cache));
+        this.cache = [];
+    }
+
+    stream() {
+        return this.cacheStream;
+    }
+}