query - result tabs

packages/api/src/controllers/connections.js

@@ -1,10 +1,9 @@
-
 const path = require('path');
 const { fork } = require('child_process');
 const _ = require('lodash');
 const nedb = require('nedb-promises');
 
-const datadir = require('../utility/datadir');
+const { datadir } = require('../utility/directories');
 const socket = require('../utility/socket');
 
 module.exports = {
@@ -28,7 +27,7 @@ module.exports = {
   },
   test(req, res) {
     const subprocess = fork(process.argv[1], ['connectProcess']);
-    subprocess.on('message', resp => res.json(resp));
+    subprocess.on('message', (resp) => res.json(resp));
     subprocess.send(req.body);
   },
 

packages/api/src/controllers/jsldata.js

@@ -0,0 +1,77 @@
+const _ = require('lodash');
+const path = require('path');
+const fs = require('fs');
+const lineReader = require('line-reader');
+const { jsldir } = require('../utility/directories');
+
+module.exports = {
+  openedReaders: {},
+
+  closeReader(jslid) {
+    if (!this.openedReaders[jslid]) return Promise.resolve();
+    return new Promise((resolve, reject) => {
+      this.openedReaders[jslid].reader.close((err) => {
+        if (err) reject(err);
+        resolve();
+        delete this.openedReaders[jslid];
+      });
+    });
+  },
+
+  readLine(jslid) {
+    if (!this.openedReaders[jslid]) return Promise.reject();
+    return new Promise((resolve, reject) => {
+      const { reader } = this.openedReaders[jslid];
+      if (!reader.hasNextLine()) return Promise.resolve(null);
+      reader.nextLine((err, line) => {
+        this.openedReaders[jslid].readedCount += 1;
+        if (err) reject(err);
+        resolve(line);
+      });
+    });
+  },
+
+  openReader(jslid) {
+    const file = path.join(jsldir(), `${jslid}.jsonl`);
+    return new Promise((resolve, reject) =>
+      lineReader.open(file, function (err, reader) {
+        if (err) reject(err);
+        resolve();
+        this.openedReaders[jslid] = {
+          reader,
+          readedCount: 0,
+        };
+      })
+    );
+  },
+
+  async ensureReader(jslid, offset) {
+    if (this.openedReaders[jslid] && this.openedReaders[jslid].readedCount > offset) {
+      await this.closeReader();
+    }
+    if (!this.openedReaders[jslid]) {
+      await this.openReader();
+    }
+    while (this.openedReaders[jslid].readedCount < offset) {
+      await this.readLine(jslid);
+    }
+  },
+
+  getInfo_meta: 'get',
+  getInfo(jslid) {
+    const file = path.join(jsldir(), `${jslid}.jsonl.info`);
+    return JSON.parse(fs.readFileSync(file, 'utf-8'));
+  },
+
+  getRows_meta: 'get',
+  async getRows(jslid, offset, limit) {
+    await this.ensureReader(jslid, offset);
+    const res = [];
+    for (let i = 0; i < limit; i += 1) {
+      const line = await this.readLine(jslid);
+      if (line == null) break;
+      res.push(JSON.parse(line));
+    }
+    return res;
+  },
+};

packages/api/src/controllers/sessions.js

@@ -24,6 +24,15 @@ module.exports = {
     socket.emit(`session-info-${sesid}`, info);
   },
 
+  handle_done(sesid) {
+    socket.emit(`session-done-${sesid}`);
+  },
+
+  handle_recordset(sesid, props) {
+    const { jslid } = props;
+    socket.emit(`session-recordset-${sesid}`, { jslid });
+  },
+
   create_meta: 'post',
   async create({ conid, database }) {
     const sesid = uuidv1();

packages/api/src/proc/sessionProcess.js

@@ -1,10 +1,55 @@
 const engines = require('@dbgate/engines');
+const uuidv1 = require('uuid/v1');
+const path = require('path');
+const fs = require('fs');
+
 const driverConnect = require('../utility/driverConnect');
+const { jsldir } = require('../utility/directories');
 
 let systemConnection;
 let storedConnection;
 let afterConnectCallbacks = [];
 
+class StreamHandler {
+  constructor() {
+    this.recordset = this.recordset.bind(this);
+    this.row = this.row.bind(this);
+    this.error = this.error.bind(this);
+    this.done = this.done.bind(this);
+    this.info = this.info.bind(this);
+  }
+
+  closeCurrentStream() {
+    if (this.currentStream) {
+      this.currentStream.end();
+      this.currentStream = null;
+    }
+  }
+
+  recordset(columns) {
+    this.closeCurrentStream();
+    this.jslid = uuidv1();
+    this.currentFile = path.join(jsldir(), `${this.jslid}.jsonl`);
+    this.currentStream = fs.createWriteStream(this.currentFile);
+    fs.writeFileSync(`${this.currentFile}.info`, JSON.stringify(columns));
+    process.send({ msgtype: 'recordset', jslid: this.jslid });
+  }
+  row(row) {
+    // console.log('ACCEPT ROW', row);
+    this.currentStream.write(JSON.stringify(row) + '\n');
+  }
+  error(error) {
+    process.send({ msgtype: 'error', error });
+  }
+  done(result) {
+    this.closeCurrentStream();
+    process.send({ msgtype: 'done', result });
+  }
+  info(info) {
+    process.send({ msgtype: 'info', info });
+  }
+}
+
 async function handleConnect(connection) {
   storedConnection = connection;
 
@@ -27,23 +72,8 @@ async function handleExecuteQuery({ sql }) {
   await waitConnected();
   const driver = engines(storedConnection);
 
-  await driver.stream(systemConnection, sql, {
-    recordset: (columns) => {
-      process.send({ msgtype: 'recordset', columns });
-    },
-    row: (row) => {
-      process.send({ msgtype: 'row', row });
-    },
-    error: (error) => {
-      process.send({ msgtype: 'error', error });
-    },
-    done: (result) => {
-      process.send({ msgtype: 'done', result });
-    },
-    info: (info) => {
-      process.send({ msgtype: 'info', info });
-    },
-  });
+  const handler = new StreamHandler();
+  await driver.stream(systemConnection, sql, handler);
 }
 
 const messageHandlers = {

packages/api/src/utility/datadir.js

@@ -1,18 +0,0 @@
-const os = require('os');
-const path = require('path');
-const fs = require('fs');
-
-let created = false;
-
-module.exports = function datadir() {
-  const dir = path.join(os.homedir(), 'dbgate-data');
-  if (!created) {
-    if (!fs.existsSync(dir)) {
-      console.log(`Creating data directory ${dir}`)
-      fs.mkdirSync(dir);
-    }
-    created = true;
-  }
-
-  return dir;
-};

packages/api/src/utility/directories.js

@@ -0,0 +1,37 @@
+const os = require('os');
+const path = require('path');
+const fs = require('fs');
+
+let createdDatadir = false;
+let createdJsldir = false;
+
+function datadir() {
+  const dir = path.join(os.homedir(), 'dbgate-data');
+  if (!createdDatadir) {
+    if (!fs.existsSync(dir)) {
+      console.log(`Creating data directory ${dir}`);
+      fs.mkdirSync(dir);
+    }
+    createdDatadir = true;
+  }
+
+  return dir;
+}
+
+function jsldir() {
+  const dir = path.join(datadir(), 'jsl');
+  if (!createdJsldir) {
+    if (!fs.existsSync(dir)) {
+      console.log(`Creating jsl directory ${dir}`);
+      fs.mkdirSync(dir);
+    }
+    createdJsldir = true;
+  }
+
+  return dir;
+}
+
+module.exports = {
+  datadir,
+  jsldir,
+};