var kafkaProducer = {},
    common = require('./../../utils/common.js'),
    logger = require('../../../logger'),
    semusiConfig = require('./../../config'),
    client = require('node-rest-client').Client;


(function (kafkaProducer) {

    var KafkaRest = require('kafka-rest');
    //kafka rest proxy listening to broker at 8082 
    //var kafka = new KafkaRest({ 'url': semusiConfig.KafkaUrl });

    
    //This function writes backupdata to Kafka.           
    kafkaProducer.sendDataToKafka = function(backupData){ 
        try{

            var datajson = JSON.parse(backupData.data);
            var appid = backupData.app_id;
            var did = datajson.mobileid;
            
            if(datajson.internetaccess && datajson.internetaccess.length>0){
                var internetaccessData = {
                    appid : appid,
                    did : did,
                    data : datajson.internetaccess
                }
                writeToKafka("internetaccessdata", internetaccessData,function(){});
            }
            if(datajson.applicationaccess && datajson.applicationaccess.length>0){
                var applicationaccessData = {
                    appid : appid,
                    did : did,
                    data : datajson.applicationaccess
                }
                writeToKafka("applicationaccessdata", applicationaccessData,function(){});
            }
                
            if(datajson.batterydata && datajson.batterydata.length>0){
                var batteryData = {
                    appid : appid,
                    did : did,
                    data : datajson.batterydata
                }
                writeToKafka("batterydata", batteryData,function(){});
            }
            if(datajson.sms && datajson.sms.length>0){
                var smsData = {
                    appid : appid,
                    did : did,
                    data : datajson.sms
                }
                writeToKafka("smsdata", smsData,function(){});
            }
            if(datajson.emails && datajson.emails.length>0){
                var emailsData = {
                    appid : appid,
                    did : did,
                    data : datajson.emails
                }
                writeToKafka("emaildata", emailsData,function(){});
            }
            if(datajson.appusagedata && datajson.appusagedata.length>0){
                var appusageData = {
                    appid : appid,
                    did : did,
                    data : datajson.appusagedata
                }
                writeToKafka("appusagedata", appusageData,function(){});
            }
            if(datajson.places && datajson.places.length>0){
                var placesData = {
                    appid : appid,
                    did : did,
                    data : datajson.places
                }               
                writeToKafka("placesdata", placesData,function(){});
            }
        }
        catch(e){
            logger.error(`error while parsing backup data => ${e}`);
             
        }
    }


    //This function writes API data to Kafka
    kafkaProducer.writeAPIDataToKafka = function(params,callback){
        writeToKafka(params.QueueObject.key,params.QueueObject,callback);
    };


    /*writeToKafka = function(topic,data,callback){
        //which topic you want to send data
        topic = kafka.topic(topic);

        topic.produce(JSON.stringify(data),function(err, res){
            if (err){
                 
                callback(false);
            }
            else{
                 
                callback(true);
            }
        });    
    }*/

    writeToKafka = function(topic,data,callback){
        var proxy = new client();
        var url = semusiConfig.KafkaUrl+"/topics/"+topic;
        var args = {
            data: data,
            headers:{"Content-Type": "application/vnd.kafka.json.v1+json"} 
        };
        proxy.post(url, args, function(data,response) {
            // parsed response body as js object 
            if(data){
                logger.info(`data=> ${data}`);
                callback(true);
            }
            else{
                logger.info(`data=> ${data}`); 
                callback(false);
            }
        });      
    }

    
}(kafkaProducer));

module.exports = kafkaProducer; 