distcomponents.js

// this does data collection + inserting , processing and inserting using p2p components self discovery and connection
//
// this is not complete it missing removing of sockets when disconnected and limiting zmq high water limit to something small or and maybe to reconect the same socket on connection loss.// maybe add predictable unique names to sockets of componets so data will reflow by zmq correctly on reconnect. maybe to maintain a starting point to reprocess from and load from database.
//
// to run this you need zmq.js and distcomponents.js and npm insta  xxhashjs, and install telepathine module from git
// made on 24/6/2014 expected to run on node 0.10...

process.on('uncaughtException', function (err)
{
 run=false;
 console.log('Caught exception2: ' + (err.stack||err));
});


//collect
//inserter listens to data
//calculator

//flat=require('./flat.js');
//require('./db.js');

require('./zmq.js')
zmq_telepathine_start();



if(!global.appid)appid=Math.round(Math.random()*10000)+1;



myapp_announcer_start=function()
{
    var component_description={
	 name:'myapp_announcer',
	 inputs:  {
			   //'announcer':{'zmqport':zmqport,externalip:zmq_getExternalIp()}
			  },
			 //output not neded only needed to detect unconnected fully components which is rare case
	 outputs: {
	           //  'dbinserter':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()},
			   //         'dataprocessor':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()}
	          }
	};
	var clients={
				   //'myapp_dbinserter': 
				   //[
				    //{  url:'url://fff.com:4500',type:'announcer'  }
				   //]
	          }
	var clients_all=[];
	Object.defineProperty(component_description, "clients", { value : clients } );
	Object.defineProperty(component_description, "clients_all", { value : clients_all } );

	Object.defineProperty(component_description, "connect", { value : function(c){return connect(c)} }); //, enumerable:false is default
	Object.defineProperty(component_description, "isconnected", { value : function(c){return isconnected(c)} }); //, enumerable:false is default
	
	var clients=component_description.clients
	//var dbinserter_clients=component_description.clients.myapp_dbinserter
	
	
	function isconnected(other_component_description)
	{
	    var endpoint=other_component_description.inputs[component_description.name];
		var myexternalip=zmq_getExternalIp();
		var endpointport=endpoint.zmqport.replace(/127.0.0.1|0.0.0.0|\*/,myexternalip==endpoint.externalip?'127.0.0.1':endpoint.externalip);
		return clients_all.filter(function(zmqcon){ return zmqcon.last_endpoint==endpointport}).length>0;
    }	
	
	function connect(other_component_description)
	{
	    if(!(component_description.name in other_component_description.inputs))
		{
		 console.log(new Error("myapp_announcer_start - trying to connect wrong components. "+other_component_description.name+' doesnt have in inputs '+component_description.name+'.').stack," to ","other_component_description=",other_component_description," from " ,'component_description=',component_description)
		 return
		}
			
		//if(isconnected(other_component_description))
		//{
		// console.log(new Error("myapp_announcer_start - already connected components. "+other_component_description.name+' doesnt have in inputs '+component_description.name+'.').stack," to ","other_component_description=",other_component_description," from " ,'component_description=',component_description)
		// return
		//}
		
		
		var endpoint=other_component_description.inputs[component_description.name];
		
		var port=endpoint.zmqport;
		var myexternalip=zmq_getExternalIp();
		port=port.replace(/127.0.0.1|0.0.0.0|\*/,myexternalip==endpoint.externalip?'127.0.0.1':endpoint.externalip);
		var zmq1 = zmqconnect(port);
		if(!clients[other_component_description.name]){clients[other_component_description.name]=[];}
		clients[other_component_description.name].push(zmq1);
		clients_all.push(zmq1);
	}
	 
	//function sendclients(type,data)
	//{
	//  var ct=clients[type];	  
	//  for(var i=0;i<ct.length;i++)
	//  {
	//   ct[i].send(data);
	//  }
	//}
	
	function sendclients_dbinserter(data)
	{
	  //console.log('announcer: clients_all.len',clients_all.length);
	  //var ct=dbinserter_clients;	  
	  var ct=clients_all;	  
	  for(var i=0;i<ct.length;i++)
	  {
	   console.log('announcer: sending data to',ct[i].last_endpoint);
	   ct[i].send(data);
	  }
	}
	 
	var clientid=Math.round(Math.random()*1000)+1;
    var dedupsendstate1= {prev_send:null,count_send:0};
	
	if(global.run==undefined)run=true;
	
	// the kind of simple emiter of data to insert
	setInterval(function(){/// T
	if(!run)return;
	var d = new Date();
    var n = (d.getMinutes()*60)+d.getSeconds();
	n=Math.floor(n/3);
	sendclients_dbinserter( zmqdedupsend(['time',n],dedupsendstate1,clientid) )
	},3000);
	
	
	zmq_telepathine_addcomponent(component_description);
	
	return component_description;
}



