123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255 |
- var session = require('./session');
- var util = require('./util');
- var originApiMap = {};
- var transferToTaskMethod = function (apiMap, apiName) {
- originApiMap[apiName] = apiMap[apiName];
- apiMap[apiName] = function (params, callback) {
- if (params.SkipTask) {
- originApiMap[apiName].call(this, params, callback);
- } else {
- this._addTask(apiName, params, callback);
- }
- };
- };
- var initTask = function (cos) {
- var queue = [];
- var tasks = {};
- var uploadingFileCount = 0;
- var nextUploadIndex = 0;
- // 接口返回简略的任务信息
- var formatTask = function (task) {
- var t = {
- id: task.id,
- Bucket: task.Bucket,
- Region: task.Region,
- Key: task.Key,
- FilePath: task.FilePath,
- state: task.state,
- loaded: task.loaded,
- size: task.size,
- speed: task.speed,
- percent: task.percent,
- hashPercent: task.hashPercent,
- error: task.error,
- };
- if (task.FilePath) t.FilePath = task.FilePath;
- if (task._custom) t._custom = task._custom; // 控制台使用
- return t;
- };
- var emitListUpdate = (function () {
- var timer;
- var emit = function () {
- timer = 0;
- cos.emit('task-list-update', {list: util.map(queue, formatTask)});
- cos.emit('list-update', {list: util.map(queue, formatTask)});
- };
- return function () {
- if (!timer) timer = setTimeout(emit);
- }
- })();
- var clearQueue = function () {
- if (queue.length <= cos.options.UploadQueueSize) return;
- for (var i = 0;
- i < nextUploadIndex && // 小于当前操作的 index 才清理
- i < queue.length && // 大于队列才清理
- queue.length > cos.options.UploadQueueSize // 如果还太多,才继续清理
- ;) {
- var isActive = queue[i].state === 'waiting' || queue[i].state === 'checking' || queue[i].state === 'uploading';
- if (!queue[i] || !isActive) {
- tasks[queue[i].id] && (delete tasks[queue[i].id]);
- queue.splice(i, 1);
- nextUploadIndex--;
- } else {
- i++;
- }
- }
- emitListUpdate();
- };
- var startNextTask = function () {
- // 检查是否允许增加执行进程
- if (uploadingFileCount >= cos.options.FileParallelLimit) return;
- // 跳过不可执行的任务
- while (queue[nextUploadIndex] && queue[nextUploadIndex].state !== 'waiting') nextUploadIndex++;
- // 检查是否已遍历结束
- if (nextUploadIndex >= queue.length) return;
- // 上传该遍历到的任务
- var task = queue[nextUploadIndex];
- nextUploadIndex++;
- uploadingFileCount++;
- task.state = 'checking';
- task.params.onTaskStart && task.params.onTaskStart(formatTask(task));
- !task.params.UploadData && (task.params.UploadData = {});
- var apiParams = util.formatParams(task.api, task.params);
- originApiMap[task.api].call(cos, apiParams, function (err, data) {
- if (!cos._isRunningTask(task.id)) return;
- if (task.state === 'checking' || task.state === 'uploading') {
- task.state = err ? 'error' : 'success';
- err && (task.error = err);
- uploadingFileCount--;
- emitListUpdate();
- startNextTask();
- task.callback && task.callback(err, data);
- if (task.state === 'success') {
- if (task.params) {
- delete task.params.UploadData;
- delete task.params.Body;
- delete task.params;
- }
- delete task.callback;
- }
- }
- clearQueue();
- });
- emitListUpdate();
- // 异步执行下一个任务
- setTimeout(startNextTask);
- };
- var killTask = function (id, switchToState) {
- var task = tasks[id];
- if (!task) return;
- var waiting = task && task.state === 'waiting';
- var running = task && (task.state === 'checking' || task.state === 'uploading');
- if (switchToState === 'canceled' && task.state !== 'canceled' ||
- switchToState === 'paused' && waiting ||
- switchToState === 'paused' && running) {
- if (switchToState === 'paused' && task.params.Body && typeof task.params.Body.pipe === 'function') {
- console.error('stream not support pause');
- return;
- }
- task.state = switchToState;
- cos.emit('inner-kill-task', {TaskId: id, toState: switchToState});
- try {
- var UploadId = task && task.params && task.params.UploadData.UploadId
- } catch(e) {}
- if (switchToState === 'canceled' && UploadId) session.removeUsing(UploadId)
- emitListUpdate();
- if (running) {
- uploadingFileCount--;
- startNextTask();
- }
- if (switchToState === 'canceled') {
- if (task.params) {
- delete task.params.UploadData;
- delete task.params.Body;
- delete task.params;
- }
- delete task.callback;
- }
- }
- clearQueue();
- };
- cos._addTasks = function (taskList) {
- util.each(taskList, function (task) {
- cos._addTask(task.api, task.params, task.callback, true);
- });
- emitListUpdate();
- };
- var isTaskReadyWarning = true;
- cos._addTask = function (api, params, callback, ignoreAddEvent) {
- // 复制参数对象
- params = util.formatParams(api, params);
- // 生成 id
- var id = util.uuid();
- params.TaskId = id;
- params.onTaskReady && params.onTaskReady(id);
- if (params.TaskReady) {
- params.TaskReady(id);
- isTaskReadyWarning && console.warn('warning: Param "TaskReady" has been deprecated. Please use "onTaskReady" instead.');
- isTaskReadyWarning = false;
- }
- var task = {
- // env
- params: params,
- callback: callback,
- api: api,
- index: queue.length,
- // task
- id: id,
- Bucket: params.Bucket,
- Region: params.Region,
- Key: params.Key,
- FilePath: params.FilePath || '',
- state: 'waiting',
- loaded: 0,
- size: 0,
- speed: 0,
- percent: 0,
- hashPercent: 0,
- error: null,
- _custom: params._custom,
- };
- var onHashProgress = params.onHashProgress;
- params.onHashProgress = function (info) {
- if (!cos._isRunningTask(task.id)) return;
- task.hashPercent = info.percent;
- onHashProgress && onHashProgress(info);
- emitListUpdate();
- };
- var onProgress = params.onProgress;
- params.onProgress = function (info) {
- if (!cos._isRunningTask(task.id)) return;
- task.state === 'checking' && (task.state = 'uploading');
- task.loaded = info.loaded;
- task.speed = info.speed;
- task.percent = info.percent;
- onProgress && onProgress(info);
- emitListUpdate();
- };
- // 异步获取 filesize
- util.getFileSize(api, params, function (err, size) {
- // 开始处理上传
- if (err) return callback(util.error(err)); // 如果获取大小出错,不加入队列
- // 获取完文件大小再把任务加入队列
- tasks[id] = task;
- queue.push(task);
- task.size = size;
- !ignoreAddEvent && emitListUpdate();
- startNextTask();
- clearQueue();
- });
- return id;
- };
- cos._isRunningTask = function (id) {
- var task = tasks[id];
- return !!(task && (task.state === 'checking' || task.state === 'uploading'));
- };
- cos.getTaskList = function () {
- return util.map(queue, formatTask);
- };
- cos.cancelTask = function (id) {
- killTask(id, 'canceled');
- };
- cos.pauseTask = function (id) {
- killTask(id, 'paused');
- };
- cos.restartTask = function (id) {
- var task = tasks[id];
- if (task && (task.state === 'paused' || task.state === 'error')) {
- task.state = 'waiting';
- emitListUpdate();
- nextUploadIndex = Math.min(nextUploadIndex, task.index);
- startNextTask();
- }
- };
- cos.isUploadRunning = function () {
- return uploadingFileCount || nextUploadIndex < queue.length;
- };
- };
- module.exports.transferToTaskMethod = transferToTaskMethod;
- module.exports.init = initTask;
|