爱智EdgerOS之深入解析离线下载任务

一、需求分析

在日常使用计算机的过程中，看到喜欢的资源不可避免地想把它下载到我们的设备上保存下来，比如图片，音视频资源，文档资源等，基于这种应用场景，现在来看看在爱智设备上可以如何实现呢？
因为是离线下载，所以爱智 APP Web 端向爱智 APP Server 端发送一个请求，爱智后端启动多任务线程完成下载任务，这时候前端就可以关闭了。
需要将下载的进度实时返回到前端，使用 Socket.IO 将进度推送到前端，下载任务和 Socket.IO 之间需要异步通信，使用 SigSlot。

在这里插入图片描述

二、什么是 SigSlot？

SigSlot 是一个事件驱动的异步通信组件，支持多任务和多进程，这也是为什么选择使用 SigSlot 在 Task 中进行通信的原因，它继承自 EventEmitter，是一个典型的订阅和发布通信机制。
SigSlot 的功能还远不止于此，当应用申请开启 GSS 支持后，来自同一开发供应商的应用程序可以通过 GSS 的功能互相订阅和发布消息。在这里，只在同一应用中的多个线程中使用它的异步通信功能。

三、WebGet

WebGet 模块用于获取 http 数据，它支持分段请求数据，以及断点续传等，通过调用 WebGet 上的 file 方法来将数据保存到文件中，可以在选项中指定数据的起始位置、每段数据的大小以及并行请求的数量。
如果文件存在并且设置了 reload 为 true，WebGet 对象将检查日志并启动断点恢复过程。

四、实现过程

① 前端实现

前端部分是正常的发送请求，以及监听进度部分，这里给出大致的代码段：

// 前端以 vue3 为例实现
// home.jsx
import { defineComponent, ref, onMounted } from 'vue'
import { useRouter } from 'vue-router'
import { download } from '../apis/index'

export default defineComponent({
  name: 'Home',
  setup (props, ctx) {
    const router = useRouter()
    const downloadList = ref([])

    onMounted(() => {
      socketio.socket.on('progress', data => {
        downloadList.value.forEach(item => {
          if (item.hash === data.hash) {
            item.progress = data.data
            item.name = data.name
          }
        })
        if (data.data === 100) {
          console.log('>>>> download over <<<<')
        }
      })
    })
  }
})

这里在 onMounted 的生命周期中接收 Socket.IO 传递过来的下载进度的数据，然后更新到对应的 DataView 中。

② Server 端实现

Server 端部分主要看三个部分的内容，router 中触发下载任务：

// routers/rest.js
const Router = require('webapp').Router;
const SigSlot = require('sigslot');
const fs = require('fs')

const sigslot = new SigSlot('download');

/* Create router */
const router = Router.create();

// download task
const task = new Task('./download-task.js', 'download-task', {
  directory: module.directory
})

// 下载文件接口
router.post('/download', function(req, res) {
  sigslot.emit('download', {
    url: req.body.url,
    hash: req.body.hash
  })
  res.json({
    code: 0,
    msg: 'success'
  })
});
// 获取下载文件的资源列表
router.get('/download-list', function(req, res) {
  res.json({
    code: 0,
    msg: 'ok',
    data: fs.dumpdir('./public/download')
  })
})

/* Export router */
module.exports = router;

在路由一开始就将下载的线程运行了起来，在其内部监听一个 download 的 Sigslot 任务，路由这里接收到下载的请求后，就将参数通过 Sigslot 发送到了 Task 线程。
接下来看一下 Task 线程中的内容：


// routers/download-task.js
const WebGet = require('webget');
const SigSlot = require('sigslot');
const fs = require('fs')

const sigslot = new SigSlot('download');
fs.mkdir('./public/download', 0o666, true)

// Subscribe
sigslot.slot('download', (msg) => {
  const url = msg.url;
  const fileNameArr = url.split('/')
  const fileName = fileNameArr[fileNameArr.length - 1]
  const path = `./public/download/${fileName}`;
  let totalSize = 0

  WebGet.file(url, path, {
    limits: 512,
    lines: 2,
    reload: true
  }, (loader) => {
    loader.on('response', (info) => {
      totalSize = info.requestSize;
      // console.log(`Download begin, original=${info.originalSize}, total=${totalSize}, loaded=${info.loadedSize}`);
    });

    loader.on('data', (chunk, info) => {
      const progress = info.completeSize / totalSize * 100;
      // console.log(`Recv data, size=${chunk.byteLength}, offset=${info.offset}, progress=${progress}%`);
      sigslot.emit('progress', {
        data: Math.round(progress),
        hash: msg.hash,
        name: fileName
      })
    });

    loader.on('end', () => {
      console.log(`Download finish, file: ${path}`);
    });

    loader.on('error', (e) => {
      console.log('Download error:', e.message);
    });
  })
});

require('iosched').forever();

Task 中只做一件事，外层 SigSlot 订阅下载任务，然后使用 WebGet.file 进行资源下载，按照需求配置分片以及断点续传等功能。WebGet.file 的回调函数返回 WebGet 对象，可以在 WebGet 对象的 data 事件中获取计算进度的数据，最后再通过 SigSlot 发布到 Socket.IO 的订阅事件中。
最后看返回给前端的下载进度：


// main.js
/* Import system modules */
const WebApp = require('webapp');
const io = require('socket.io');
const SigSlot = require('sigslot');
const bodyParser = require('middleware').bodyParser;

const sigslot = new SigSlot('download');

/* Import routers */
const myRouter = require('./routers/rest');

/* Create App */
const app = WebApp.createApp();

/* Set static path */
app.use(WebApp.static('./public'));

// body parser
app.use(bodyParser.json()); // for parsing application/json
app.use(bodyParser.urlencoded()); // for parsing application/x-www-form-urlencoded

/* Set test rest */
app.use('/api', myRouter);

/* Rend test */
app.get('/temp.html', function (req, res) {
  res.render('temp', { time: Date.now() });
});

/* Start App */
app.start();

const socketio = io(app, {
  serveClient: false,
  pingInterval: 10000,
  pingTimeout: 5000
})

socketio.on('connection', socket => {
  console.log('>> socket connected <<')

  // Subscribe
  sigslot.slot('progress', (msg) => {
    // console.log('>> main download arrived:', msg.data);
    socket.emit('progress', msg)
  });

})

/* Event loop */
require('iosched').forever();