managed-upload.js 10.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360
  1. // var debug = require('debug')('ali-oss:multipart');
  2. const util = require('util');
  3. const path = require('path');
  4. const mime = require('mime');
  5. const copy = require('copy-to');
  6. const { isBlob } = require('../common/utils/isBlob');
  7. const { isFile } = require('../common/utils/isFile');
  8. const { isBuffer } = require('../common/utils/isBuffer');
  9. const proto = exports;
  10. /**
  11. * Multipart operations
  12. */
  13. /**
  14. * Upload a file to OSS using multipart uploads
  15. * @param {String} name
  16. * @param {String|File|Buffer} file
  17. * @param {Object} options
  18. * {Object} options.callback The callback parameter is composed of a JSON string encoded in Base64
  19. * {String} options.callback.url the OSS sends a callback request to this URL
  20. * {String} options.callback.host The host header value for initiating callback requests
  21. * {String} options.callback.body The value of the request body when a callback is initiated
  22. * {String} options.callback.contentType The Content-Type of the callback requests initiatiated
  23. * {Object} options.callback.customValue Custom parameters are a map of key-values, e.g:
  24. * customValue = {
  25. * key1: 'value1',
  26. * key2: 'value2'
  27. * }
  28. */
  29. proto.multipartUpload = async function multipartUpload(name, file, options = {}) {
  30. this.resetCancelFlag();
  31. options.disabledMD5 = options.disabledMD5 === undefined ? true : !!options.disabledMD5;
  32. if (options.checkpoint && options.checkpoint.uploadId) {
  33. if (file && isFile(file)) options.checkpoint.file = file;
  34. return await this._resumeMultipart(options.checkpoint, options);
  35. }
  36. const minPartSize = 100 * 1024;
  37. if (!options.mime) {
  38. if (isFile(file)) {
  39. options.mime = mime.getType(path.extname(file.name));
  40. } else if (isBlob(file)) {
  41. options.mime = file.type;
  42. } else if (isBuffer(file)) {
  43. options.mime = '';
  44. } else {
  45. options.mime = mime.getType(path.extname(file));
  46. }
  47. }
  48. options.headers = options.headers || {};
  49. this._convertMetaToHeaders(options.meta, options.headers);
  50. const fileSize = await this._getFileSize(file);
  51. if (fileSize < minPartSize) {
  52. options.contentLength = fileSize;
  53. const result = await this.put(name, file, options);
  54. if (options && options.progress) {
  55. await options.progress(1);
  56. }
  57. const ret = {
  58. res: result.res,
  59. bucket: this.options.bucket,
  60. name,
  61. etag: result.res.headers.etag
  62. };
  63. if ((options.headers && options.headers['x-oss-callback']) || options.callback) {
  64. ret.data = result.data;
  65. }
  66. return ret;
  67. }
  68. if (options.partSize && !(parseInt(options.partSize, 10) === options.partSize)) {
  69. throw new Error('partSize must be int number');
  70. }
  71. if (options.partSize && options.partSize < minPartSize) {
  72. throw new Error(`partSize must not be smaller than ${minPartSize}`);
  73. }
  74. const initResult = await this.initMultipartUpload(name, options);
  75. const { uploadId } = initResult;
  76. const partSize = this._getPartSize(fileSize, options.partSize);
  77. const checkpoint = {
  78. file,
  79. name,
  80. fileSize,
  81. partSize,
  82. uploadId,
  83. doneParts: []
  84. };
  85. if (options && options.progress) {
  86. await options.progress(0, checkpoint, initResult.res);
  87. }
  88. return await this._resumeMultipart(checkpoint, options);
  89. };
  90. /*
  91. * Resume multipart upload from checkpoint. The checkpoint will be
  92. * updated after each successful part upload.
  93. * @param {Object} checkpoint the checkpoint
  94. * @param {Object} options
  95. */
  96. proto._resumeMultipart = async function _resumeMultipart(checkpoint, options) {
  97. const that = this;
  98. if (this.isCancel()) {
  99. throw this._makeCancelEvent();
  100. }
  101. const { file, fileSize, partSize, uploadId, doneParts, name } = checkpoint;
  102. const internalDoneParts = [];
  103. if (doneParts.length > 0) {
  104. copy(doneParts).to(internalDoneParts);
  105. }
  106. const partOffs = this._divideParts(fileSize, partSize);
  107. const numParts = partOffs.length;
  108. let multipartFinish = false;
  109. let uploadPartJob = (self, partNo) => {
  110. // eslint-disable-next-line no-async-promise-executor
  111. return new Promise(async (resolve, reject) => {
  112. try {
  113. if (!self.isCancel()) {
  114. const pi = partOffs[partNo - 1];
  115. const content = await self._createBuffer(file, pi.start, pi.end);
  116. const data = {
  117. content,
  118. size: pi.end - pi.start
  119. };
  120. let result;
  121. try {
  122. result = await self._uploadPart(name, uploadId, partNo, data, options);
  123. } catch (error) {
  124. if (error.status === 404) {
  125. throw self._makeAbortEvent();
  126. }
  127. throw error;
  128. }
  129. if (!self.isCancel() && !multipartFinish) {
  130. checkpoint.doneParts.push({
  131. number: partNo,
  132. etag: result.res.headers.etag
  133. });
  134. if (options.progress) {
  135. await options.progress(doneParts.length / (numParts + 1), checkpoint, result.res);
  136. }
  137. resolve({
  138. number: partNo,
  139. etag: result.res.headers.etag
  140. });
  141. } else {
  142. resolve();
  143. }
  144. } else {
  145. resolve();
  146. }
  147. } catch (err) {
  148. const tempErr = new Error();
  149. tempErr.name = err.name;
  150. tempErr.message = err.message;
  151. tempErr.stack = err.stack;
  152. tempErr.partNum = partNo;
  153. copy(err).to(tempErr);
  154. reject(tempErr);
  155. }
  156. });
  157. };
  158. const all = Array.from(new Array(numParts), (x, i) => i + 1);
  159. const done = internalDoneParts.map(p => p.number);
  160. const todo = all.filter(p => done.indexOf(p) < 0);
  161. const defaultParallel = 5;
  162. const parallel = options.parallel || defaultParallel;
  163. // upload in parallel
  164. const jobErr = await this._parallel(todo, parallel, value => {
  165. return new Promise((resolve, reject) => {
  166. uploadPartJob(that, value)
  167. .then(result => {
  168. if (result) {
  169. internalDoneParts.push(result);
  170. }
  171. resolve();
  172. })
  173. .catch(err => {
  174. reject(err);
  175. });
  176. });
  177. });
  178. multipartFinish = true;
  179. const abortEvent = jobErr.find(err => err.name === 'abort');
  180. if (abortEvent) throw abortEvent;
  181. if (this.isCancel()) {
  182. uploadPartJob = null;
  183. throw this._makeCancelEvent();
  184. }
  185. if (jobErr && jobErr.length > 0) {
  186. jobErr[0].message = `Failed to upload some parts with error: ${jobErr[0].toString()} part_num: ${
  187. jobErr[0].partNum
  188. }`;
  189. throw jobErr[0];
  190. }
  191. return await this.completeMultipartUpload(name, uploadId, internalDoneParts, options);
  192. };
  193. /**
  194. * Get file size
  195. */
  196. proto._getFileSize = async function _getFileSize(file) {
  197. if (isBuffer(file)) {
  198. return file.length;
  199. } else if (isBlob(file) || isFile(file)) {
  200. return file.size;
  201. }
  202. throw new Error('_getFileSize requires Buffer/File/Blob.');
  203. };
  204. /*
  205. * Readable stream for Web File
  206. */
  207. const { Readable } = require('stream');
  208. function WebFileReadStream(file, options) {
  209. if (!(this instanceof WebFileReadStream)) {
  210. return new WebFileReadStream(file, options);
  211. }
  212. Readable.call(this, options);
  213. this.file = file;
  214. this.reader = new FileReader();
  215. this.start = 0;
  216. this.finish = false;
  217. this.fileBuffer = null;
  218. }
  219. util.inherits(WebFileReadStream, Readable);
  220. WebFileReadStream.prototype.readFileAndPush = function readFileAndPush(size) {
  221. if (this.fileBuffer) {
  222. let pushRet = true;
  223. while (pushRet && this.fileBuffer && this.start < this.fileBuffer.length) {
  224. const { start } = this;
  225. let end = start + size;
  226. end = end > this.fileBuffer.length ? this.fileBuffer.length : end;
  227. this.start = end;
  228. pushRet = this.push(this.fileBuffer.slice(start, end));
  229. }
  230. }
  231. };
  232. WebFileReadStream.prototype._read = function _read(size) {
  233. if (
  234. (this.file && this.start >= this.file.size) ||
  235. (this.fileBuffer && this.start >= this.fileBuffer.length) ||
  236. this.finish ||
  237. (this.start === 0 && !this.file)
  238. ) {
  239. if (!this.finish) {
  240. this.fileBuffer = null;
  241. this.finish = true;
  242. }
  243. this.push(null);
  244. return;
  245. }
  246. const defaultReadSize = 16 * 1024;
  247. size = size || defaultReadSize;
  248. const that = this;
  249. this.reader.onload = function onload(e) {
  250. that.fileBuffer = Buffer.from(new Uint8Array(e.target.result));
  251. that.file = null;
  252. that.readFileAndPush(size);
  253. };
  254. if (this.start === 0) {
  255. this.reader.readAsArrayBuffer(this.file);
  256. } else {
  257. this.readFileAndPush(size);
  258. }
  259. };
  260. function getBuffer(file) {
  261. // Some browsers do not support Blob.prototype.arrayBuffer, such as IE
  262. if (file.arrayBuffer) return file.arrayBuffer();
  263. return new Promise((resolve, reject) => {
  264. const reader = new FileReader();
  265. reader.onload = function (e) {
  266. resolve(e.target.result);
  267. };
  268. reader.onerror = function (e) {
  269. reject(e);
  270. };
  271. reader.readAsArrayBuffer(file);
  272. });
  273. }
  274. proto._createBuffer = async function _createBuffer(file, start, end) {
  275. if (isBlob(file) || isFile(file)) {
  276. const _file = file.slice(start, end);
  277. const fileContent = await getBuffer(_file);
  278. return Buffer.from(fileContent);
  279. } else if (isBuffer(file)) {
  280. return file.subarray(start, end);
  281. } else {
  282. throw new Error('_createBuffer requires File/Blob/Buffer.');
  283. }
  284. };
  285. proto._getPartSize = function _getPartSize(fileSize, partSize) {
  286. const maxNumParts = 10 * 1000;
  287. const defaultPartSize = 1 * 1024 * 1024;
  288. if (!partSize) partSize = defaultPartSize;
  289. const safeSize = Math.ceil(fileSize / maxNumParts);
  290. if (partSize < safeSize) {
  291. partSize = safeSize;
  292. console.warn(
  293. `partSize has been set to ${partSize}, because the partSize you provided causes partNumber to be greater than 10,000`
  294. );
  295. }
  296. return partSize;
  297. };
  298. proto._divideParts = function _divideParts(fileSize, partSize) {
  299. const numParts = Math.ceil(fileSize / partSize);
  300. const partOffs = [];
  301. for (let i = 0; i < numParts; i++) {
  302. const start = partSize * i;
  303. const end = Math.min(start + partSize, fileSize);
  304. partOffs.push({
  305. start,
  306. end
  307. });
  308. }
  309. return partOffs;
  310. };