-
Notifications
You must be signed in to change notification settings - Fork 1
/
index.js
executable file
·93 lines (81 loc) · 2.84 KB
/
index.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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
#!/usr/bin/env node
const express = require('express')
const SSE = require('express-sse')
const queuesFromRedis = require('./lib/queues-from-redis')
const redis = require('./lib/redis')
const { join, dirname } = require('path')
const { realpathSync } = require('fs')
module.exports = main
async function main ({ clear = true, namespace = 'bull', history = 100, delay = 5000, port = process.env.HTTP_PORT || process.env.PORT || 4000 } = {}) {
const app = express()
const sse = new SSE()
const staticDirPath = join(dirname(realpathSync(process.argv[1])), '..', 'client')
let lastRequestAt
app.use((req, res, next) => {
lastRequestAt = Date.now()
next()
})
app.use('/', express.static(staticDirPath))
app.get('/stream', sse.init)
app.listen(port)
console.log(`visit http://localhost:${port}`)
const redisOptions = {
host: process.env.REDIS_HOST || '0.0.0.0',
password: process.env.REDIS_PASSWORD,
port: process.env.REDIS_PORT ? +process.env.REDIS_PORT : 6379,
db: process.env.REDIS_DB || '0'
}
console.log(JSON.stringify({ redisOptions, namespace }))
const client = redis.getClient(redisOptions)
const queues = await queuesFromRedis(client, namespace)
let lastRunAt
run()
setInterval(run, delay)
async function run () {
if (lastRequestAt < Date.now() - 1000 * 60 * 10) {
console.log('it seems there are non active clients, skipping updates')
return
}
if (lastRunAt && lastRunAt > Date.now() - delay) {
console.log(`already ran during last interval of ${delay}`)
return
}
lastRunAt = Date.now()
// queues = await queuesFromRedis(client, namespace)
clear && console.clear()
clear && console.log(`visit http://localhost:${port}`)
const data = []
for (const queue of queues) {
await Promise.all([
queue.getActive(0, history),
queue.getActiveCount(),
queue.getCompleted(0, history),
queue.getCompletedCount(),
queue.getFailed(0, history),
queue.getFailedCount(),
queue.getWaiting(0, history),
queue.getWaitingCount(),
queue.getDelayed(0, history),
queue.getDelayedCount()
])
.then(([active, activeLength, completed, completedLength, failed, failedLength, waiting, waitingLength, delayed, delayedLength]) => {
process.stdout.write(`-- ${queue.name.padEnd(20)} \tactive: ${activeLength}\tcompleted: ${completedLength}\tfailed: ${failedLength}\twaiting: ${waitingLength}\tdelayed: ${delayedLength}\n`)
data.push({
name: queue.name,
active,
activeLength,
completed,
completedLength,
failed,
failedLength,
waiting,
waitingLength,
delayed,
delayedLength
})
})
}
sse.send(data)
data.length = 0
}
}