dennii/dbgate
1
0
Fork 0
mirror of https://github.com/DeNNiiInc/dbgate.git synced 2026-04-18 21:55:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into redis

Browse Source
This commit is contained in:
Jan Prochazka
2022-03-13 17:33:35 +01:00
parent 6138cfc2da cf734a26ee
commit 6590830344
35 changed files with 394 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
packages/api/src/shell/copyStream.js
View File

@@ -1,18 +1,47 @@
const EnsureStreamHeaderStream = require('../utility/EnsureStreamHeaderStream');
const Stream = require('stream');
const ColumnMapTransformStream = require('../utility/ColumnMapTransformStream');
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();
// }
// });
// });
function copyStream(input, output) {
return new Promise((resolve, reject) => {
const finisher = output['finisher'] || output;
finisher.on('finish', resolve);
finisher.on('error', reject);
if (output.requireFixedStructure) {
const ensureHeader = new EnsureStreamHeaderStream();
input.pipe(ensureHeader);
ensureHeader.pipe(output);
} else {
input.pipe(output);
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);
// }
});
}