// qb_kafa_tests.js
require('longjohn')

var _ = require('lodash')
  , kafka = require('kafka-node')

var qbPkg = require('qb')
  , QB = qbPkg.backend(require('../lib/backend'))

var qb1, qb2;

//var connectionString = "localhost:2181"
var connectionString = "dev.raafl.com:2181"

// If we are getting a test.done complaint, turn this on. It helps find errors
process.on('uncaughtException', function (err) {
  console.error(err.stack);
});
process.setMaxListeners(100);

var tests = exports.tests = {};

tests.setUp = function (cb) {
  qb1 = QB(
    { instance_id: 0
    , num_instances: 1
    , num_partitions: 1
    , commitInterval: 1
    , connection_string: connectionString
    , task_options: { foobar: { topic: "foobar"
                              , consumer_group: "foobarers"
                              //, key: "foo"
                              }
                    , foobar2:  { topic: "foobar"
                                , consumer_group: "foobar2ers"
                                , key: "foo"
                                }
                    }
    })

  cb()
}

tests.tearDown = function (cb) {
  qb1 && qb1.end(cb)
}

tests.basic = function basic (test) {
  test.expect(8)
  var calledFoobar = false;
  var calledFoobar2 = false;
  qb1.on('error', test.ifError)
     .can('foobar', function (task, done) {
       test.equal(task.foo, 'bar');
       calledFoobar = true;
       done();
     })
     .can('foobar2', function (task, done) {
       test.equal(task.foo, 'bar');
       calledFoobar2 = true;
       done();
     })
     .post('process')
       .use(function (type, task, next) {
         test.equal(task.foo, 'bar');
         test.equal((type == 'foobar' ? calledFoobar : calledFoobar2), true);
         next();
       })
     .on('finish', function (type, task, next) {
       if (calledFoobar && calledFoobar2) {
        setImmediate(test.done);
       }
     })
     .on('ready', function () {
       qb1.push('foobar', {foo: 'bar'}, test.ifError);
       qb1.push('foobar2', {foo: 'bar'}, test.ifError)
     })
     .start()
}

@@ -187,18 +187,16 @@
 , qb2;%0A%0A
-//
 var conn
@@ -227,16 +227,18 @@
 t:2181%22%0A
+//
 var conn
@@ -1144,24 +1144,101 @@
 .end(cb)%0A%7D%0A%0A
+// starting with a %22clean%22 topic (no offset lag) push and process 2 messages%0A
 tests.basic 
@@ -1277,17 +1277,18 @@
 .expect(
-8
+10
 )%0A  var 
@@ -1302,21 +1302,17 @@
 oobar = 
-false
+0
 ;%0A  var 
@@ -1331,13 +1331,9 @@
 2 = 
-false
+0
 ;%0A  
@@ -1398,32 +1398,67 @@
  (task, done) %7B%0A
+       console.log('foobar', task)%0A
        test.equa
@@ -1496,24 +1496,18 @@
 edFoobar
- = true;
+++
 %0A       
@@ -1563,24 +1563,60 @@
 sk, done) %7B%0A
+       console.log('foobar2', task)%0A
        test.
@@ -1663,16 +1663,10 @@
 bar2
- = true;
+++
 %0A   
@@ -1791,86 +1791,8 @@
 ');%0A
-         test.equal((type == 'foobar' ? calledFoobar : calledFoobar2), true);%0A
     
@@ -1887,16 +1887,22 @@
 dFoobar 
+=== 2 
 && calle
@@ -1909,16 +1909,22 @@
 dFoobar2
+ === 2
 ) %7B%0A    

fix some quirks with basic test