diff options
| -rw-r--r-- | model.js | 14 | ||||
| -rw-r--r-- | models/asset.js | 27 | ||||
| -rw-r--r-- | worker.js | 51 |
3 files changed, 66 insertions, 26 deletions
@@ -5,18 +5,24 @@ var JsonField = require("sequelize-json"); module.exports = (seq, Seq) => { let Match = seq.import("./models/match.js"), + Asset = seq.import("./models/asset.js"), Roster = seq.import("./models/roster.js"), + //Team = seq.import("./models/team.js"), Participant = seq.import("./models/participant.js"), - Player = seq.import("./models/player.js"); + ParticipantItemUse = seq.import("./models/participant_item_use.js"), + Player = seq.import("./models/player.js"), + Item = seq.import("./models/item.js"); - //Match.hasMany(Asset, {as: "Assets"}); - //Asset.belongsTo(Match); + Match.hasMany(Asset, {as: "Assets"}); + Asset.belongsTo(Match); Roster.belongsTo(Match, { foreignKey: "match_api_id", targetKey: "api_id" }); //Roster.hasOne(Team, {as: "Team"}); //Team.belongsTo(Roster); Participant.belongsTo(Roster, { foreignKey: "roster_api_id", targetKey: "api_id" }); Participant.hasOne(Player, { foreignKey: "player_api_id", targetKey: "api_id" }); + ParticipantItemUse.belongsTo(Participant, { foreignKey: "participant_api_id", targetKey: "api_id" }); + ParticipantItemUse.hasOne(Item); //return {Match, Roster, Team, Participant, Player, Asset}; - return {Match, Roster, Participant, Player}; + return {Match, Roster, Participant, ParticipantItemUse, Player, Asset, Item}; }; diff --git a/models/asset.js b/models/asset.js index 54a9a82..af8d040 100644 --- a/models/asset.js +++ b/models/asset.js @@ -10,9 +10,34 @@ module.exports = function(sequelize, DataTypes) { }, api_id: { type: DataTypes.STRING(191), + allowNull: false, + unique: true + }, + match_api_id: { + type: DataTypes.STRING(191), + allowNull: false + }, + url: { + type: DataTypes.STRING(191), + allowNull: false + }, + content_type: { + type: DataTypes.STRING(191), + allowNull: false + }, + created_at: { + type: DataTypes.TIME, + allowNull: false + }, + description: { + type: DataTypes.STRING(191), + allowNull: false + }, + filename: { + type: DataTypes.STRING(191), allowNull: false }, - type: { + name: { type: DataTypes.STRING(191), allowNull: false } @@ -7,7 +7,7 @@ var amqp = require("amqplib"), var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", - BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 2 + 3*2 + 3*2), + BATCHSIZE = process.env.PROCESSOR_BATCH || 50, // matches IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { @@ -37,11 +37,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", if (timer === undefined) timer = setTimeout(process, IDLE_TIMEOUT) if (queue.length == BATCHSIZE) - process(); + await process(); }, { noAck: false }); async function process() { - console.log("processing batch"); + console.log("processing batch", queue.length); // clean up to allow processor to accept while we wait for db let matchmsgs = queue.slice(); @@ -71,7 +71,9 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // UPSERT try { let matches = matchmsgs.map((msg) => JSON.parse(msg.content)); - await matches.forEach(async (match) => { + await Promise.all(matches.map(async (match) => { + console.log("processing match", match.id); + // flatten jsonapi nested response into our db structure-like shape match.rosters = match.rosters.map((roster) => { roster.participants = roster.participants.map((participant) => { @@ -80,6 +82,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", }); return flatten(roster); }); + match.assets = match.assets.map((asset) => flatten(asset)); match = flatten(match); // upsert match @@ -89,15 +92,15 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // upsert children // before, add foreign keys and other missing information (shardId) - await match.rosters.forEach(async (roster) => { + await Promise.all(match.rosters.map(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 ] + include: [ model.Participant/*, model.Team */] }); - await roster.participants.forEach(async (participant) => { + await Promise.all(roster.participants.map(async (participant) => { participant.player.shard_id = participant.shard_id; await model.Player.upsert(participant.player); @@ -107,32 +110,38 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await model.Participant.upsert(participant, { include: [ model.Player ] }); - }); + })); 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 model.Team.upsert(roster.team);*/ } - }); + })); - await match.assets.forEach(async (asset) => { + await Promise.all(match.assets.map(async (asset) => { + asset.match_api_id = match.api_id; + asset.shard_id = match.shard_id; + asset.url = asset.uRL; // camelcasization failed ;) await model.Asset.upsert(asset); - }); - }); + })); + })); // COMMIT await transaction.commit(); - await ch.ack(matchmsgs.pop(), true); // ack all messages until the last + console.log("acking batch"); + await Promise.all(matchmsgs.map(async (msg) => { + await ch.ack(msg); + })); // request child jobs, notify player - await matches.forEach(async (m) => { - await m.rosters.forEach(async (r) => { - await r.participants.forEach(async (p) => { + await Promise.all(matches.map(async (m) => { + await Promise.all(m.rosters.map(async (r) => { + await Promise.all(r.participants.map(async (p) => { await ch.publish("amq.topic", p.player.name, new Buffer("process_commit")); - }); - }); - }); - } catch (err) { // TODO catch only SQL error, also catch errors in the forEach + })); + })); + })); + } catch (err) { // TODO catch only SQL error, also catch errors in the promises console.error(err); await ch.nack(matchmsgs.pop(), true, true); // nack all messages until the last and requeue // TODO don't requeue broken records |