using stream.pipeline for better handling errors

packages/api/src/shell/copyStream.js

@@ -1,6 +1,6 @@
 const EnsureStreamHeaderStream = require('../utility/EnsureStreamHeaderStream');
-const Stream = require('stream');
 const ColumnMapTransformStream = require('../utility/ColumnMapTransformStream');
+const streamPipeline = require('../utility/streamPipeline');
 
 /**
  * Copies reader to writer. Used for import, export tables and transfer data between tables
@@ -9,7 +9,7 @@ const ColumnMapTransformStream = require('../utility/ColumnMapTransformStream');
  * @param {object} options - options
  * @returns {Promise}
  */
-function copyStream(input, output, options) {
+async function copyStream(input, output, options) {
   const { columns } = options || {};
 
   const transforms = [];
@@ -20,49 +20,19 @@ function copyStream(input, output, options) {
     transforms.push(new EnsureStreamHeaderStream());
   }
 
-  // return new Promise((resolve, reject) => {
-  //   Stream.pipeline(input, ...transforms, output, err => {
-  //     if (err) {
-  //       reject(err);
-  //     } else {
-  //       resolve();
-  //     }
-  //   });
-  // });
-
-  return new Promise((resolve, reject) => {
-    const finisher = output['finisher'] || output;
-    finisher.on('finish', resolve);
-
-    input.on('error', err => {
-      // console.log('&&&&&&&&&&&&&&&&&&&&&&& CATCH ERROR IN COPY STREAM &&&&&&&&&&&&&&&&&&&&&&');
-      // console.log(err);
-      process.send({
-        msgtype: 'copyStreamError',
-        runid: this.runid,
-        copyStreamError: err,
-      });
+  try {
+    await streamPipeline(input, transforms, output);
+  } catch (err) {
+    process.send({
+      msgtype: 'copyStreamError',
+      runid: this.runid,
+      copyStreamError: {
+        message: err.message,
+        ...err,
+      },
     });
-
-    input.on('error', reject);
-
-    finisher.on('error', reject);
-
-    let lastStream = input;
-    for (const tran of transforms) {
-      lastStream.pipe(tran);
-      lastStream = tran;
-    }
-    lastStream.pipe(output);
-
-    //  if (output.requireFixedStructure) {
-    //   const ensureHeader = new EnsureStreamHeaderStream();
-    //   input.pipe(ensureHeader);
-    //   ensureHeader.pipe(output);
-    // } else {
-    //   input.pipe(output);
-    // }
-  });
+    throw err;
+  }
 }
 
 module.exports = copyStream;

packages/api/src/shell/importDatabase.js

@@ -5,6 +5,7 @@ const { splitQueryStream } = require('dbgate-query-splitter/lib/splitQueryStream
 const download = require('./download');
 const stream = require('stream');
 const { getLogger } = require('dbgate-tools');
+const streamPipeline = require('../utility/streamPipeline');
 
 const logger = getLogger('importDb');
 
@@ -43,17 +44,6 @@ class ImportStream extends stream.Transform {
   }
 }
 
-function awaitStreamEnd(stream) {
-  return new Promise((resolve, reject) => {
-    stream.once('end', () => {
-      resolve(true);
-    });
-    stream.once('error', err => {
-      reject(err);
-    });
-  });
-}
-
 async function importDatabase({ connection = undefined, systemConnection = undefined, driver = undefined, inputFile }) {
   logger.info(`Importing database`);
 
@@ -72,9 +62,8 @@ async function importDatabase({ connection = undefined, systemConnection = undef
       returnRichInfo: true,
     });
     const importStream = new ImportStream(dbhan, driver);
-    // @ts-ignore
-    splittedStream.pipe(importStream);
-    await awaitStreamEnd(importStream);
+
+    await streamPipeline(splittedStream, importStream);
   } finally {
     if (!systemConnection) {
       await driver.close(dbhan);

packages/api/src/shell/jsonLinesReader.js

@@ -53,8 +53,7 @@ async function jsonLinesReader({ fileName, encoding = 'utf-8', limitRows = undef
   );
   const liner = byline(fileStream);
   const parser = new ParseStream({ limitRows });
-  liner.pipe(parser);
-  return parser;
+  return [liner, parser];
 }
 
 module.exports = jsonLinesReader;

packages/api/src/shell/jsonReader.js

@@ -10,7 +10,6 @@ const download = require('./download');
 
 const logger = getLogger('jsonReader');
 
-
 class ParseStream extends stream.Transform {
   constructor({ limitRows, jsonStyle, keyField }) {
     super({ objectMode: true });
@@ -72,8 +71,12 @@ async function jsonReader({
     // @ts-ignore
     encoding
   );
+
   const parseJsonStream = parser();
-  fileStream.pipe(parseJsonStream);
+
+  const resultPipe = [fileStream, parseJsonStream];
+
+  // fileStream.pipe(parseJsonStream);
 
   const parseStream = new ParseStream({ limitRows, jsonStyle, keyField });
 
@@ -81,15 +84,20 @@ async function jsonReader({
 
   if (rootField) {
     const filterStream = pick({ filter: rootField });
-    parseJsonStream.pipe(filterStream);
-    filterStream.pipe(tramsformer);
-  } else {
-    parseJsonStream.pipe(tramsformer);
+    resultPipe.push(filterStream);
+    // parseJsonStream.pipe(filterStream);
+    // filterStream.pipe(tramsformer);
   }
+  // else {
+  //   parseJsonStream.pipe(tramsformer);
+  // }
 
-  tramsformer.pipe(parseStream);
+  resultPipe.push(tramsformer);
+  resultPipe.push(parseStream);
 
-  return parseStream;
+  // tramsformer.pipe(parseStream);
+
+  return resultPipe;
 }
 
 module.exports = jsonReader;

packages/api/src/shell/jsonWriter.js

@@ -99,9 +99,10 @@ async function jsonWriter({ fileName, jsonStyle, keyField = '_key', rootField, e
   logger.info(`Writing file ${fileName}`);
   const stringify = new StringifyStream({ jsonStyle, keyField, rootField });
   const fileStream = fs.createWriteStream(fileName, encoding);
-  stringify.pipe(fileStream);
-  stringify['finisher'] = fileStream;
-  return stringify;
+  return [stringify, fileStream];
+  // stringify.pipe(fileStream);
+  // stringify['finisher'] = fileStream;
+  // return stringify;
 }
 
 module.exports = jsonWriter;

packages/api/src/shell/modifyJsonLinesReader.js

@@ -141,8 +141,9 @@ async function modifyJsonLinesReader({
   );
   const liner = byline(fileStream);
   const parser = new ParseStream({ limitRows, changeSet, mergedRows, mergeKey, mergeMode });
-  liner.pipe(parser);
-  return parser;
+  return [liner, parser];
+  // liner.pipe(parser);
+  // return parser;
 }
 
 module.exports = modifyJsonLinesReader;

packages/api/src/shell/sqlDataWriter.js

@@ -44,9 +44,10 @@ async function sqlDataWriter({ fileName, dataName, driver, encoding = 'utf-8' })
   logger.info(`Writing file ${fileName}`);
   const stringify = new SqlizeStream({ fileName, dataName });
   const fileStream = fs.createWriteStream(fileName, encoding);
-  stringify.pipe(fileStream);
-  stringify['finisher'] = fileStream;
-  return stringify;
+  return [stringify, fileStream];
+  // stringify.pipe(fileStream);
+  // stringify['finisher'] = fileStream;
+  // return stringify;
 }
 
 module.exports = sqlDataWriter;

packages/api/src/utility/streamPipeline.js

@@ -0,0 +1,18 @@
+const stream = require('stream');
+const _ = require('lodash');
+
+function streamPipeline(...processedStreams) {
+  const streams = _.flattenDeep(processedStreams);
+  return new Promise((resolve, reject) => {
+    // @ts-ignore
+    stream.pipeline(...streams, err => {
+      if (err) {
+        reject(err);
+      } else {
+        resolve();
+      }
+    });
+  });
+}
+
+module.exports = streamPipeline;

packages/types/engines.d.ts

@@ -144,6 +144,8 @@ export interface DatabaseHandle<TClient = any> {
   treeKeySeparator?: string;
 }
 
+export type StreamResult = stream.Readable | (stream.Readable | stream.Writable)[];
+
 export interface EngineDriver<TClient = any> extends FilterBehaviourProvider {
   engine: string;
   title: string;
@@ -191,15 +193,11 @@ export interface EngineDriver<TClient = any> extends FilterBehaviourProvider {
   close(dbhan: DatabaseHandle<TClient>): Promise<any>;
   query(dbhan: DatabaseHandle<TClient>, sql: string, options?: QueryOptions): Promise<QueryResult>;
   stream(dbhan: DatabaseHandle<TClient>, sql: string, options: StreamOptions);
-  readQuery(dbhan: DatabaseHandle<TClient>, sql: string, structure?: TableInfo): Promise<stream.Readable>;
-  readJsonQuery(dbhan: DatabaseHandle<TClient>, query: any, structure?: TableInfo): Promise<stream.Readable>;
+  readQuery(dbhan: DatabaseHandle<TClient>, sql: string, structure?: TableInfo): Promise<StreamResult>;
+  readJsonQuery(dbhan: DatabaseHandle<TClient>, query: any, structure?: TableInfo): Promise<StreamResult>;
   // eg. PostgreSQL COPY FROM stdin
-  writeQueryFromStream(dbhan: DatabaseHandle<TClient>, sql: string): Promise<stream.Writable>;
-  writeTable(
-    dbhan: DatabaseHandle<TClient>,
-    name: NamedObjectInfo,
-    options: WriteTableOptions
-  ): Promise<stream.Writable>;
+  writeQueryFromStream(dbhan: DatabaseHandle<TClient>, sql: string): Promise<StreamResult>;
+  writeTable(dbhan: DatabaseHandle<TClient>, name: NamedObjectInfo, options: WriteTableOptions): Promise<StreamResult>;
   analyseSingleObject(
     dbhan: DatabaseHandle<TClient>,
     name: NamedObjectInfo,

plugins/dbgate-plugin-csv/src/backend/reader.js

@@ -95,9 +95,10 @@ async function reader({ fileName, encoding = 'utf-8', header = true, delimiter,
   });
   const fileStream = fs.createReadStream(downloadedFile, encoding);
   const csvPrepare = new CsvPrepareStream({ header });
-  fileStream.pipe(csvStream);
-  csvStream.pipe(csvPrepare);
-  return csvPrepare;
+  return [fileStream, csvStream, csvPrepare];
+  // fileStream.pipe(csvStream);
+  // csvStream.pipe(csvPrepare);
+  // return csvPrepare;
 }
 
 reader.initialize = (dbgateEnv) => {

plugins/dbgate-plugin-csv/src/backend/writer.js

@@ -31,11 +31,13 @@ async function writer({ fileName, encoding = 'utf-8', header = true, delimiter,
   const csvPrepare = new CsvPrepareStream({ header });
   const csvStream = csv.stringify({ delimiter, quoted });
   const fileStream = fs.createWriteStream(fileName, encoding);
-  csvPrepare.pipe(csvStream);
-  csvStream.pipe(fileStream);
-  csvPrepare['finisher'] = fileStream;
+  // csvPrepare.pipe(csvStream);
+  // csvStream.pipe(fileStream);
+  // csvPrepare['finisher'] = fileStream;
   csvPrepare.requireFixedStructure = true;
-  return csvPrepare;
+
+  return [csvPrepare, csvStream, fileStream];
+  // return csvPrepare;
 }
 
 module.exports = writer;

plugins/dbgate-plugin-xml/src/backend/reader.js

@@ -63,8 +63,10 @@ async function reader({ fileName, encoding = 'utf-8', itemElementName }) {
 
   const fileStream = fs.createReadStream(fileName, encoding);
   const parser = new ParseStream({ itemElementName });
-  fileStream.pipe(parser);
-  return parser;
+
+  return [fileStream, parser];
+  // fileStream.pipe(parser);
+  // return parser;
 }
 
 module.exports = reader;

plugins/dbgate-plugin-xml/src/backend/writer.js

@@ -73,9 +73,10 @@ async function writer({ fileName, encoding = 'utf-8', itemElementName, rootEleme
   logger.info(`Writing file ${fileName}`);
   const stringify = new StringifyStream({ itemElementName, rootElementName });
   const fileStream = fs.createWriteStream(fileName, encoding);
-  stringify.pipe(fileStream);
-  stringify['finisher'] = fileStream;
-  return stringify;
+  return [stringify, fileStream];
+  // stringify.pipe(fileStream);
+  // stringify['finisher'] = fileStream;
+  // return stringify;
 }
 
 module.exports = writer;