44 lines
1.0 KiB
JavaScript
44 lines
1.0 KiB
JavaScript
|
const queue = require('../../queue/queue');
|
||
|
|
const logger = require('../../util/logger');
|
||
|
|
const config = require('../../config/config');
|
||
|
|
const { deliverOutboxEntry } = require('../../host/resultService');
|
||
|
|
|
||
|
|
let running = false;
|
||
|
|
let workerPromise;
|
||
|
|
|
||
|
|
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
|
||
|
|
|
||
|
|
async function loop() {
|
||
|
|
while (running) {
|
||
|
|
try {
|
||
|
|
const batch = await queue.claimPending(config.worker.batchSize, config.worker.workerId);
|
||
|
|
if (!batch.length) {
|
||
|
|
await sleep(config.worker.pollInterval);
|
||
|
|
continue;
|
||
|
|
}
|
||
|
|
|
||
|
|
for (const entry of batch) {
|
||
|
|
await deliverOutboxEntry(entry);
|
||
|
|
}
|
||
|
|
} catch (err) {
|
||
|
|
logger.error({ err: err.message }, 'delivery worker error');
|
||
|
|
await sleep(config.worker.pollInterval);
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
async function startWorker() {
|
||
|
|
if (running) return;
|
||
|
|
running = true;
|
||
|
|
workerPromise = loop();
|
||
|
|
}
|
||
|
|
|
||
|
|
async function stopWorker() {
|
||
|
|
running = false;
|
||
|
|
if (workerPromise) {
|
||
|
|
await workerPromise;
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
module.exports = { startWorker, stopWorker };
|