From bd9919db7e8cafcf9ff76c3135e574b20aafceea Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 8 Aug 2017 16:07:42 +0200 Subject: add forward to reaper --- worker.js | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/worker.js b/worker.js index dd376ad..67e9637 100644 --- a/worker.js +++ b/worker.js @@ -21,8 +21,8 @@ const RABBITMQ_URI = process.env.RABBITMQ_URI, // maximum number of elements to be inserted in one statement CHUNKSIZE = parseInt(process.env.CHUNKSIZE) || 100, MAXCONNS = parseInt(process.env.MAXCONNS) || 10, // how many concurrent actions - DOANALYZEMATCH = process.env.DOANALYZEMATCH == "true", - ANALYZE_QUEUE = process.env.ANALYZE_QUEUE || "analyze", + DOREAPMATCH = process.env.DOREAPMATCH == "true", + REAP_QUEUE = process.env.REAP_QUEUE || "reap", LOAD_TIMEOUT = parseFloat(process.env.LOAD_TIMEOUT) || 5000, // ms IDLE_TIMEOUT = parseFloat(process.env.IDLE_TIMEOUT) || 700; // ms @@ -67,6 +67,7 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { const ch = await rabbit.createChannel(); await ch.assertQueue(QUEUE, { durable: true }); await ch.assertQueue(QUEUE + "_failed", { durable: true }); + await ch.assertQueue(REAP_QUEUE, { durable: true }); // as long as the queue is filled, msg are not ACKed // server sends as long as there are less than `prefetch` unACKed await ch.prefetch(BATCHSIZE); @@ -178,6 +179,16 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { m.properties.headers.notify, new Buffer("phase_update")) }); + if (DOREAPMATCH) + await Promise.map(telemetry_objects, async (t) => + await ch.sendToQueue(REAP_QUEUE, new Buffer( + JSON.stringify({ + match_api_id: t.match_api_id, + start: t.start < 0? 0 : t.start, // TODO see below + end: t.end + }), { persistent: true }) + ) + ); } // finish a whole batch -- cgit v1.3.1