'use strict';
// simple example of nodejs app publishing json messages from a kafka queue via socket.io
//
// install using: npm install express http kafka-node optimist socket.io monitor.io
//
// runit like nodejs index.js --topic=some_topic --zk=server:2181 --port=80

var app = require('express')();
var hbs = require('express-hbs');
var server = require('http').Server(app);
var kafka = require('kafka-node');

var HighLevelConsumer = kafka.HighLevelConsumer;
var Offset = kafka.Offset;
var Client = kafka.Client;
var argv = require('yargs')
           .boolean('from_beginning')
           .boolean('verbose').argv;
if(argv.help){
  console.log("--zk=zk1:2181,zk2:2181 --topic=topic1,topic2,topic3 --port=8080");
  process.exit(code=0)
}
var port = argv.port || 9080;
var zk = argv.zk || 'localhost:2181';
var topiclist = (argv.topic || 'weather_display').split(",");
var fromBeginning = argv.from_beginning;
var verbose =argv.verbose;
var topics = [];
for(var i=0;i<topiclist.length;i++){
  topics.push({ topic: topiclist[i] });
}
var client = new Client(zk);
var options = { autoCommit: true, fromBeginning: fromBeginning, fetchMaxWaitMs: 1000, fetchMaxBytes: 1024*1024 };
var consumer = new HighLevelConsumer(client, topics, options);
var offset = new Offset(client);

var io = require('socket.io')(server);
server.listen(port);

/*
 * comment in the next two lines to use monitor.io
 */
//var monitorio = require("monitor.io");
//io.use(monitorio({port: port+1 }));

app.engine('hbs', hbs.express4());
app.set('view engine', 'hbs');


app.get('/', function (req, res) {
    var laport = req.app.settings.port || port;
    var url = req.protocol + '://' + req.host  + ( laport == 80 || laport == 443 ? '' : ':'+laport ) + "/";
    res.render('index',{url: url, topics: topics});
});

consumer.on('message', function (message) {
    if(verbose){
      console.log(this.id, message);
    }else{
      //console.log(message.topic);
    }
    try{
      var msg = message.value;
      try{
        msg = JSON.parse(message.value);
      }catch(e){
         //ok it's not json
      }
      io.emit(message.topic, msg);
    }catch(err){
      console.log('error', err);
    }
});
consumer.on('error', function (err) {
    console.log('error', err);
});
consumer.on('offsetOutOfRange', function (topic) {
    topic.maxNum = 2;
    offset.fetch([topic], function (err, offsets) {
        var min = Math.min.apply(null, offsets[topic.topic][topic.partition]);
        consumer.setOffset(topic.topic, topic.partition, min);
    });
});
