module.exports = function(socketNamespaces) {
    const kafka = require("kafka-node");
    
    const client = new kafka.Client("localhost:2181");
    
    const topics = [
        {
            topic: "tcp.realtime"
        }
    ];
    /* const options = {
        autoCommit: true,
        fetchMaxWaitMs: 1000,
        fetchMaxBytes: 1024 * 1024,
        encoding: "buffer"
    }; */
    
    var options = {
        //host: 'zookeeper:2181',  // zookeeper host omit if connecting directly to broker (see kafkaHost below)
        kafkaHost: 'localhost:9092', // connect directly to kafka broker (instantiates a KafkaClient)
        zk : undefined,   // put client zk settings if you need them (see Client)
        batch: undefined, // put client batch settings if you need them (see Client)
        ssl: false, // optional (defaults to false) or tls options hash
        groupId: 'ws.realtime',
        sessionTimeout: 15000,
        // An array of partition assignment protocols ordered by preference.
        // 'roundrobin' or 'range' string for built ins (see below to pass in custom assignment protocol)
        protocol: ['roundrobin'],
       
        // Offsets to use for new groups other options could be 'earliest' or 'none' (none will emit an error if no offsets were saved)
        // equivalent to Java client's auto.offset.reset
        fromOffset: 'latest', // default
        commitOffsetsOnFirstJoin: true, // on the very first time this consumer group subscribes to a topic, record the offset returned in fromOffset (latest/earliest)
        // how to recover from OutOfRangeOffset error (where save offset is past server retention) accepts same value as fromOffset
        outOfRangeOffset: 'latest', // default
        migrateHLC: false,    // for details please see Migration section below
        migrateRolling: false,
        // Callback to allow consumers with autoCommit false a chance to commit before a rebalance finishes
        // isAlreadyMember will be false on the first connection, and true on rebalances triggered after that
        onRebalance: null//(isAlreadyMember, callback) => { callback(); } // or null
      };

    const consumer  = new kafka.ConsumerGroup(options, 'tcp.realtime');
    consumer.on("message", function (message) {
        /* Print latest offset. */
        /* var offset = new kafka.Offset(client);

        offset.fetch([{ topic: 'tcp.realtime', partition: 0, time: -1 }], function (err, data) {
                var latestOffset = data['tcp.realtime']['0'][0];
                console.log("Consumer current offset: " + latestOffset);
        }); */
        // Read string into a buffer.
        var buf = new Buffer(message.value, "binary"); 
        var decodedMessage = JSON.parse(buf.toString());
        //console.log(decodedMessage);
        
        switch (decodedMessage.namespace) {
            case 'gpsio':
                if (decodedMessage.channel.endsWith('acc')) {
                    socketNamespaces[decodedMessage.namespace]
                        .to(decodedMessage.room)
                        .emit(decodedMessage.channel, decodedMessage.data[0], decodedMessage.data[1], decodedMessage.data[2], decodedMessage.data[3]);
                }
                else {
                    socketNamespaces[decodedMessage.namespace]
                    .to(decodedMessage.room)
                    .emit(decodedMessage.channel, decodedMessage.data[0], decodedMessage.data[1], decodedMessage.data[2]);
                }
            break;
            
            case 'notifIO':
                socketNamespaces[decodedMessage.namespace]
                    .emit(decodedMessage.channel, decodedMessage.data[0]);
            break;
            
            case 'sbNotifIO':
                socketNamespaces[decodedMessage.namespace]
                    .emit(decodedMessage.channel, decodedMessage.data[0]);
            break;
            default:
            break;
        }
    });
    
    consumer.on("error", function(err) {
        console.log("error", err);
    });
    
    consumer.on('offsetOutOfRange', function (err) {
        console.log("offsetOutOfRange", err);
    });

    process.on("SIGINT", function() {
        consumer.close(true, function() {
            process.exit();
        });
    });
}