task.js 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255
  1. var session = require('./session');
  2. var util = require('./util');
  3. var originApiMap = {};
  4. var transferToTaskMethod = function (apiMap, apiName) {
  5. originApiMap[apiName] = apiMap[apiName];
  6. apiMap[apiName] = function (params, callback) {
  7. if (params.SkipTask) {
  8. originApiMap[apiName].call(this, params, callback);
  9. } else {
  10. this._addTask(apiName, params, callback);
  11. }
  12. };
  13. };
  14. var initTask = function (cos) {
  15. var queue = [];
  16. var tasks = {};
  17. var uploadingFileCount = 0;
  18. var nextUploadIndex = 0;
  19. // 接口返回简略的任务信息
  20. var formatTask = function (task) {
  21. var t = {
  22. id: task.id,
  23. Bucket: task.Bucket,
  24. Region: task.Region,
  25. Key: task.Key,
  26. FilePath: task.FilePath,
  27. state: task.state,
  28. loaded: task.loaded,
  29. size: task.size,
  30. speed: task.speed,
  31. percent: task.percent,
  32. hashPercent: task.hashPercent,
  33. error: task.error,
  34. };
  35. if (task.FilePath) t.FilePath = task.FilePath;
  36. if (task._custom) t._custom = task._custom; // 控制台使用
  37. return t;
  38. };
  39. var emitListUpdate = (function () {
  40. var timer;
  41. var emit = function () {
  42. timer = 0;
  43. cos.emit('task-list-update', {list: util.map(queue, formatTask)});
  44. cos.emit('list-update', {list: util.map(queue, formatTask)});
  45. };
  46. return function () {
  47. if (!timer) timer = setTimeout(emit);
  48. }
  49. })();
  50. var clearQueue = function () {
  51. if (queue.length <= cos.options.UploadQueueSize) return;
  52. for (var i = 0;
  53. i < nextUploadIndex && // 小于当前操作的 index 才清理
  54. i < queue.length && // 大于队列才清理
  55. queue.length > cos.options.UploadQueueSize // 如果还太多,才继续清理
  56. ;) {
  57. var isActive = queue[i].state === 'waiting' || queue[i].state === 'checking' || queue[i].state === 'uploading';
  58. if (!queue[i] || !isActive) {
  59. tasks[queue[i].id] && (delete tasks[queue[i].id]);
  60. queue.splice(i, 1);
  61. nextUploadIndex--;
  62. } else {
  63. i++;
  64. }
  65. }
  66. emitListUpdate();
  67. };
  68. var startNextTask = function () {
  69. // 检查是否允许增加执行进程
  70. if (uploadingFileCount >= cos.options.FileParallelLimit) return;
  71. // 跳过不可执行的任务
  72. while (queue[nextUploadIndex] && queue[nextUploadIndex].state !== 'waiting') nextUploadIndex++;
  73. // 检查是否已遍历结束
  74. if (nextUploadIndex >= queue.length) return;
  75. // 上传该遍历到的任务
  76. var task = queue[nextUploadIndex];
  77. nextUploadIndex++;
  78. uploadingFileCount++;
  79. task.state = 'checking';
  80. task.params.onTaskStart && task.params.onTaskStart(formatTask(task));
  81. !task.params.UploadData && (task.params.UploadData = {});
  82. var apiParams = util.formatParams(task.api, task.params);
  83. originApiMap[task.api].call(cos, apiParams, function (err, data) {
  84. if (!cos._isRunningTask(task.id)) return;
  85. if (task.state === 'checking' || task.state === 'uploading') {
  86. task.state = err ? 'error' : 'success';
  87. err && (task.error = err);
  88. uploadingFileCount--;
  89. emitListUpdate();
  90. startNextTask();
  91. task.callback && task.callback(err, data);
  92. if (task.state === 'success') {
  93. if (task.params) {
  94. delete task.params.UploadData;
  95. delete task.params.Body;
  96. delete task.params;
  97. }
  98. delete task.callback;
  99. }
  100. }
  101. clearQueue();
  102. });
  103. emitListUpdate();
  104. // 异步执行下一个任务
  105. setTimeout(startNextTask);
  106. };
  107. var killTask = function (id, switchToState) {
  108. var task = tasks[id];
  109. if (!task) return;
  110. var waiting = task && task.state === 'waiting';
  111. var running = task && (task.state === 'checking' || task.state === 'uploading');
  112. if (switchToState === 'canceled' && task.state !== 'canceled' ||
  113. switchToState === 'paused' && waiting ||
  114. switchToState === 'paused' && running) {
  115. if (switchToState === 'paused' && task.params.Body && typeof task.params.Body.pipe === 'function') {
  116. console.error('stream not support pause');
  117. return;
  118. }
  119. task.state = switchToState;
  120. cos.emit('inner-kill-task', {TaskId: id, toState: switchToState});
  121. try {
  122. var UploadId = task && task.params && task.params.UploadData.UploadId
  123. } catch(e) {}
  124. if (switchToState === 'canceled' && UploadId) session.removeUsing(UploadId)
  125. emitListUpdate();
  126. if (running) {
  127. uploadingFileCount--;
  128. startNextTask();
  129. }
  130. if (switchToState === 'canceled') {
  131. if (task.params) {
  132. delete task.params.UploadData;
  133. delete task.params.Body;
  134. delete task.params;
  135. }
  136. delete task.callback;
  137. }
  138. }
  139. clearQueue();
  140. };
  141. cos._addTasks = function (taskList) {
  142. util.each(taskList, function (task) {
  143. cos._addTask(task.api, task.params, task.callback, true);
  144. });
  145. emitListUpdate();
  146. };
  147. var isTaskReadyWarning = true;
  148. cos._addTask = function (api, params, callback, ignoreAddEvent) {
  149. // 复制参数对象
  150. params = util.formatParams(api, params);
  151. // 生成 id
  152. var id = util.uuid();
  153. params.TaskId = id;
  154. params.onTaskReady && params.onTaskReady(id);
  155. if (params.TaskReady) {
  156. params.TaskReady(id);
  157. isTaskReadyWarning && console.warn('warning: Param "TaskReady" has been deprecated. Please use "onTaskReady" instead.');
  158. isTaskReadyWarning = false;
  159. }
  160. var task = {
  161. // env
  162. params: params,
  163. callback: callback,
  164. api: api,
  165. index: queue.length,
  166. // task
  167. id: id,
  168. Bucket: params.Bucket,
  169. Region: params.Region,
  170. Key: params.Key,
  171. FilePath: params.FilePath || '',
  172. state: 'waiting',
  173. loaded: 0,
  174. size: 0,
  175. speed: 0,
  176. percent: 0,
  177. hashPercent: 0,
  178. error: null,
  179. _custom: params._custom,
  180. };
  181. var onHashProgress = params.onHashProgress;
  182. params.onHashProgress = function (info) {
  183. if (!cos._isRunningTask(task.id)) return;
  184. task.hashPercent = info.percent;
  185. onHashProgress && onHashProgress(info);
  186. emitListUpdate();
  187. };
  188. var onProgress = params.onProgress;
  189. params.onProgress = function (info) {
  190. if (!cos._isRunningTask(task.id)) return;
  191. task.state === 'checking' && (task.state = 'uploading');
  192. task.loaded = info.loaded;
  193. task.speed = info.speed;
  194. task.percent = info.percent;
  195. onProgress && onProgress(info);
  196. emitListUpdate();
  197. };
  198. // 异步获取 filesize
  199. util.getFileSize(api, params, function (err, size) {
  200. // 开始处理上传
  201. if (err) return callback(util.error(err)); // 如果获取大小出错,不加入队列
  202. // 获取完文件大小再把任务加入队列
  203. tasks[id] = task;
  204. queue.push(task);
  205. task.size = size;
  206. !ignoreAddEvent && emitListUpdate();
  207. startNextTask();
  208. clearQueue();
  209. });
  210. return id;
  211. };
  212. cos._isRunningTask = function (id) {
  213. var task = tasks[id];
  214. return !!(task && (task.state === 'checking' || task.state === 'uploading'));
  215. };
  216. cos.getTaskList = function () {
  217. return util.map(queue, formatTask);
  218. };
  219. cos.cancelTask = function (id) {
  220. killTask(id, 'canceled');
  221. };
  222. cos.pauseTask = function (id) {
  223. killTask(id, 'paused');
  224. };
  225. cos.restartTask = function (id) {
  226. var task = tasks[id];
  227. if (task && (task.state === 'paused' || task.state === 'error')) {
  228. task.state = 'waiting';
  229. emitListUpdate();
  230. nextUploadIndex = Math.min(nextUploadIndex, task.index);
  231. startNextTask();
  232. }
  233. };
  234. cos.isUploadRunning = function () {
  235. return uploadingFileCount || nextUploadIndex < queue.length;
  236. };
  237. };
  238. module.exports.transferToTaskMethod = transferToTaskMethod;
  239. module.exports.init = initTask;