let workerNodes = {}, timeline = {}
const constants = require('../constants_local.json')
const Heap = require('heap');
// const libSupport = require('./lib');
const dgram = require('dgram');
const udpProxy = dgram.createSocket('udp4');

const loadThreashold = 1
var workerHeap = new Heap(function(worker1, worker2) {
    return (worker1.system_info.avg_load[0] - worker2.system_info.avg_load[0]);
});
let coldstart_worker = {};

let resourceMap =  new Map(),
    functionToResourceMap = new Map();



let kafka = require('kafka-node'),
    Producer = kafka.Producer,
    client = new kafka.KafkaClient({
        kafkaHost: constants.network.external.kafka_host,
        autoConnect: true
    }),
    producer = new Producer(client),
    Consumer = kafka.Consumer,
    consumer = new Consumer(client,
        [
            { topic: constants.topics.heartbeat }, // receives heartbeat messages from workers, also acts as worker join message
            { topic: constants.topics.request_dm_2_rm }, // receives deployment details from RM
            { topic: constants.topics.check_autoscale },
            { topic: constants.topics.autoscale },
            { topic: constants.topics.coldstart_worker } // give the information about worker having low load
        ],
        [
            { autoCommit: true }
        ])

function getAddress() {
    return workerHeap.peek().address;
    // return Object.keys(workerNodes)[0];
}

// payloads = [
//     { topic: 'heartbeat', messages: 'hi', partition: 0 }
// ];
// producer.on('ready', function () {
//     producer.send(payloads, function (err, data) {
//         console.log(data);
//     });
// });

function updateColdstartOnNIC() {
    let lowloadworker = workerHeap.peek();
    if(lowloadworker.address != coldstart_worker.address)
    {
        console.log("min load worker changed !!!");
        coldstart_worker = lowloadworker;

        let payload = [{
            topic: constants.topics.coldstart_worker ,
            messages: JSON.stringify({
                    "resource_id": coldstart_worker.resource_id,
                    "timestamp": Date.now(),
                    "nodes": 
                        { 
                            node_id: coldstart_worker.address, 
                            port: coldstart_worker.port, 
                            mac: coldstart_worker.mac
                        }
            }),
            partition: 0
        }]
        producer.send(payload, () => {
            console.log(`Replied`)
        })
        
    }
}

consumer.on('message', function (message) {

    let topic = message.topic
    message = message.value
    // console.log("message ",message)
    
    if (topic !== constants.topics.heartbeat)
        console.log(message);

    if (topic === constants.topics.heartbeat) {
        message = JSON.parse(message)
        if (Date.now() - message.timestamp < 1000)
        {
            if (!workerNodes[message.address]) {
                workerNodes[message.address] = message
                console.log("New worker discovered. Worker List: ")
                console.log(workerNodes);
                workerHeap.push(workerNodes[message.address])
                // if(Object.keys(workerNodes).length === 1)
                // {
                //     updateColdstartOnNIC();
                // }
            }
            else{
                // console.log("Got heartbeat updating load of wroker ", message.address)
                if(Math.abs(workerNodes[message.address].system_info.avg_load[0] - message.system_info.avg_load[0]) > loadThreashold )
                {
                    workerNodes[message.address].system_info = message.system_info
                    workerNodes[message.address].timestamp = message.timestamp
                    console.log("updated wroker load : ", workerNodes[message.address])
                    workerHeap.updateItem(workerNodes[message.address])
                    // updateColdstartOnNIC();
                }
                else{
                    console.log("Change in worker load is less than threshold")
                    //f
                }
                
            }
            // console.log("\nheap : ",workerHeap)
        }
    } else if (topic === constants.topics.request_dm_2_rm) {
        message = JSON.parse(message)
        console.log(message);
        
        let payload = [{
            topic: constants.topics.response_rm_2_dm,
            messages: JSON.stringify({
                    "resource_id": message.resource_id,
                    "timestamp": Date.now(),
                    "nodes": [
                        { node_id: getAddress(), port: null }]
            }),
            partition: 0
        }]
        producer.send(payload, () => {
            console.log(`Replied`)
        })
    } else if (topic === constants.topics.check_autoscale) {
        message = JSON.parse(message)
        console.log("request to autoscale : ", message);

    } else if ( topic == constants.topics.autoscale) {
        message = JSON.parse(message)
        console.log("request to scale : ", message);
        let request_id = Math.floor(Math.random() * 1000)
        // req.body.request_id = request_id
        // res.request_id = request_id
        // requestFlightQueue.set(request_id, res)
        // let payload = "HELLO"
        // payload.request_id = request_id
        // let data = payload.data
        let fid= message.functionHash.slice(9) 
        // res.data_set_time = Date.now()
        let packet = libSupport.packPacket({
            chain_id: 0,
            exec_id: request_id,
            function_id: fid,
            data,
            function_count: 1,
            autostart: 1
        })
        // res.pack_time = Date.now()
        udpProxy.send(packet, 0, packet.length, "8000", "192.168.2.2", function (err, bytes) {
            logger.info(`forwarded request via UDP, IP 192.168.2.2 Port 8000`)
            res.send_time = Date.now()
        })
        // udpProxy.send(packet, 0, packet.length, resource.port, resource.node_id, function (err, bytes) {
        //     logger.info(`forwarded request via UDP, IP 192.168.2.5 Port ${resource.port}`)
        //     res.send_time = Date.now()
        // })
    } else if (topic == constants.topics.deployed) {
        message = JSON.parse(message)
        //fuction is deployed successfully
        // if function map availabe add it in the heap
        // if not availabe create new heap map
        // let id = message.functionHash + message.runtime
        // if (resourceMap.has(message.resource_id)) {
        //     let resource = resourceMap.get(message.resource_id)
        //     resource.node_id = message.node_id.trim()
        // }
    
        // if (functionToResourceMap.has(id)) {
        //     let resourceHeap = functionToResourceMap.get(id)
        //     heap.push(resourceHeap, {
        //         resource_id: message.resource_id,
        //         open_request_count: 0,
        //         cpu_utilization: 0
        //     }, libSupport.compare_uti)
        //     logger.warn("Horizontally scaling up: " +
        //         JSON.stringify(functionToResource.get(id)));
    
        // } else {
        //     /**
        //     * function to resource map - holds a min heap of resources associated with a function
        //     * the min heap is sorted based on a metric [TBD] like CPU usage, request count, mem usage etc
        //     * TODO: decide on metric to use for sorting.
        //     */
        //     let resourceHeap = []
        //     heap.push(resourceHeap, {
        //         resource_id: message.resource_id,
        //         open_request_count: 0,
        //         cpu_utilization: 0
        //     }, libSupport.compare_uti)
        //     functionToResourceMap.set(id, resourceHeap)
        //     logger.warn("Creating new resource pool"
        //         + JSON.stringify(functionToResourceMap.get(id)));
    
        // }
    }


})
