pg stream query support

packages/engines/postgres/index.js

@@ -1,10 +1,11 @@
+const _ = require('lodash');
 const PostgreAnalyser = require('./PostgreAnalyser');
 const PostgreDumper = require('./PostgreDumper');
 
 /** @type {import('@dbgate/types').SqlDialect} */
 const dialect = {
   rangeSelect: true,
-  stringEscapeChar: "\\",
+  stringEscapeChar: '\\',
   quoteIdentifier(s) {
     return '"' + s + '"';
   },
@@ -13,7 +14,13 @@ const dialect = {
 /** @type {import('@dbgate/types').EngineDriver} */
 const driver = {
   async connect(nativeModules, { server, port, user, password, database }) {
-    const client = new nativeModules.pg.Client({ host: server, port, user, password, database: database || 'postgres' });
+    const client = new nativeModules.pg.Client({
+      host: server,
+      port,
+      user,
+      password,
+      database: database || 'postgres',
+    });
     await client.connect();
     client._nativeModules = nativeModules;
     return client;
@@ -22,6 +29,67 @@ const driver = {
     const res = await client.query(sql);
     return { rows: res.rows, columns: res.fields };
   },
+  async stream(client, sql, options) {
+    const query = new client._nativeModules.pgQueryStream(sql);
+    const stream = client.query(query);
+
+    // const handleInfo = (info) => {
+    //   const { message, lineNumber, procName } = info;
+    //   options.info({
+    //     message,
+    //     line: lineNumber,
+    //     procedure: procName,
+    //     time: new Date(),
+    //     severity: 'info',
+    //   });
+    // };
+
+    let wasHeader = false;
+
+    const handleEnd = (result) => {
+      // console.log('RESULT', result);
+      options.done(result);
+    };
+
+    const handleReadable = () => {
+      let row = stream.read();
+      if (!wasHeader && row) {
+        options.recordset(_.keys(row).map((columnName) => ({ columnName })));
+        wasHeader = true;
+      }
+
+      while (row) {
+        options.row(row);
+        row = stream.read();
+      }
+    };
+
+    // const handleFields = (columns) => {
+    //   // console.log('FIELDS', columns[0].name);
+    //   options.recordset(columns);
+    //   // options.recordset(extractColumns(columns));
+    // };
+
+    const handleError = (error) => {
+      console.log('ERROR', error);
+      const { message, lineNumber, procName } = error;
+      options.info({
+        message,
+        line: lineNumber,
+        procedure: procName,
+        time: new Date(),
+        severity: 'error',
+      });
+    };
+
+    stream.on('error', handleError);
+    stream.on('readable', handleReadable);
+    // stream.on('result', handleRow)
+    // stream.on('data', handleRow)
+    stream.on('end', handleEnd);
+
+    return stream;
+  },
   async getVersion(client) {
     const { rows } = await this.query(client, 'SELECT version()');
     const { version } = rows[0];