var kafka       = require('kafka-node'),
    uuid        = require('uuid'),
    offsets     = require('../lib/offsets.js'),
    config      = require('../config'),
    consumers   = require('../lib/consumers.js'),
    topics      = require('../lib/topics.js'),
    log         = require('../logger.js').logger,

    getConsumerId = function (group, instanceId) {
        return group + '/' + instanceId;
    },
    getConsumer = function (group, instanceId, cb) {
        return consumers.get(group, instanceId, cb);
    },

    createConsumerInstance = function (consumer, topic) {
        return consumers.createInstance(consumer, topic);
    },

    consumerTimeoutMs = config.consumer.timoutMs,

    deleteConsumer = function (consumer, cb) {
        return consumers.delete(consumer, cb);
    },

    getMessages = function (consumer, cb) {
        return consumers.getMessages(consumer, cb);
    };


module.exports = function (app) {

    setInterval(consumers.timeout, 10000);

    app.post('/consumers/:group', function (req, res) {

        var group = req.params.group;

        var consumer = {
            group: group,
            autoOffsetReset: req.body['auto.offset.reset'],
            autoCommitEnable: req.body['auto.commit.enable']
        };
        logger.trace(consumer, 'controllers/consumers : New consumer.');
            consumers.add(consumer, function(err, data){
                if (err) {
                    logger.error({error: err}, 'unable to add consumer');
                    return res.status(500).json({error: err});
                }

                return res.json({
                    instance_id: consumer.instanceId,
                    base_uri: req.protocol + '://' + req.hostname + ':' + config.port + req.path + '/instances/' + consumer.instanceId
                });
            });
    });

    app.get('/consumers/:group/instances/:id/topics/:topic', function (req, res) {
        function retrieveMessages(consumer, retry) {
            logger.trace({consumer: consumer}, 'retrieving messages');
            return getMessages(consumer, function (err, messages){
                if (err) {
                    return res.status(500).json({
                        error: err,
                        message: 'unable to retrieve messages'
                    });
                }

                if ((!messages || messages.length === 0) && retry){
                    setTimeout(function() {
                        retrieveMessages(consumer, false);
                    }, 100);
                }

                logger.trace({messages: messages}, 'sending back messages');
                return res.json(messages);
            });
        }

        logger.trace({params: req.params}, 'controllers/consumers : getting consumer');
        getConsumer(req.params.group, req.params.id, function(err, consumer){
            var topic = req.params.topic;

            if (!consumer || err) {
                return res.status(404).json({ msg: 'Consumer not found.', error: err });
            }

            if (consumer.topics.indexOf(topic) == -1) {

                topics.exists(topic, function (err, data) {
                    if (err) {
                        return res.json({ error: 'Could not find topic ' + topic });
                    }
                    if (!consumer.instance) {
                        logger.trace('controllers/consumers : no consumer instance, creating');
                        createConsumerInstance(consumer, topic);
                        logger.trace('controllers/consumers : consumer instance created');
                    }
                    else {
                        //TODO: support adding topics
                    }
                    logger.trace('controllers/consumers : add topic to consumer topic list');
                    consumer.topics.push(req.params.topic);

                    setTimeout(function () {
                        retrieveMessages(consumer, true);
                    }, 1000);
                });

            }
            else {
                retrieveMessages(consumer, true);
            }
        });
    });

    app.post('/consumers/:group/instances/:id/offsets', function (req, res) {
        logger.trace('request received, commit offsets');
        getConsumer(req.params.group, req.params.id, function (err, consumer){
            if (err) {
                return res.status(404).json({ error: err });
            }

            logger.trace({consumer : consumer.offsetMap}, 'consumer data for offset commit');
            if (consumer.offsetMap.length === 0) {
                return res.json([]);
            }

            offsets.commitOffsets(consumer.group, consumer.offsetMap, function(err, data) {
                logger.trace('offsets commited');
                if (err) {
                    return res.status(500).json({ 'error': err });
                }

                logger.trace('sending offset response to client');
                return res.json([]);
            });
        });
    });

    app.delete('/consumers/:group/instances/:id', function (req, res) {

        getConsumer(req.params.group, req.params.id, function (err, consumer){
            if (!consumer || err) {
                return res.status(404).json({ message: 'Consumer not found.', error: err });
            }

            deleteConsumer(consumer, function () {
                return res.json({});
            });
        });

    });

};

@@ -748,31 +748,24 @@
         
-return 
 consumer
 s.delete
@@ -756,17 +756,23 @@
 consumer
-s
+Manager
 .delete(
@@ -822,36 +822,223 @@
 nction (consumer
-, cb
+) %7B%0A        var messages = consumer.messages.splice(0, consumer.messages.length);%0A%0A        if (messages.length === 0) %7B%0A            return %5B%5D;%0A        %7D%0A%0A        if (consumer.autoCommitEnable
 ) %7B%0A        retu
@@ -1037,49 +1037,451 @@
     
+ 
-return consumers.getMessages(consumer, cb
+   logger.debug(%7B consumer: consumer.id %7D, 'controllers/consumers : Autocommit.');%0A            consumer.instance.commit(true);%0A        %7D%0A        consumer.lastPoll = Date.now();%0A%0A        return messages.map(function (m) %7B%0A            return %7B%0A                topic: m.topic,%0A                partition: m.partition,%0A                offset: m.offset,%0A                key: m.key.toString(),%0A                value: m.value%0A            %7D;%0A        %7D
 );%0A 

refactor logic to get messages from consumer to method and return messages when creating consumer