diff options
| -rw-r--r-- | .gitignore | 59 | ||||
| -rw-r--r-- | Dockerfile | 9 | ||||
| -rw-r--r-- | package.json | 22 | ||||
| -rw-r--r-- | worker.js | 90 |
4 files changed, 180 insertions, 0 deletions
diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..00cbbdf --- /dev/null +++ b/.gitignore @@ -0,0 +1,59 @@ +# Logs +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* + +# Runtime data +pids +*.pid +*.seed +*.pid.lock + +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov + +# Coverage directory used by tools like istanbul +coverage + +# nyc test coverage +.nyc_output + +# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files) +.grunt + +# Bower dependency directory (https://bower.io/) +bower_components + +# node-waf configuration +.lock-wscript + +# Compiled binary addons (http://nodejs.org/api/addons.html) +build/Release + +# Dependency directories +node_modules/ +jspm_packages/ + +# Typescript v1 declaration files +typings/ + +# Optional npm cache directory +.npm + +# Optional eslint cache +.eslintcache + +# Optional REPL history +.node_repl_history + +# Output of 'npm pack' +*.tgz + +# Yarn Integrity file +.yarn-integrity + +# dotenv environment variables file +.env + diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9c3debd --- /dev/null +++ b/Dockerfile @@ -0,0 +1,9 @@ +FROM node:7.7-alpine + +RUN mkdir -p /usr/src/app +WORKDIR /usr/src/app + +COPY . /usr/src/app +RUN npm install && npm cache clean + +CMD ["node", "worker.js"] diff --git a/package.json b/package.json new file mode 100644 index 0000000..7cf7d80 --- /dev/null +++ b/package.json @@ -0,0 +1,22 @@ +{ + "name": "apigrabber", + "version": "2.0.0", + "description": "", + "main": "worker.js", + "dependencies": { + "adm-zip": "^0.4.7", + "amqplib": "^0.5.1", + "bluebird": "^3.5.0", + "request": "^2.81.0", + "request-promise": "^4.2.0", + "sleep-promise": "^2.0.0", + "winston": "^2.3.1", + "winston-loggly-bulk": "^1.4.2" + }, + "devDependencies": {}, + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1" + }, + "author": "schneefux", + "license": "UNLICENSED" +} diff --git a/worker.js b/worker.js new file mode 100644 index 0000000..c97b7c2 --- /dev/null +++ b/worker.js @@ -0,0 +1,90 @@ +#!/usr/bin/node +/* jshint esnext:true */ +/* download data from AWS and push into process queues */ +"use strict"; + +const amqp = require("amqplib"), + Promise = require("bluebird"), + winston = require("winston"), + loggly = require("winston-loggly-bulk"), + request = require("request-promise"), + sleep = require("sleep-promise"), + jsonapi = require("../orm/jsonapi"), + AdmZip = require("adm-zip"); + +const RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", + QUEUE = process.env.QUEUE || "sample", + PROCESS_QUEUE = process.env.PROCESS_QUEUE || "process", + PROCESS_BRAWL_QUEUE = process.env.PROCESS_BRAWL_QUEUE || "process_brawl", + LOGGLY_TOKEN = process.env.LOGGLY_TOKEN, + SAMPLERS = parseInt(process.env.SAMPLERS) || 5; + +const logger = new (winston.Logger)({ + transports: [ + new (winston.transports.Console)({ + timestamp: true, + colorize: true + }) + ] +}); + +// loggly integration +if (LOGGLY_TOKEN) + logger.add(winston.transports.Loggly, { + inputToken: LOGGLY_TOKEN, + subdomain: "kvahuja", + tags: ["backend", "sampler", QUEUE], + json: true + }); + +(async () => { + let rabbit, ch; + + while (true) { + try { + rabbit = await amqp.connect(RABBITMQ_URI); + ch = await rabbit.createChannel(); + await ch.assertQueue(QUEUE, {durable: true}); + break; + } catch (err) { + logger.error("error connecting", err); + await sleep(5000); + } + } + + await ch.prefetch(SAMPLERS); + ch.consume(QUEUE, async (msg) => { + const payload = JSON.parse(msg.content.toString()); + if (msg.properties.type == "sample") + await getSample(payload); + + logger.info("done", payload); + ch.ack(msg); + }, { noAck: false }); + + // download a sample ZIP and send to processor + async function getSample(url) { + logger.info("downloading sample", url); + const zipdata = await request({ + uri: url, + encoding: null + }), + zip = new AdmZip(zipdata); + await Promise.map(zip.getEntries(), async (entry) => { + if (entry.isDirectory) return; + const match = jsonapi.parse(JSON.parse(entry.getData().toString("utf8"))); + await sendMatchToProcessor(match); + }); + logger.info("sample processed", url); + } + + // send to seperated queues or just to `process` + async function sendMatchToProcessor(match) { + if (["casual", "ranked"].indexOf(match.attributes.gameMode) != -1) + await ch.sendToQueue(PROCESS_QUEUE, new Buffer(JSON.stringify(match)), + { persistent: true, type: "match" }) + else + await ch.sendToQueue(PROCESS_BRAWL_QUEUE, new Buffer(JSON.stringify(match)), + { persistent: true, type: "match" }) + } +})(); |