From f11e44e7922bc879858acef0adf9111e3acf7996 Mon Sep 17 00:00:00 2001 From: coderaiser Date: Fri, 30 May 2014 09:35:57 -0400 Subject: [PATCH] feature(pipe) create: params -> read, write, options, callback --- lib/server/files.js | 34 +++++++------- lib/server/main.js | 18 ++++---- lib/server/pipe.js | 93 +++++++++++++++++++++------------------ lib/server/rest.js | 73 +++++++++++++++--------------- lib/server/rest/fs/get.js | 18 +++----- lib/server/rest/fs/put.js | 26 +++++------ 6 files changed, 132 insertions(+), 130 deletions(-) diff --git a/lib/server/files.js b/lib/server/files.js index df94dde1..5fbe738b 100644 --- a/lib/server/files.js +++ b/lib/server/files.js @@ -41,7 +41,11 @@ function readPipe(params) { var name, names, lenght, - p = params; + p = params, + options = { + gzip : p.gzip, + notEnd : true + }; if (p.names) { names = p.names.slice(); @@ -54,23 +58,17 @@ } else { name = names.shift(); - pipe.create({ - from : name, - write : p.write, - gzip : p.gzip, - notEnd : true, - callback: function(error) { - if (error) - Util.exec(p.callback, error); - else - readPipe({ - dir : p.dir, - names : names, - write : p.write, - gzip : p.gzip, - callback : p.callback - }); - } + pipe.create(name, p.write, options, function(error) { + if (error) + Util.exec(p.callback, error); + else + readPipe({ + dir : p.dir, + names : names, + write : p.write, + gzip : p.gzip, + callback : p.callback + }); }); } } diff --git a/lib/server/main.js b/lib/server/main.js index 9856293a..22ac684a 100644 --- a/lib/server/main.js +++ b/lib/server/main.js @@ -286,7 +286,8 @@ isGzip = isGZIP(p.request) && p.gzip; fs.lstat(p.name, function(error, stat) { - var time, length, range; + var time, length, range, + options = {}; if (error) { sendError(params, error); @@ -308,15 +309,14 @@ mainSetHeader(params); - pipe.create({ - from : p.name, - write : p.response, + options = { gzip : isGzip && !range, - range : range, - callback: function(error) { - if (error) - sendError(params, error); - } + range : range + }; + + pipe.create(p.name, p.response, options, function(error) { + if (error) + sendError(params, error); }); } }); diff --git a/lib/server/pipe.js b/lib/server/pipe.js index 57bfbed4..e83934cc 100644 --- a/lib/server/pipe.js +++ b/lib/server/pipe.js @@ -21,53 +21,62 @@ * @param params * @param callback */ - function create(params) { - var gzip, read, write, isFsWrite, - p = params, - func = p.callback, - options = { + function create(read, write, options, callback) { + var gzip, isFsWrite, func, + isStrRead = Util.isString(read), + isStrWrite = Util.isString(write), + isFunc = Util.isFunction(options), + o = {}, + optionsRead = { bufferSize: 4 * 1024 }; - if (p) { - if (p.range) - options = { - start : p.range.start, - end : p.range.end, - }; - - read = p.read || fs.createReadStream(p.from, options); - - if (p.write) - write = p.write; - else { - write = fs.createWriteStream(p.to); - isFsWrite = true; - } - - if (p.gzip || p.gunzip) { - if (p.gzip) - gzip = zlib.createGzip(); - else - gzip = zlib.createGunzip(); - - on('error', read, func); - read = read.pipe(gzip); - } - - on('error', write, func); - on('error', read, func); - - Util.exec.if(!isFsWrite, function() { - read.pipe(write, { - end: !p.notEnd - }); - - on('end', read, func); - }, function(callback) { - on('open', write, callback); + Util.checkArgs(arguments, ['read', 'write', 'options']); + + if (isFunc) + callback = options; + else + o = options; + + func = Util.exec.ret(callback); + + if (options.range) + Util.extend(optionsRead, { + start : o.range.start, + end : o.range.end, }); + + + if (isStrRead) + read = fs.createReadStream(read, optionsRead); + + if (isStrWrite) { + write = fs.createWriteStream(write); + isFsWrite = true; } + + if (o.gzip || o.gunzip) { + if (o.gzip) + gzip = zlib.createGzip(); + else + gzip = zlib.createGunzip(); + + on('error', read, func); + read = read.pipe(gzip); + } + + on('error', write, func); + on('error', read, func); + + Util.exec.if(!isFsWrite, function() { + read.pipe(write, { + end: !o.notEnd + }); + + on('end', read, func); + }, function(callback) { + on('open', write, callback); + }); } function on(event, emitter, callback) { diff --git a/lib/server/rest.js b/lib/server/rest.js index 33623cb4..b9015e84 100644 --- a/lib/server/rest.js +++ b/lib/server/rest.js @@ -40,13 +40,8 @@ NOT_LOG = true, fse = { - copy : ncp || function(from, to, callback) { - pipe.create({ - from : from, - to : to, - callback : callback - }); - }, + copy : ncp || pipe.create, + delete : rimraf || function(path, callback) { dir.isDir(path, function(error, isDir) { if (error) @@ -278,7 +273,7 @@ * @param pParams {command, method, body, requrest, response} */ function onPUT(params) { - var p, cmd, files, name, json, config, data, + var p, cmd, files, name, json, config, data, read, write, options, ret = main.checkParams(params, ['body']); if (ret) { @@ -344,42 +339,46 @@ break; case 'zip': - if (!Util.checkObjTrue(files, ['from'])) + if (!files.from) { sendError(params, p.data); - else - pipe.create({ - from : files.from, - to : files.to || files.from + '.zip', - gzip : true, - callback : function(error) { - var name = path.basename(files.from); - - if (error) - sendError(params, error); - else - sendMsg(params, 'zip', name); - } - }); + } else { + read = files.from, + write = files.to || files.from + '.zip'; + options = { + gzip: true + }; + + pipe.create(read, write, options, function(error) { + var name = path.basename(files.from); + + if (error) + sendError(params, error); + else + sendMsg(params, 'zip', name); + }); + } break; case 'unzip': - if (!Util.checkObjTrue(files, ['from'])) + if (!files.from) { sendError(params, p.data); - else - pipe.create({ - from : files.from, - to : files.to || Util.rmStrOnce(files.from, ['.zip', '.gzip']), - gunzip : true, - callback : function(error) { - var name = path.basename(files.from); - - if (error) - sendError(params, error); - else - sendMsg(params, 'unzip', name); - } + } else { + read = files.from; + write = files.to || Util.rmStrOnce(files.from, ['.zip', '.gzip']); + options = { + gunzip: true + }, + + pipe.create(read, write, options, function(error) { + var name = path.basename(read); + + if (error) + sendError(params, error); + else + sendMsg(params, 'unzip', name); }); + } break; diff --git a/lib/server/rest/fs/get.js b/lib/server/rest/fs/get.js index d4088a2f..733e2e84 100644 --- a/lib/server/rest/fs/get.js +++ b/lib/server/rest/fs/get.js @@ -47,17 +47,13 @@ error = CloudFunc.formatMsg('hash', error, 'error'); func(error); } else - pipe.create({ - from : name, - write : hash, - callback : function (error) { - var hex; - - if (!error) - hex = hash.get(); - - func(error, hex); - } + pipe.create(name, hash, function (error) { + var hex; + + if (!error) + hex = hash.get(); + + func(error, hex); }); break; diff --git a/lib/server/rest/fs/put.js b/lib/server/rest/fs/put.js index 682f104f..750863d7 100644 --- a/lib/server/rest/fs/put.js +++ b/lib/server/rest/fs/put.js @@ -16,7 +16,8 @@ exports.onPut = onPut; function onPut(name, query, readStream, callback) { - var func = Util.exec.ret(callback), + var options, + func = Util.exec.ret(callback), baseName = path.basename(name); switch(query) { @@ -32,18 +33,17 @@ break; default: - pipe.create({ - read : readStream, - gunzip : query === 'unzip', - to : name, - callback : function(error) { - var msg; - - if (!error) - msg = CloudFunc.formatMsg('save', baseName); - - func(error, msg); - } + options = { + gunzip : query === 'unzip' + }; + + pipe.create(readStream, name, function(error) { + var msg; + + if (!error) + msg = CloudFunc.formatMsg('save', baseName); + + func(error, msg); }); break;