/
socket.js
64 lines (57 loc) · 2.08 KB
/
socket.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
const SocketServer = require('ws').Server
const { jsonStringify } = require('./helpers/utils')
const emitter = require('./emitter')
const redis = require('./redis')
const logger = require('./logger')
const { STORAGE_PREFIX, shortenToken } = require('./Uploader')
/**
* the socket is used to send progress events during an upload
*
* @param {import('http').Server | import('https').Server} server
*/
module.exports = (server) => {
const wss = new SocketServer({ server })
const redisClient = redis.client()
// A new connection is usually created when an upload begins,
// or when connection fails while an upload is on-going and,
// client attempts to reconnect.
wss.on('connection', (ws, req) => {
const fullPath = req.url
// the token identifies which ongoing upload's progress, the socket
// connection wishes to listen to.
const token = fullPath.replace(/^.*\/api\//, '')
logger.info(`connection received from ${token}`, 'socket.connect')
/**
*
* @param {{action: string, payload: object}} data
*/
function sendProgress (data) {
ws.send(jsonStringify(data), (err) => {
if (err) logger.error(err, 'socket.progress.error', shortenToken(token))
})
}
// if the redisClient is available, then we attempt to check the storage
// if we have any already stored progress data on the upload.
if (redisClient) {
redisClient.get(`${STORAGE_PREFIX}:${token}`, (err, data) => {
if (err) logger.error(err, 'socket.redis.error', shortenToken(token))
if (data) {
const dataObj = JSON.parse(data.toString())
if (dataObj.action) sendProgress(dataObj)
}
})
}
emitter().emit(`connection:${token}`)
emitter().on(token, sendProgress)
ws.on('message', (jsonData) => {
const data = JSON.parse(jsonData.toString())
// whitelist triggered actions
if (['pause', 'resume', 'cancel'].includes(data.action)) {
emitter().emit(`${data.action}:${token}`)
}
})
ws.on('close', () => {
emitter().removeListener(token, sendProgress)
})
})
}