var debug = require('debug')('kafka');
var config, client, producer;

var DEBUG = process.env.DEBUG && (process.env.DEBUG != "undefined");

function send(message, topic, callback) {
    if (!producer || !topic ) return;
    //if (DEBUG) debug("start send data:" + message);
    if (!callback){
        callback =  function (err, data) {
            if (err) {
                console.error("send data to kafka error:" + err);
            }else if (DEBUG){
                debug("send data success!");
            }
        }
    }

    //attributes controls compression of the message set. (default: 0)
    //0: No compression
    //1: Compress using GZip
    //2: Compress using snappy

    if  (message instanceof Array){
        producer.send([
            {topic: topic, messages: message, attributes: 1}
        ], callback);
    }else{
        producer.send([
            {topic: topic, messages: [message], attributes: 1}
        ], callback);
    }
}


function init(c) {
    var kafka = require('kafka-node');
    var HighLevelProducer = kafka.HighLevelProducer;
    var Client = kafka.Client;
    config = c;
    client = new Client(config.connectionString);
    producer = new HighLevelProducer(client);

    producer.on('error', function (err) {
        console.error('kafka log error:', err)
    })

    producer.on('ready', function () {
        if (DEBUG) debug("connect kafka!");
    });
}

module.exports.send = send;
module.exports.init = init;
module.exports.config = function(){ return config};
