From dc7c44b797803e861809fbe1ab18e9e7fd7e5aeb Mon Sep 17 00:00:00 2001
From: Jan Prochazka <jan.prochazka@gmail.com>
Date: Thu, 11 Jun 2020 10:19:38 +0200
Subject: [PATCH] csvReader follows dbgate stream api

---
 packages/api/src/shell/csvReader.js | 47 ++++++++++++++++++++++++++---
 test/importTable.js                 | 24 +++++++--------
 2 files changed, 55 insertions(+), 16 deletions(-)

diff --git a/packages/api/src/shell/csvReader.js b/packages/api/src/shell/csvReader.js
index 204221e2..68e9bd9b 100644
--- a/packages/api/src/shell/csvReader.js
+++ b/packages/api/src/shell/csvReader.js
@@ -1,15 +1,54 @@
+const _ = require('lodash');
 const csv = require('csv');
 const fs = require('fs');
+const stream = require('stream');
 
-async function csvReader({ fileName, encoding = 'utf-8', ...options }) {
+class CsvPrepareStream extends stream.Transform {
+  constructor({ header }) {
+    super({ objectMode: true });
+    this.structure = null;
+    this.header = header;
+  }
+  _transform(chunk, encoding, done) {
+    if (this.structure) {
+      this.push(
+        _.zipObject(
+          this.structure.columns.map((x) => x.columnName),
+          chunk
+        )
+      );
+      done();
+    } else {
+      if (this.header) {
+        this.structure = { columns: chunk.map((columnName) => ({ columnName })) };
+        this.push(this.structure);
+      } else {
+        this.structure = { columns: chunk.map((value, index) => ({ columnName: `col${index + 1}` })) };
+        this.push(this.structure);
+        this.push(
+          _.zipObject(
+            this.structure.columns.map((x) => x.columnName),
+            chunk
+          )
+        );
+      }
+      done();
+    }
+  }
+}
+
+async function csvReader({ fileName, encoding = 'utf-8', header = true, delimiter, quoted }) {
   console.log(`Reading file ${fileName}`);
   const csvStream = csv.parse({
-    columns: true,
-    ...options,
+    // @ts-ignore
+    delimiter,
+    quoted,
   });
   const fileStream = fs.createReadStream(fileName, encoding);
+  const csvPrepare = new CsvPrepareStream({ header });
   fileStream.pipe(csvStream);
-  return csvStream;
+  csvStream.pipe(csvPrepare);
+  return csvPrepare;
 }
 
 module.exports = csvReader;
diff --git a/test/importTable.js b/test/importTable.js
index 524a9677..010ffa8b 100644
--- a/test/importTable.js
+++ b/test/importTable.js
@@ -3,21 +3,21 @@ const dbgateApi = require('@dbgate/api');
 async function run() {
   const csvReader = await dbgateApi.csvReader({
     fileName: 'test.csv',
-    header: true,
+    // header: false,
   });
 
 
-  const tableWriter = await dbgateApi.tableWriter({
-    connection: {
-      server: 'localhost',
-      engine: 'mysql',
-      user: 'root',
-      password: 'test',
-      port: '3307',
-      database: 'Chinook',
-    },
-    pureName: 'importedTable'
-  });
+  // const tableWriter = await dbgateApi.tableWriter({
+  //   connection: {
+  //     server: 'localhost',
+  //     engine: 'mysql',
+  //     user: 'root',
+  //     password: 'test',
+  //     port: '3307',
+  //     database: 'Chinook',
+  //   },
+  //   pureName: 'importedTable'
+  // });
 
   const consoleWriter = await dbgateApi.consoleObjectWriter();