myapp_dbinserter_start=function()
{
    var component_description={
	 name:'myapp_dbinserter',
	 inputs:  {
			   'myapp_announcer':{'zmqport':zmqport,externalip:zmq_getExternalIp()} // myapp_announcer can connect to this port
			   ,'myapp_processor':{'zmqport':zmqport,externalip:zmq_getExternalIp()} // myapp_processor can connect to this port
			  },
			 //output not neded only needed to detect unconnected fully components which is rare case
	 outputs: {
	           //  'myapp_dbinserter':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()},
			   //         'dataprocessor':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()}
	          }
	};
	
	var clients= {
				   //'myapp_dbinserter': 
				   //[
				    //{  url:'url://fff.com:4500',type:'announcer'  }
				   //]
	          }
	var clients_all=[];
	Object.defineProperty(component_description, "clients", { value : clients } );
	Object.defineProperty(component_description, "clients_all", { value : clients_all } );
				  
	Object.defineProperty(component_description, "servers", { value : []} );

	Object.defineProperty(component_description, "connect", { value : function(c){return false; return connect(c)} , enumerable:false });
	Object.defineProperty(component_description, "isconnected", { value : function(c){return false; isconnected(c)} }); //, enumerable:false is default
	
	//accept myapp_announcer:
	var dbinserterport = 'tcp://*:0';
	var zmqs_myapp_log=zmqlisten(dbinserterport,'dbinserter');
	component_description.servers.push(zmqs_myapp_log)
	component_description.inputs.myapp_announcer.zmqport=zmqs_myapp_log.last_endpoint;
	console.log("setting:",'last_component.'+component_description.name+'.myapp_announcer.peer',zmq_telepathine_self.peer_name)
	zmq_telepathine.set('last_component.'+component_description.name+'.myapp_announcer',zmqs_myapp_log.last_endpoint);
	zmq_telepathine.set('last_component.'+component_description.name+'.myapp_announcer.peer',zmq_telepathine_self.peer_name);
	
	var receivestate1={emitedh:[],emitedt:[],emitedd:[]}
	zmqs_myapp_log.on("message", function(str)
	{
	 var re=dedupreceive(str.toString(),receivestate1);
	 if(re!==undefined)
	 {
	  var masterpeername= zmq_telepathine.get('last_component.'+component_description.name+'.myapp_announcer.peer');
	  if( !( zmq_telepathine.peers[ masterpeername ].alive   || masterpeername==zmq_telepathine_self.peer_name ) )
 	  {
 	   console.log('peer dead back to us1');
	   zmq_telepathine.set('last_component.'+component_description.name+'.myapp_announcer',zmqs_myapp_log.last_endpoint);
	   zmq_telepathine.set('last_component.'+component_description.name+'.myapp_announcer.peer',zmq_telepathine_self.peer_name);
	   console.log("setting:",'last_component.'+component_description.name+'.myapp_announcer.peer',zmq_telepathine_self.peer_name);
	  }
	  if( zmq_telepathine.get('last_component.'+component_description.name+'.myapp_announcer') !=zmqs_myapp_log.last_endpoint) {console.log('i am not master1'); return;}
	  //var x=flat.flat(re);x['app']=appid;//commented out to remove flat dependency for demonstaration
	  x=re
	  //console.log('dbinsert');
      console.log('dbinsert','public.myapp_log',x);
	  //dbinsert('public.myapp_log',x);
	 }
	});
	
	
	//accept myapp_processor:
	var dbinserterport2 = 'tcp://*:0';
	var zmqs_insert=zmqlisten(dbinserterport2);
	component_description.servers.push(zmqs_insert)
	component_description.inputs.myapp_processor.zmqport=zmqs_insert.last_endpoint;
	zmq_telepathine.set('last_component.'+component_description.name+'.myapp_processor',zmqs_insert.last_endpoint);
	zmq_telepathine.set('last_component.'+component_description.name+'.myapp_processor.peer',zmq_telepathine_self.peer_name);
	var receivestate2={emitedh:[],emitedt:[],emitedd:[]}
	zmqs_insert.on("message", function(str)
	{
	 var re=dedupreceive(str.toString(),receivestate2);
	 if(re!==undefined)
	 {
	  var masterpeername= zmq_telepathine.get('last_component.'+component_description.name+'.myapp_processor.peer');
	  if( !(zmq_telepathine.peers[ masterpeername ].alive  || masterpeername==zmq_telepathine_self.peer_name ) )
 	  {
 	   console.log('peer dead back to us2');
	   zmq_telepathine.set('last_component.'+component_description.name+'.myapp_processor',zmqs_insert.last_endpoint);
	   zmq_telepathine.set('last_component.'+component_description.name+'.myapp_processor.peer',zmq_telepathine_self.peer_name);
	  }
	  if( zmq_telepathine.get('last_component.'+component_description.name+'.myapp_processor') !=zmqs_insert.last_endpoint) {console.log('i am not master2'); return;}
	 
	  //var x=flat.flat(re);x['app']=appid;
	  //console.log('dbinsert2');
      console.log('dbinsert2',re[0],re[1]);
	  //dbinsert('public.myapp_log',x);
	 }
	});
	
	
	
	zmq_telepathine_addcomponent(component_description);
	return component_description;
}



