lib/api.js
function(cb) {
    // return early if already connected
    if (_socket) {
      return cb(null, {host: _options.host, port: _options.port});
    }

    _socket = net.createConnection(_options.port, _options.host);

    _socket.on('connect', function() {
      if (cb) {
        cb(null, {host: _options.host, port: _options.port});
      }
    });

    _socket.on('data', function(buf) {
      var res;
      var rblock;
      var decoder;

      if (!_rcvBuf) {
        _rcvBuf = kb();
      }
      _rcvBuf.append(buf);
      res = response.decodeHead(_rcvBuf.get());
      // console.log(res)

      // buffer can have more than one msg...
      // don't want to drop msgs
      while (_rcvBuf.length() > 0 && _rcvBuf.length() >= res.size + 4) {
        decoder = _cbt.remove(res.correlation);
        rblock = decoder.decode(_rcvBuf.get(), res);

        _rcvBuf.slice(res.size + 4);

        if (_rcvBuf.length() > 0) {
          res = response.decodeHead(_rcvBuf.get());
        }

        decoder.callback(rblock.err, rblock.result);
      }
    });

    _socket.on('end', function() {});

    _socket.on('timeout', function(){
      console.log('socket timeout');
    });

    _socket.on('drain', function(){});

    _socket.on('error', function(err){
      console.log('SOCKET ERROR: ' + err);
    });

    _socket.on('close', function(){});
  }
function(cb) {
    // return early if already connected
    if (_socket) {
      return cb(null, {host: _options.host, port: _options.port});
    }

    _socket = net.createConnection(_options.port, _options.host);

    _socket.on('connect', function() {
      if (cb) {
        cb(null, {host: _options.host, port: _options.port});
      }
    });

    _socket.on('data', function(buf) {
      var res;
      var rblock;
      var decoder;

      if (!_rcvBuf) {
        _rcvBuf = kb();
      }
      _rcvBuf.append(buf);
      res = response.decodeHead(_rcvBuf.get());
      // console.log(res)

      // buffer can have more than one msg...
      // don't want to drop msgs
      while (_rcvBuf.length() > 0 && _rcvBuf.length() >= res.size + 4) {
        decoder = _cbt.remove(res.correlation);
        rblock = decoder.decode(_rcvBuf.get(), res);

        _rcvBuf.slice(res.size + 4);

        if (_rcvBuf.length() > 0) {
          res = response.decodeHead(_rcvBuf.get());
        }

        decoder.callback(rblock.err, rblock.result);
      }
    });

    _socket.on('end', function() {});

    _socket.on('timeout', function(){
      console.log('socket timeout');
    });

    _socket.on('drain', function(){});

    _socket.on('error', function(err){
      console.log('SOCKET ERROR: ' + err);
    });

    _socket.on('close', function(){});
  }
function ( cb ) { // return early if already connected if ( _socket ) { return cb ( null , { host : _options . host , port : _options . port } ) ; } _socket = net . createConnection ( _options . port , _options . host ) ; _socket . on ( 'connect' , function ( ) { if ( cb ) { cb ( null , { host : _options . host , port : _options . port } ) ; } } ) ; _socket . on ( 'data' , function ( buf ) { var res ; var rblock ; var decoder ; if ( ! _rcvBuf ) { _rcvBuf = kb ( ) ; } _rcvBuf . append ( buf ) ; res = response . decodeHead ( _rcvBuf . get ( ) ) ; // console.log(res) // buffer can have more than one msg... // don't want to drop msgs while ( _rcvBuf . length ( ) > 0 && _rcvBuf . length ( ) >= res . size + 4 ) { decoder = _cbt . remove ( res . correlation ) ; rblock = decoder . decode ( _rcvBuf . get ( ) , res ) ; _rcvBuf . slice ( res . size + 4 ) ; if ( _rcvBuf . length ( ) > 0 ) { res = response . decodeHead ( _rcvBuf . get ( ) ) ; } decoder . callback ( rblock . err , rblock . result ) ; } } ) ; _socket . on ( 'end' , function ( ) { } ) ; _socket . on ( 'timeout' , function ( ) { console . log ( 'socket timeout' ) ; } ) ; _socket . on ( 'drain' , function ( ) { } ) ; _socket . on ( 'error' , function ( err ) { console . log ( 'SOCKET ERROR: ' + err ) ; } ) ; _socket . on ( 'close' , function ( ) { } ) ; }
tearup the connection to kafka
tearup the connection to kafka