refactoring

Resolve #7779

1 files changed, 60 insertions, 0 deletions

diff --git a/packages/backend/src/daemons/queue-stats.ts b/packages/backend/src/daemons/queue-stats.ts
new file mode 100644
index 0000000000..77f09b18d6
--- /dev/null
+++ b/packages/backend/src/daemons/queue-stats.ts
@@ -0,0 +1,60 @@
+import Xev from 'xev';
+import { deliverQueue, inboxQueue } from '../queue/queues';
+
+const ev = new Xev();
+
+const interval = 10000;
+
+/**
+ * Report queue stats regularly
+ */
+export default function() {
+	const log = [] as any[];
+
+	ev.on('requestQueueStatsLog', x => {
+		ev.emit(`queueStatsLog:${x.id}`, log.slice(0, x.length || 50));
+	});
+
+	let activeDeliverJobs = 0;
+	let activeInboxJobs = 0;
+
+	deliverQueue.on('global:active', () => {
+		activeDeliverJobs++;
+	});
+
+	inboxQueue.on('global:active', () => {
+		activeInboxJobs++;
+	});
+
+	async function tick() {
+		const deliverJobCounts = await deliverQueue.getJobCounts();
+		const inboxJobCounts = await inboxQueue.getJobCounts();
+
+		const stats = {
+			deliver: {
+				activeSincePrevTick: activeDeliverJobs,
+				active: deliverJobCounts.active,
+				waiting: deliverJobCounts.waiting,
+				delayed: deliverJobCounts.delayed
+			},
+			inbox: {
+				activeSincePrevTick: activeInboxJobs,
+				active: inboxJobCounts.active,
+				waiting: inboxJobCounts.waiting,
+				delayed: inboxJobCounts.delayed
+			},
+		};
+
+		ev.emit('queueStats', stats);
+
+		log.unshift(stats);
+		if (log.length > 200) log.pop();
+
+		activeDeliverJobs = 0;
+		activeInboxJobs = 0;
+	}
+
+	tick();
+
+	setInterval(tick, interval);
+}