1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
|
#!/usr/bin/node
/* jshint esnext:true */
"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"),
api = require("../orm/api");
const MADGLORY_TOKEN = process.env.MADGLORY_TOKEN,
QUEUE = process.env.QUEUE || "grab",
PROCESS_QUEUE = process.env.PROCESS_QUEUE || "process",
SAMPLE_QUEUE = process.env.SAMPLE_QUEUE || "sample",
RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
LOGGLY_TOKEN = process.env.LOGGLY_TOKEN,
GRABBERS = parseInt(process.env.GRABBERS) || 4;
if (MADGLORY_TOKEN == undefined) throw "Need an API token";
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", "apigrabber", 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});
await ch.assertQueue(PROCESS_QUEUE, {durable: true});
break;
} catch (err) {
logger.error("error connecting", err);
await sleep(5000);
}
}
await ch.prefetch(GRABBERS);
// main queue
ch.consume(QUEUE, async (msg) => {
let payload = JSON.parse(msg.content.toString()),
notify = msg.properties.headers.notify; // where to send progress report
if (msg.properties.type == "matches")
await getAPI(payload, "matches", notify);
if (msg.properties.type == "samples")
await getAPI(payload, "samples");
logger.info("done", payload);
ch.ack(msg);
}, { noAck: false });
// loop over API data objects
async function getAPI(payload, where, notify="global") {
const data = await Promise.each(await api.requests(where,
payload.region, payload.params, logger),
async (data, idx, len) => {
if (where == "matches") { // send match structure
await ch.sendToQueue(PROCESS_QUEUE,
new Buffer(JSON.stringify(data)), {
persistent: true, type: "match",
headers: idx == len - 1? { notify: notify } : {}
});
// forward "notify" for the last match on the last page
if (notify) await ch.publish("amq.topic", notify,
new Buffer("match_pending"));
}
if (where == "samples")
// forward to sampler
await ch.sendToQueue(SAMPLE_QUEUE,
new Buffer(JSON.stringify(data.attributes.URL)),
{ persistent: true, type: "sample" });
return data;
}
);
if (data.length == 0 && notify)
await ch.publish("amq.topic", notify,
new Buffer("matches_none"));
}
})();
process.on("unhandledRejection", err => {
logger.error("Uncaught Promise Error:", err.stack);
});
|
