var through     = require("through");
var geoip       = require("./geoip");
var userAgent   = require("./user-agent-detector");
var url         = require("./url-disector");

var processStream = function(data, stream){
    var count = 0;
    var passStream = function(){
        count += 1;
        if(count == 3){
            stream.queue(JSON.stringify(data));
        }
    };

    geoip(data.remoteAddress, function(err, result){
        if(result){ data.geo = result;}
        passStream();
    });
    userAgent(data.userAgent, function(err, result){
        if(result){ data.userAgent = result ;}
        passStream();
    });
    url(data.referer, function(err, result){
        if(result){ data.referer = result;}
        passStream();
    });

};

module.exports = through(function write(data){
    if(typeof data !== 'string') return;
    var parsedData = JSON.parse(data);
    processStream(parsedData, this);
}, function end(){
    this.queue();
});


@@ -167,16 +167,163 @@
 tor%22);%0A%0A
+var createCallback = function(type)%7B%0A    return function(err, result)%7B%0A        if(result)%7B data%5Btype%5D = result; %7D%0A        passStream();%0A    %7D;%0A%7D;%0A%0A
 var proc
@@ -381,20 +381,43 @@
  0;%0A    
-var 
+this.data = data;%0A    this.
 passStre
@@ -430,16 +430,28 @@
 unction(
+type, result
 )%7B%0A     
@@ -587,100 +587,43 @@
 ress
-, function(err
+ 
 , 
+c
 re
-sult)%7B%0A        if(result)%7B data.geo = result;%7D%0A        passStream();%0A    %7D
+ateCallback('geo').bind(this)
 );%0A 
@@ -653,226 +653,119 @@
 gent
-, function(err
+ 
 , 
+c
 re
-sult)%7B%0A        if(result)%7B data.userAgent = result ;%7D%0A        passStream();%0A    %7D);%0A    url(data.referer, function(err, result)%7B%0A        if(result)%7B data.referer = result;%7D%0A        passStream();%0A    %7D
+ateCallback('userAgent').bind(this));%0A    url(data.referer         , createCallback('referer').bind(this)
 );%0A%0A

refactor enrichstream