diff options
Diffstat (limited to 'worker.js')
| -rw-r--r-- | worker.js | 38 |
1 files changed, 27 insertions, 11 deletions
@@ -96,19 +96,27 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { // connect to rabbit & db const seq = new Seq(DATABASE_URI, { - //logging: false, - max: MAXCONNS + logging: false, + pool: { + max: MAXCONNS + } }); const ch = await rabbit.createChannel(); await ch.assertQueue(QUEUE, { durable: true }); await ch.assertQueue(QUEUE + "_failed", { durable: true }); + await ch.assertQueue(ANALYZE_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); const model = require("../orm/model")(seq, Seq); + logger.info("configuration", { + QUEUE, BATCHSIZE, CHUNKSIZE, MAXCONNS, LOAD_TIMEOUT, IDLE_TIMEOUT, + DOANALYZEMATCH, ANALYZE_QUEUE + }); + // performance logging let load_timer = undefined, idle_timer = undefined, @@ -168,22 +176,29 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { const match = JSON.parse(msg.content); // deduplicate and reject immediately if (await model.Match.count({ where: { api_id: match.id } }) > 0) { + logger.info("duplicate match", match.id); if (msg.properties.headers.notify) { await ch.publish("amq.topic", msg.properties.headers.notify, - new Buffer("matches_dupe")) + new Buffer("matches_dupe")); // send match_dupe to web player.ign.api_id await ch.publish("amq.topic", msg.properties.headers.notify + "." + match.id, - new Buffer("match_dupe")) + new Buffer("match_dupe")); + // HOTFIX TODO remove me, web should listen to match_dupe + await ch.publish("amq.topic", + msg.properties.headers.notify, + new Buffer("match_update")); } //await ch.nack(msg, false, false); } else if (match.rosters.length < 2 || match.rosters[0].id == "null") { + logger.info("invalid match", match.id); // it is really `"null"`. // reject invalid matches (handling API bugs) //await ch.nack(msg, false, false); await ch.sendToQueue(QUEUE + "_failed", msg.content, { persistent: true, + type: msg.properties.type, headers: msg.properties.headers }); } else { @@ -218,6 +233,7 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { profiler = undefined; logger.info("processing batch", { + messages: msgs.size, players: player_data.size, matches: match_data.size }); @@ -228,11 +244,10 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { idle_timer = undefined; load_timer = undefined; - if (player_data.size + match_data.size == 0) { - logger.info("buffers empty, nothing to do"); + if (msgs.size == 0) { + logger.info("nothing to do"); return; } - const player_objects = new Set(player_data), match_objects = new Set(match_data); player_data.clear(); @@ -247,11 +262,9 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { // notify web await Promise.map(msgs, async (m) => { if (m.properties.headers.notify == undefined) return; - let notif = ""; switch (m.properties.type) { // new match case "match": - notif = "matches_update"; // notify player.name.api_id about match_update await Promise.map(match_objects, async (mat) => await ch.publish("amq.topic", @@ -263,7 +276,8 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { break; case "player": // player obj updated - notif = "stats_update"; + await ch.publish("amq.topic", m.properties.headers.notify, + new Buffer("stats_update")); break; } }); @@ -276,7 +290,8 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { await ch.sendToQueue(ANALYZE_QUEUE, new Buffer(m.id), { persistent: true })); } catch (err) { - if (err instanceof Seq.TimeoutError) { + if (err instanceof Seq.TimeoutError || + (err instanceof Seq.DatabaseError && err.errno == 1213)) { // deadlocks / timeout logger.error("SQL error", err); //await Promise.map(msgs, async (m) => @@ -287,6 +302,7 @@ amqp.connect(RABBITMQ_URI).then(async (rabbit) => { await Promise.map(msgs, async (m) => { await ch.sendToQueue(QUEUE + "_failed", m.content, { persistent: true, + type: m.properties.type, headers: m.properties.headers }); //await ch.nack(m, false, false); |