myapp_processor_start=function()
{
    var component_description={
	 name:'myapp_processor',
	 inputs:  {
			   'myapp_announcer':{'zmqport':zmqport,externalip:zmq_getExternalIp()} // myapp_announcer can connect to this port
			   //,'myapp_processor':{'zmqport':zmqport,externalip:zmq_getExternalIp()} // myapp_processor can connect to this port
			  },
			 //output not neded only needed to detect unconnected fully components which is rare case
	 outputs: {
	           //  'myapp_dbinserter':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()},
	          }
	}
	
	
	var clients= {
				   'myapp_dbinserter': 
				   [
				    //{  url:'url://fff.com:4500',type:'announcer'  }
				   ]
	          }
	var clients_all=[];
	Object.defineProperty(component_description, "clients", { value : clients } );
	Object.defineProperty(component_description, "clients_all", { value : clients_all } );
	
	Object.defineProperty(component_description, "connect", { value : function(c){return connect(c)} , enumerable:false });
	Object.defineProperty(component_description, "isconnected", { value : function(c){return isconnected(c)} }); //, enumerable:false is default
	
	var clients=component_description.clients
	var dbinserter_clients=component_description.clients.myapp_dbinserter
	
	function isconnected(other_component_description)
	{
	    var endpoint=other_component_description.inputs[component_description.name];
		var myexternalip=zmq_getExternalIp();
		var endpointport=endpoint.zmqport.replace(/127.0.0.1|0.0.0.0|\*/,myexternalip==endpoint.externalip?'127.0.0.1':endpoint.externalip);
		return clients_all.filter(function(zmqcon){ return zmqcon.last_endpoint==endpointport}).length>0;
    }
	
	function on_zmq_client_message(zmq_client,message)
	{
	 console.log('on_zmq_client_message(zmq_client,message) '+message.toString());
	}
	
    function connect(other_component_description)
	{
	    if(!(component_description.name in other_component_description.inputs))
		{
		 console.log(new Error("myapp_announcer_start - trying to connect wrong components. "+other_component_description.name+' doesnt have in inputs '+component_description.name+'.').stack," to ","other_component_description=",other_component_description," from " ,'component_description=',component_description)
		 return
		}
			
		//if(isconnected(other_component_description))
		//{
		// console.log(new Error("myapp_announcer_start - already connected components. "+other_component_description.name+' doesnt have in inputs '+component_description.name+'.').stack," to ","other_component_description=",other_component_description," from " ,'component_description=',component_description)
		// return
		//}
		
		
		var endpoint=other_component_description.inputs[component_description.name];
		
		var port=endpoint.zmqport;
		var myexternalip=zmq_getExternalIp();
		port=port.replace(/127.0.0.1|0.0.0.0|\*/,myexternalip==endpoint.externalip?'127.0.0.1':endpoint.externalip);
		var zmq1 = zmqconnect(port);
		if(!clients[other_component_description.name]){clients[other_component_description.name]=[];}
		clients[other_component_description.name].push(zmq1);
		clients_all.push(zmq1);
	}
	 
	//function sendclients(type,data)
	//{
	//  var ct=clients[type];	  
	//  for(var i=0;i<ct.length;i++)
	//  {
	//   ct[i].send(data);
	//  }
	//}
	
	function sendclients_dbinserter(data)
	{
	  var ct=dbinserter_clients;	  
	  for(var i=0;i<ct.length;i++)
	  {
	   ct[i].send(data);
	  }
	}
	
	var clientid=Math.round(Math.random()*1000)+1;
    var dedupsendstate1= {prev_send:null,count_send:0};
	
	function ex_dbinsert(t,d)
	{
	 sendclients_dbinserter( zmqdedupsend([t,d],dedupsendstate1,clientid) )
	 //sendclients_dbinserter([t,d])
	}
	
	
	var dbinserterport = 'tcp://*:0';
	var zmqs_myapp_log=zmqlisten(dbinserterport);
	component_description.inputs.myapp_announcer.zmqport=zmqs_myapp_log.last_endpoint;
	
	var receivestate1={emitedh:[],emitedt:[],emitedd:[]}
	zmqs_myapp_log.on("message", function(str)
	{
	 console.log('myapp processor: received data');
	 var re=dedupreceive(str.toString(),receivestate1);
	 if(re!==undefined)
	 {
	  //if(re[0]=='data') myapp_extract.process(re[1]); /// T
	  if(re[0]=='time') ex_dbinsert('public.time',re[1]); /// T
	  console.log('myapp processor:  data=',re);
	  //var x=flat.flat(re);x['app']=appid;
      //console.log('dbinsert','public.myapp_log',x);
	  //dbinsert('public.myapp_log',x);
	 }
	});
	zmq_telepathine_addcomponent(component_description);
	return component_description;
}



