diff options
| -rw-r--r-- | models/match.js | 3 | ||||
| -rw-r--r-- | models/participant.js | 2 | ||||
| -rw-r--r-- | models/roster.js | 4 | ||||
| -rw-r--r-- | worker.js | 141 |
4 files changed, 94 insertions, 56 deletions
diff --git a/models/match.js b/models/match.js index 46c4f80..ec5bdab 100644 --- a/models/match.js +++ b/models/match.js @@ -23,8 +23,7 @@ module.exports = function(sequelize, DataTypes) { }, created_at: { type: DataTypes.TIME, - allowNull: false, - defaultValue: sequelize.literal('CURRENT_TIMESTAMP') + allowNull: false }, duration: { type: DataTypes.INTEGER(11), diff --git a/models/participant.js b/models/participant.js index 9bd1ff7..9c17e74 100644 --- a/models/participant.js +++ b/models/participant.js @@ -59,7 +59,7 @@ module.exports = function(sequelize, DataTypes) { }, gold: { type: DataTypes.INTEGER(11), - allowNull: false + allowNull: true }, gold_mine_captures: { type: DataTypes.INTEGER(11), diff --git a/models/roster.js b/models/roster.js index 53a4411..3848c6a 100644 --- a/models/roster.js +++ b/models/roster.js @@ -41,10 +41,6 @@ module.exports = function(sequelize, DataTypes) { type: DataTypes.STRING(191), allowNull: false }, - team_color: { - type: DataTypes.STRING(191), - allowNull: false - }, turret_kills: { type: DataTypes.INTEGER(11), allowNull: false @@ -6,7 +6,9 @@ var amqp = require("amqplib"), Seq = require("sequelize"); var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", - DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite"; + DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", + BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 2 + 3*2 + 3*2), + IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { let seq = new Seq(DATABASE_URI), @@ -14,21 +16,45 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", rabbit = await amqp.connect(RABBITMQ_URI), ch = await rabbit.createChannel(); + let queue = [], + timer = undefined; + /* recreate for debugging await seq.query("SET FOREIGN_KEY_CHECKS=0"); await seq.sync({force: true}); */ - //await seq.sync(); + await seq.sync(); await ch.assertQueue("process", {durable: true}); - await ch.prefetch(1); + // 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); ch.consume("process", async (msg) => { - let match = JSON.parse(msg.content); + queue.push(msg); + + // fill queue until batchsize or idle + if (timer === undefined) + timer = setTimeout(process, IDLE_TIMEOUT) + if (queue.length == BATCHSIZE) + process(); + }, { noAck: false }); + + async function process() { + console.log("processing batch"); + + // clean up to allow processor to accept while we wait for db + let matchmsgs = queue.slice(); + queue = []; + clearTimeout(timer); + timer = undefined; - // TODO commit less often if possible, avoid deadlocks + // BEGIN let transaction = await seq.transaction({ autocommit: false }); + // helper to convert API response into flat JSON + // db structure is (almost) 1:1 the API structure + // so we can insert the flat API response as-is function flatten(obj) { let attrs = obj.attributes || {}, stats = attrs.stats || {}, @@ -42,57 +68,74 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", return o; } - /* bring jsonapi nested response into our db structure-like shape */ - match.rosters = match.rosters.map((roster) => { - roster.participants = roster.participants.map((participant) => { - participant.player = flatten(participant.player); - return flatten(participant); - }); - return flatten(roster); - }); - match = flatten(match); - console.log(match); - - /* upsert everything */ - await model.Match.upsert(match, { - include: [ model.Roster/*, model.Asset*/ ] - }); + // UPSERT + try { + let matches = matchmsgs.map((msg) => JSON.parse(msg.content)); + await matches.forEach(async (match) => { + // flatten jsonapi nested response into our db structure-like shape + match.rosters = match.rosters.map((roster) => { + roster.participants = roster.participants.map((participant) => { + participant.player = flatten(participant.player); + return flatten(participant); + }); + return flatten(roster); + }); + match = flatten(match); - await match.rosters.forEach(async (roster) => { - roster.match_api_id = match.api_id; - roster.shard_id = match.shard_id; - await model.Roster.upsert(roster, { - include: [ model.Participant/*, model.Team*/ ] - }); + // upsert match + await model.Match.upsert(match, { + include: [ model.Roster, model.Asset ] + }); - await roster.participants.forEach(async (participant) => { - participant.shard_id = roster.shard_id; - participant.player.shard_id = participant.shard_id; - - await model.Player.upsert(participant.player); + // upsert children + // before, add foreign keys and other missing information (shardId) + await match.rosters.forEach(async (roster) => { + roster.match_api_id = match.api_id; + roster.shard_id = match.shard_id; + await model.Roster.upsert(roster, { + include: [ model.Participant, model.Team ] + }); - participant.roster_api_id = roster.api_id; - participant.player_api_id = participant.player.api_id; - await model.Participant.upsert(participant, { - include: [ model.Player ] - }); - }); + await roster.participants.forEach(async (participant) => { + participant.player.shard_id = participant.shard_id; + await model.Player.upsert(participant.player); - //if (roster.team != null) model.Team.upsert(roster.team); - }); + participant.shard_id = roster.shard_id; + participant.roster_api_id = roster.api_id; + participant.player_api_id = participant.player.api_id; + await model.Participant.upsert(participant, { + include: [ model.Player ] + }); + }); - /*match.assets.forEach((asset) => { - model.Asset.upsert(asset); - });*/ + if (roster.team != null) { + roster.team.shard_id = roster.shard_id; + roster.team.roster_api_id = roster.api_id; + await model.Team.upsert(roster.team); + } + }); - await transaction.commit(); // TODO rollback on err - ch.ack(msg); + await match.assets.forEach(async (asset) => { + await model.Asset.upsert(asset); + }); + }); - await match.rosters.forEach(async (r) => { - await r.participants.forEach(async (p) => { - await ch.publish("amq.topic", p.player.name, new Buffer("process_commit")); + // COMMIT + await transaction.commit(); + await ch.ack(matchmsgs.pop(), true); // ack all messages until the last + // request child jobs, notify player + await matches.forEach(async (m) => { + await m.rosters.forEach(async (r) => { + await r.participants.forEach(async (p) => { + await ch.publish("amq.topic", p.player.name, new Buffer("process_commit")); + }); + }); }); - }); - }, { noAck: false }); + } catch (err) { // TODO catch only SQL error, also catch errors in the forEach + console.error(err); + await ch.nack(matchmsgs.pop(), true, true); // nack all messages until the last and requeue + // TODO don't requeue broken records + } + } })(); |