dbgate/packages/api/src/shell/copyStream.js

const EnsureStreamHeaderStream = require('../utility/EnsureStreamHeaderStream');
const Stream = require('stream');
const ColumnMapTransformStream = require('../utility/ColumnMapTransformStream');

/**
 * Copies reader to writer. Used for import, export tables and transfer data between tables
 * @param {readerType} input - reader object
 * @param {writerType} output - writer object
 * @param {object} options - options
 * @returns {Promise}
 */
function copyStream(input, output, options) {
  const { columns } = options || {};

  const transforms = [];
  if (columns) {
    transforms.push(new ColumnMapTransformStream(columns));
  }
  if (output.requireFixedStructure) {
    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);
    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);
    // }
  });
}

module.exports = copyStream;