var net = require('net');
var redis = require('redis');
var moment = require('moment');
var kafka = require('kafka-node');
var isEmpty = require('is-empty');
var config = require('./config');
var method = require('./method');
var log4js = require('log4js');
var crc16 = require('./hj212-2017/crc16_2017_Complement');

var redisClient = redis.createClient(config.REDIS_OPTS);
log4js.configure('./log4js.json');
var logger1 = log4js.getLogger('log_date');

var Producer = kafka.Producer;
var client = new kafka.KafkaClient({kafkaHost: config.KAFKA_HOST});

var producerOption = {
    requireAcks: 1,
    ackTimeoutMs: 100,
    partitionerType: 3 //默认为第一个分区
};
var  producer = new Producer(client,producerOption);

global.dev_sockets = {};
global.app_sockets = {};

let macs = [];

function pushData(mac, cp_stores, socket, entryTime, cn){
    var originaltime = cp_stores['DataTime'];
    var datatime = originaltime.substr(0, 4) + "/" + originaltime.substr(4, 2) + "/" + originaltime.substr(6, 2) + " " + originaltime.substr(8, 2) + ":" + originaltime.substr(10, 2) + ":" + originaltime.substr(12, 2);
    var timeTamp = new Date(datatime).getTime();
    var data = {
        mac: mac,
        time: moment().valueOf(),
        entryTime: entryTime
        //time: timeTamp
    };

    for (var key in cp_stores){
        var value = cp_stores[key];
        data[key] = value;
    };

    redisClient.get("is_start", function (err, reply) {
        if (err) {
            console.log(err);
        } else {
            if (reply == 1) {
                // 发往调试页面 http://dbg.7drlb.com:3000/3rd_party
                //producer.pushToKafka("devices_data_3rd_party");
            }
        }
    });

    var topic = '';
    if (cn === '2011'){
        topic = 'uav_second_data';
    }else {
        topic = 'test';
    }
    producer.send([{topic: topic, messages: JSON.stringify(data), key: JSON.stringify(mac)}], function (err, data) {
        /*dataInsert.delete();
        dataIns.delete();*/
        if (err){
            //console.log('++++++'+err);
            //logger.error(err);
        }
        if (!err) {
            //console.log("send message complete!data:" + JSON.stringify(data), new Date());
        }
    });
}

function handleData(socket, mac, stores, entryTime){
    var cp_stores = method.get_cp_data_store(stores["CP"]);
    //console.log("handleData => " + JSON.stringify(cp_stores));

    var cn = stores['CN'];

    pushData(mac, cp_stores, socket, entryTime, cn);
}

function doWork(socket, data, entryTime) {
    if (isEmpty(data)) return;

    if (!data.startsWith('##') && !data.startsWith('@@')) return;

    if (data.length < 6) return;
    var length = parseInt(data.substr(2, 4));

    if (data.length < length + 10) return;

    var stores = method.get_data_store(data, length);

    if (isEmpty(stores) || isEmpty(stores['MN'])) return;

    var mac = stores['MN'].toLowerCase();

    // CRC校验
    var crc_data = data.substr(6 + length, 4);
    var data_val = data.substr(6, length);
    if (isEmpty(crc_data) || crc16(data_val, data_val.length) != crc_data.toUpperCase()) return;

    //反控
    if (data.startsWith('@@')) { //云端控制
        var command = data.replace("@@", "##");
        if (!isEmpty(dev_sockets[mac]) && dev_sockets[mac].writable) {
            // console.log("RC => " + command);
            dev_sockets[mac].write(command);
        }
        if (!isEmpty(mac)) {
            app_sockets[mac] = socket;
        }
    } else { //硬件设备
        if (data.startsWith('##') && !isEmpty(mac)) {
            dev_sockets[mac] = socket;
        }

        //console.log(stores['Flag']);
        if(stores['Flag']==='5'){
            var res = 'QN='+stores['QN']+';ST=91;CN=9014;PW=123456;MN='+mac+';Flag=4;CP=&&&&\r\n';
            logger1.info(moment().format('YYYY-MM-DD HH:mm:ss') + ' ANSWER: '  + ':' + res);
            //dev_sockets[mac].write(res);
            socket.write(res);
        }

        if (!macs.includes(mac)) {
            macs.push(mac);
            // console.log(moment().format('YYYY-MM-DD HH:mm:ss') + " :: " + mac);
        }

        handleData(socket, mac, stores, entryTime);
    }

    doWork(socket, data.slice(length + 12), entryTime);
}

net.createServer().on('connection', function (socket) {
    logger1.info(moment().format('YYYY-MM-DD HH:mm:ss') + ' CONNECTED: ' + socket.remoteAddress + ':' + socket.remotePort);
    //console.log(moment().format('YYYY-MM-DD HH:mm:ss') + ' CONNECTED: ' + socket.remoteAddress + ':' + socket.remotePort);

    socket.on('data', function (data) {
        logger1.info(moment().format('YYYY-MM-DD HH:mm:ss') + ' DATA: '  + ':' + data);
        var entryTime = moment().valueOf();
        var str_data = data.toString();
        var cur_time = moment().format('YYYY-MM-DD HH:mm:ss');
        //console.log(cur_time + " => " + str_data);

        redisClient.get("is_start", function (err, reply) {
            if (err) {
                console.log(err);
            } else {
                if (reply == 1) {
                    // 发往调试页面
                    //producer.pushToKafka("devices_data_3rd_party");
                }
            }
        });
        doWork(socket, str_data, entryTime);
    });

    socket.on('end', function () {

    });

    socket.on('error', function (error) {
        console.log(moment().format('YYYY-MM-DD HH:mm:ss') + ': ' + error);
        logger1.error(moment().format('YYYY-MM-DD HH:mm:ss') + ': ' + error)
        //socket.end();
    });

    socket.on('timeout', function () {
        //socket.end();
        logger1.error(moment().format('YYYY-MM-DD HH:mm:ss') + ': ' + "timeout");
    });

    socket.on('close', function (data) {
        console.log(moment().format('YYYY-MM-DD HH:mm:ss') + ' Closed socket: ' + socket.remoteAddress + ' ' + socket.remotePort);
        logger1.info(moment().format('YYYY-MM-DD HH:mm:ss') + ' Closed socket: ' + socket.remoteAddress + ' ' + socket.remotePort)
        for (var mac in dev_sockets) {
            if (dev_sockets[mac] == socket) {
                delete dev_sockets[mac];
                break;
            }
        }
    });
}).listen(config.UAV_PORT, config.HOST);