cloud/client/modules/upload.js

;
(function () {
    'use strict';
    const
        event = require('myevents'),
        Upload = require('uploadfile'),
        taskStatus = require('taskstatus'),
        md5file = require('md5file'),
        nedbCollection = require('nedbCollection'),
        appCache = require('app-cache'),
        uploadTask = require('uploadtask'),
        apiService = require('app-service'),
        sysConfig = require('sys-config'),
        transfer = require('transfer'),
        dbServer = require('db-server'),
        fs = require('fs');

    var cacheConfig = appCache.local_config();

    var _task = {
        parallelUploadNum: cacheConfig.task.upload_num || 3,

        start: function (upload) {
            let uploadManagement = new Upload(upload);
            upload = uploadManagement.options;
            upload.status = taskStatus.read;
            dbServer.upload.status(upload._id, taskStatus.read);
            event.emit('upload_change_ctr', upload);
            if (!uploadTask.add(upload))
                return _task.startNext();//任务已经存在了，进行下一波下载
            wfsGateway(upload).then(e => {
                upload.url = e.host;
                dbServer.upload.modifyMd5(upload._id, upload.md5, upload.url);
                upload.status = taskStatus.doing;
                event.emit('upload_change_ctr', upload);
                dbServer.upload.status(upload._id, taskStatus.doing).then(e => {
                    return _task._maxSize();
                }).then(e => {
                    if (e)
                        uploadManagement.startSB();
                    else {
                        upload.status = taskStatus.spaceLimit;
                        event.emit('upload_change_ctr', upload);
                        dbServer.upload.status(upload._id, upload.status).then(e => {
                            uploadTask.remove(upload._id);
                            return _task.startNext();
                        })
                    }
                });
            });
        },

        startNext: function (id) {
            // if (!uploadTask.queue_status)
            //     return id && dbServer.upload.status(id, taskStatus.stop);

            if (uploadTask.taskNum >= _task.parallelUploadNum) {
                return dbServer.upload.status(id, taskStatus.wait);
            };
            nextTask(id).then(upload => {
                if (!upload) return event.emit('upload_complete_ctr', { _id: id });
                if (uploadTask.taskNum >= _task.parallelUploadNum) {
                    return dbServer.upload.status(id, taskStatus.wait)
                };
                _task.start(upload);
            });
        },
        stop: function (id) {
            let options = uploadTask.find(id);
            if (options) options.status = taskStatus.pause;
            dbServer.upload.status(id, taskStatus.pause).then(e => {
                uploadTask.remove(id);
                _task.startNext();
            });
        },
        delete: function (id) {
            let options = uploadTask.find(id);
            if (options) options.status = taskStatus.stop;
            dbServer.upload.remove(id).then(e => {
                uploadTask.remove(id);
                _task.startNext();
            });
        },
        clearQueue: function () {
            return dbServer.folder.clear().then(e => {
                return dbServer.upload.clear();
            }).then(e => {
                let list = uploadTask.list();
                for (let e in list) {
                    list[e].status = taskStatus.stop;
                };
                uploadTask.clear();
            });
        },
        stopQueue: function () {
            return dbServer.upload.status({}, taskStatus.pause).then(e => {
                let list = uploadTask.list();
                uploadTask.stopAll();
                for (let e in list) {
                    list[e].status = taskStatus.stop;
                };
                uploadTask.clear();
            });
        },
        startQueue: function () {
            return dbServer.upload.status({}, taskStatus.wait).then(e => {
                let num = cacheConfig.task.upload_num;
                uploadTask.startAll();
                do {
                    num = num - 1;
                    _task.startNext();
                } while (num)
            });
        },
        _maxSize: function () {
            let list = uploadTask.list(),
                size = 0;
            for (let e in list) size += list[e].size;
            return use_space().then(e => { return (e + size) < sysConfig.storage_space; })
        },
        _onUplodComplete: function (task, options) {
            event.emit('upload_complete_ctr', options);
            let _id = options._id;
            delete options._id;
            uploadTask.remove(_id);
            dbServer
                .upload
                .remove(_id)
                .then(e => {
                    _task.startNext();
                });
            event.emit('upload_notice', options);
            event.emit('task_num_change');
            // event.emit('fileListRefresh');
        },
        _onRangeFinish: function (task, options) {
            dbServer
                .upload
                .updateRange(options._id, options.end_len)
                .then(function () {
                    event.emit('upload_change_ctr', options);
                })
        },
        _onPause: function (task, options) {
            dbServer.upload.status(task.options._id, taskStatus.pause);
            uploadTask.remove(task.options._id)
            _task.startNext();
            event.emit('upload_change_ctr', options);
        },

        _upload_error: (task, options) => {
            let _id = options._id;
            delete options._id;
            dbServer
                .upload
                .status(options._id, options.status)
                .then(e => {
                    _task.startNext();
                    event.emit('upload_change_ctr', options);
                });
        }
    };

    function MD5(upload) {
        return md5file(upload.path).then(e => e);
    }

    function wfsGateway(upload) {
        if (upload.md5)
            return transfer.wfsGateway.upload(upload.md5); // ({ md5: file_md5, url: e.host })
        else {
            return MD5(upload).then(md5 => { upload.md5 = md5; return wfsGateway(upload) });
        }
    }

    function nextTask(id) {
        return !id ? dbServer.upload.nextTask() : dbServer.upload.findOne({ '_id': id });
    }
    event.on('upload_error', _task._upload_error);

    event.on('upload_complete', _task._onUplodComplete);

    event.on('upload_range_finish', _task._onRangeFinish);

    event.on('upload_pause', _task._onPause);

    function use_space() {
        return apiService.use_space().then((d) => {
            return d.data.totalAmount;
        });
    }

    transfer.uploadmanager = _task;
}());