stream_logs.js

var redis = require("redis"),
    client = redis.createClient();
var amqp = require('amqp');
var amqpc = amqp.createConnection({ host: 'somehost', login: 'guest', password: 'guest', vhost: '/' });
var geoip = require('geoip-lite');

var ll = require("lazylines");
amqpc.on('ready', function () {
    amqpc.exchange('heatmap', {passive: true}, function(e){
        process.stdin.resume();
        var inp = new ll.LineReadStream(process.stdin, 'ascii');
        inp.on("line", function (line) {
            var tmp = ll.chomp(line).split(';'), ip = tmp[0], paths = tmp[1].split(',');
            var geo = geoip.lookup(ip);
            if (geo && geo['ll'])
                for (var i in paths) {
                    client.get(paths[i], function(err, reply) {
                        e.publish('latlong.path', JSON.stringify({
                            src: geo['ll'],
                            dest: JSON.parse(reply)
                        }));
                    });
                }

        });
    });
});