zmq_telepathine_start=function ()
{
 setTimeout(function(){
 myapp_announcer_start();
 myapp_dbinserter_start();
 myapp_processor_start();
 },7000); // time out to be able to set run=false  to stop it from emiting announces to do debug work, not requiered
}

var  repl = require("repl");repl.start({ useGlobal:true,  useColors:true, });

zmq.js

/*
 *
 * Pipeline
 *
 */


var zmq = require('zmq'),EventEmitter = require("events").EventEmitter,XXH=require('xxhashjs');


zmqport = 'tcp://127.0.0.1:7845';
zmqsockets=[];

zmqclientid="";
zmqdedupprint=false;

zmqconnect=function (zmqport)
{
      //if(clientid!==undefined)zmqclientid=clientid;
	  var socket = zmq.socket('pub'); // push = upstream
	  
	  zmqsockets.push(socket);
	  socket.identity = 'upstream' + process.pid;
	  socket.connect(zmqport);
	  console.log('zmqconnect connected!',socket.last_endpoint);
	  return socket;
}


zmqdedupsend=function(data,state,clientid,hashdata)// var state= {prev_send:null,count_send:0,zmqsend:zmqsend}
{
 //if(clientid===undefined)clientid=zmqclientid;
 //console.log('send ',data);
 var hashstring,d
 
 if(hashdata) //custom case
 {
  hashstring=JSON.stringify(hashdata);
  d=JSON.stringify(data);
 }
 else if(data&&data.a1_otimestamp) //my case
 {
  var dataclone=JSON.parse(d=JSON.stringify(data))
  dataclone.a1_otimestamp=Math.round(dataclone.a1_otimestamp/30000);
  hashstring=JSON.stringify(dataclone);
 }
 else if(data&&data.timestamp) //probable case
 {
  var dataclone=JSON.parse(d=JSON.stringify(data))
  dataclone.timestamp=Math.round(dataclone.timestamp/30000);
  hashstring=JSON.stringify(dataclone);
 }
 else // no relativly simular timestamps
 {
  d=hashstring=JSON.stringify(data);
 }
 if(state.prev_send==hashstring) state.count_send++; else  state.count_send=0;
 var h = XXH( hashstring , 0xABCD ).toString(16)+state.count_send;
 state.prev_send=hashstring;
 return clientid+' '+h+' '+d;
}
//example:
//var sendstate1= {prev_send:null,count_send:0}
//var clientid1=1
//zmqsocket.send(zmqdedupsend(object,sendstate1,clientid1));


//zmqevents = new EventEmitter();

zmqlisten=function (port,name) 
{
  var socket = zmq.socket('sub'); //pull = downstream
  socket.subscribe('');
  zmqsockets.push(socket);// for debug
  //socket.subscribe('');
  socket.identity = 'downstream' + (name||process.pid);
  
  socket.bindSync(port!==undefined?port:zmqport)
  console.log('zmqlisten - bound!',socket.last_endpoint);
  socket.port=socket.last_endpoint;
	
  //socket.on('message', function(data) {
  ////	zmqevents.emit(data.toString());
	//console.log(socket.identity + ': received data ' + data.toString(),r);
  //});
  
  return socket;
}


