report wwritten rows

packages/tools/src/createBulkInsertStreamBase.ts

@@ -3,6 +3,7 @@ import _intersection from 'lodash/intersection';
 import _fromPairs from 'lodash/fromPairs';
 import { getLogger } from './getLogger';
 import { prepareTableForImport } from './tableTransforms';
+import { RowProgressReporter } from './rowProgressReporter';
 
 const logger = getLogger('bulkStreamBase');
 
@@ -21,6 +22,7 @@ export function createBulkInsertStreamBase(driver: EngineDriver, stream, dbhan,
   writable.columnNames = null;
   writable.columnDataTypes = null;
   writable.requireFixedStructure = driver.databaseEngineTypes.includes('sql');
+  writable.rowsReporter = new RowProgressReporter(options.progressName);
 
   writable.addRow = async row => {
     if (writable.structure) {
@@ -92,6 +94,7 @@ export function createBulkInsertStreamBase(driver: EngineDriver, stream, dbhan,
       // require('fs').writeFileSync('/home/jena/test.sql', dmp.s);
       // console.log(dmp.s);
       await driver.query(dbhan, dmp.s, { discardResult: true });
+      writable.rowsReporter.add(rows.length);
     } else {
       for (const row of rows) {
         const dmp = driver.createDumper();
@@ -106,6 +109,7 @@ export function createBulkInsertStreamBase(driver: EngineDriver, stream, dbhan,
         dmp.putRaw(')');
         // console.log(dmp.s);
         await driver.query(dbhan, dmp.s, { discardResult: true });
+        writable.rowsReporter.add(1);
       }
     }
     if (options.commitAfterInsert) {
@@ -129,6 +133,7 @@ export function createBulkInsertStreamBase(driver: EngineDriver, stream, dbhan,
 
   writable._final = async callback => {
     await writable.send();
+    writable.rowsReporter.finish();
     callback();
   };
 

packages/tools/src/rowProgressReporter.ts

@@ -0,0 +1,45 @@
+export class RowProgressReporter {
+  counter = 0;
+  timeoutHandle = null;
+
+  constructor(public progressName, public field = 'writtenRowCount') {}
+
+  add(count: number) {
+    this.counter += count;
+    if (!this.progressName) {
+      return;
+    }
+
+    if (this.timeoutHandle) {
+      return;
+    }
+    this.timeoutHandle = setTimeout(() => {
+      this.timeoutHandle = null;
+      this.send();
+    }, 1000);
+  }
+
+  finish() {
+    if (!this.progressName) {
+      return;
+    }
+
+    if (this.timeoutHandle) {
+      clearTimeout(this.timeoutHandle);
+      this.timeoutHandle = null;
+    }
+    this.send();
+  }
+
+  send() {
+    if (!this.progressName) {
+      return;
+    }
+
+    process.send({
+      msgtype: 'progress',
+      progressName: this.progressName,
+      [this.field]: this.counter,
+    });
+  }
+}