dedupreceive=function (str,state) { //var state={emitedh:[],emitedt:[],emitedd:[]}
   var r=str.split(' ',2);r[2]=str.substring(r[0].length+r[1].length+2, str.length);
   var client=r[0],hash=r[1],data=r[2];
   if(state.emitedh.lastIndexOf(hash)==-1)
   {
    //console.log('orig ',data);
    state.emitedt.push(new Date().getTime());
    state.emitedh.push(hash);
    //state.emitedd.push(data);
    var t=new Date().getTime();
    if(t-state.emitedt[0]>120000)
    {
     var i,emitedt=state.emitedt;
     for(i=0;i<emitedt.length&&t-emitedt[i]>60000;i++){}
	 if(zmqdedupprint) console.log('cleanup array remove '+i+' from '+emitedt.length)
  	 state.emitedt.splice(0,i);
     state.emitedh.splice(0,i);
     //state.emitedd.splice(0,i);
    }
    if(zmqdedupprint) console.log("dedupreceive ",client,hash,data);
	return JSON.parse(data);
   }
   else
   {
    if(zmqdedupprint) console.log('dedupreceive dup ',client,hash,data);
    return undefined;
   }  
}
//example
//var receivestate1={emitedh:[],emitedt:[],emitedd:[]}
//zmqsocket.on("message", function(str)
//{
// var re=dedupreceive(str.toString(),receivestate1);
// if(re!==undefined) console.log(re);
//});

//example2: other possible use  localy
//
//var receivestate1={emitedh:[],emitedt:[],emitedd:[]}
////var sendstates={};
//function onrow(clientid,object)
//{
// if(!(clientid in sendstates))sendstates[clientid]={prev_send:null,count_send:0};
// var sendstate1=sendstates[clientid]
// var re=dedupreceive(zmqdedupsend(object,sendstate1),receivestate1);
// if(re!==undefined) console.log(re);
// else  console.log('duplicate' object);
//};


//////gosip discovery

function getExternalIp() {
	var ifconfig = require('os').networkInterfaces();
	var device, i, I, protocol;
 
	for (device in ifconfig) {
		// ignore network loopback interface
		if (device.indexOf('lo') !== -1 || !ifconfig.hasOwnProperty(device)) {
			continue;
		}
		for (i=0, I=ifconfig[device].length; i<I; i++) {
			protocol = ifconfig[device][i];
 
			// filter for external IPv4 addresses
			if (protocol.family === 'IPv4' && protocol.internal === false) {
				//console.log('found', protocol.address);
				return protocol.address;
			}
		}
	}
    console.log('External Ip Not found!');
	return '127.0.0.1';
}
zmq_getExternalIp=getExternalIp;


function handleexit(cb)
{
  //zmq_gossip_remove_all()
  zmq_telepathine_self.say("bye");
  setTimeout(function(){  if(cb)cb(); },3000)//Telepathine heartBeatIntervalMS + little
}

function handleexit_at_process_on(signal)
{
 process.on(signal,function () {
 console.log('Got '+signal+', will exit in 10 seconds ');
 var num=1;
 var n=setInterval(function(){ console.log(num);   num++;  },1000); 
 var c=setTimeout(function(){ if(n)clearTimeout(n);   process.exit(0);  },10000)     
 handleexit(function(){
  if(c)clearTimeout(c);
  if(n)clearTimeout(n);
  process.exit(0);
 })
 });
}
handleexit_at_process_on('SIGTERM');
handleexit_at_process_on('SIGINT');

var net=require('net')
function getPort (portrange,cb) {
  var server = net.createServer()
  server.listen(portrange, function (err) {
    server.once('close', function () {
      cb(portrange)
    })
    server.close()
  })
  server.on('error', function (err) {
    getPort(portrange+1,cb)
  })
}
zmq_getPort=getPort;

var Telepathine = require('telepathine').Telepathine;

zmq_telepathine=null
zmq_telepathine_self=null
zmq_telepathine_start=function(toipandport)
{

	var options = {
		gossipIntervalMS: 2500,
		heartBeatIntervalMS: 2500,		
		address: getExternalIp(),
		addressMap: {'127.0.0.1': getExternalIp() }
	};

	// Create peers and point them at the seed 
	// Usually this would happen in separate processes.
	// To prevent a network's single point of failure, design with multiple seeds.

	getPort(5000,function(port){
	    var autoports=[];
		if(port!=5000)autoports.push("127.0.0.1:"+5000)
		//if(port!=5001)autoports.push("127.0.0.1:"+5001)
		//if(port!=5002)autoports.push("127.0.0.1:"+5002)
		//if(port!=5003)autoports.push("127.0.0.1:"+5003)
		var a=zmq_telepathine=new Telepathine(port,!toipandport?autoports:[toipandport ], options);
		a.on('start', function (self) {
		    zmq_telepathine_self=self;
			console.log('zmq_telepathine on port '+port);
			if(zmq_telepathine_start)zmq_telepathine_start();
		});
		a.hear('componentonline', function (data, fromPeer)
		{
		    zmq_telepathine_process_add(data);
			console.log('hear  componentonline  received ', this.event, '=', data, 'from', fromPeer);
		});	
		a.hear('componentoffline', function (data, fromPeer)
		{
			console.log('hear  componentonline  received ', this.event, '=', data, 'from', fromPeer);
		});
		a.hear('bye', function (data, fromPeer)
		{
			console.log('hear  bye  received ', this.event, '=', data, 'from', fromPeer);
			setTimeout(function (){ a.peers[fromPeer].alive=false;},10);//10 ms because the bunch of updates transmited together with a positive hearbeat event that goes right after this say event. so i want to set live=false after the heart beat , or you may find a better way to not respond to heart beat like setting some varibale to true on connect and unless its true not beleive heart beats
		});
		
		a.believe('*', function (peer, k,v,expiresAt) {
			console.log(peer  + " after believe of "+k+ ' with value '+v);
			console.log('a me '+a.peer_name + " get somekey=" + a.get('somekey'),' ~=',v,'peer=',peer,a.peers[peer]?a.getRemote(peer, k):'no remote');
		});

		//convenience method for key change events, using wildcard
		a.know('*', function (peer,k, v) {
			console.log(this.peer_name + " knows via know('*'.. that peer " + peer + " set " + this.event + "=" + v);
		});
	
		a.on('peer:new', function(peerstate) {  console.log( 'peer discovered',peerstate); })
		a.on('peer:start', function(peer_name) {console.log( 'peer seems alive - peer start',peer_name); })
		a.on('peer:stop', function(peer_name) {console.log( 'peer seems dead - peer stop',peer_name);})
		a.start();
	})

	zmq_telepathine_connect=function(toipandport)
	{
	 if (toipandport === zmq_telepathine_connect.peer_name) console.log( new Error('cannot specify self as seed').stack);
	 var x={};x[toipandport]=undefined;
	 zmq_telepathine.handleNewPeers(x);
    }
}

zmq_telepathine_process_add=function(c)
{
 var acs=zmq_telepathine_added_components;
 for(var i=0;i<acs.length;i++)
 {
  //seach in any my component needs this new component
  var cc=acs[i];
  if(c.name in cc.inputs)
  {
   console.log(''+cc.name+' discoverd that a newly added component '+c.name+' may need '+cc.name+' - say '+cc.name+' is online to all')
   //reply only to him i am here
   zmq_telepathine.say('componentonline',cc);
   zmq_telepathine_process_add(JSON.parse(JSON.stringify(cc)));
  }
  if(cc.name in c.inputs)
  {
   if(!cc.isconnected(c))
   {
    console.log(''+cc.name+' discoverd it can connect to '+c.name+' - tring to connect')
    cc.connect(c);
   }
   else
    console.log(''+cc.name+' rediscoverd it can connect to '+c.name+' - already connected')

  }
 }
}

/*
var example_component_description={
 name:'announcer',
 inputs: {
          //'announcer':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()}
		 },
		 //output not neded only needed to detect unconnected fully components which is rare case
 //outputs:{'dbinserter':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()},
 //         'dataprocessor':{'zmqport':zmqport.replace(/\*|127.0.0.1/,getExternalIp()}
 //        }
};
*/
zmq_telepathine_added_components=[];

zmq_telepathine_addcomponent=function(component_description)
{
 if(zmq_telepathine===null)
  return setTimeout(function(){zmq_telepathine_addcomponent(component_description)},500);
 
 zmq_telepathine_added_components.push(component_description);
 zmq_telepathine.say('componentonline',component_description);
 zmq_telepathine_process_add(component_description);
 //zmq_mesh.send('componentonline', );
}

//zmq_telepathine_start();