[Bro-Dev] some Broker questions

[Aashish Sharma]

I am trying using BrokerStore with a master and a clone setup. Where by I was thinking of using master on manager and all the workers are clones. However, I am somewhat confused at a few things - attaching the sample policies used: 

1) I see that stores-listener.bro has clone created into it and store-connector.bro has master in it.

Does that mean the idea is to have workers run listener and manager run connector ? Which fundamentally means manager connects to the workers ?
Or this is open to 'case-by-case' basis ?

2) What exactly does "bro/event/ready" mean ? Is idea here to compartmentalize various events for various policies ?
 something like bro/event/tor-ban/balh ?  

2b) Is it right to understand that with auto_event the event will be automatically called on workers if called on manager ? 

2c) How do I trigger a clone to update the master (how often or can I trigger updates on certain conditions ? ) 

3) Since all the action happens in "event BrokerComm::outgoing_connection_established"  I don't see way to pass data to it. 

Do I need to create global variables and then use them in this event ? I mean whats a good way to "pass"/use data to this event ?

3b) How is BrokerComm::outgoing_connection_established event triggered ? Does using BrokerStore::insert in some other event also trigger the updates to master from the clone ? 

4) Somewhat whimsical issue:  Why is peer_address of string type when we have peer_port as port data type. Shouldn't peer_address be address data type ? I was hoping may be one can use dns-names thats why but I cannot seem to get that working ? 

4b) Shouldn't this event be better off as : event BrokerComm::outgoing_connection_established(p: peer)

Oh also, I see that it supports sets but seems like doesn't support tables ?

I am really liking Broker from what my current understanding is so far. Its tremendously powerful.

Thanks,
Aashish

-------------- next part --------------
const broker_port= 9999/tcp &redef ; 
redef exit_only_after_terminate = T ;


global h:  opaque of BrokerStore::Handle ; 

function dv(d: BrokerComm::Data): BrokerComm::DataVector 
{
	local rval: BrokerComm::DataVector ;
	rval[0] = d; 
	return rval ;
}

global ready: event(); 

event BrokerComm::outgoing_connection_broken(peer_address: string, peer_port: port)
{
	terminate(); 
} 

event BrokerComm::outgoing_connection_established(peer_address: string, peer_port: port, peer_name: string)
{

	local myset: set[string] = ["a", "b", "c", "d" ];
	local myvec: vector of string = [ "alpha", "beta", "gamma", "theta"] ; 

	h = BrokerStore::create_master("mystore"); 
	BrokerStore::insert(h, BrokerComm::data("one"), BrokerComm::data(110)); 
	BrokerStore::insert(h, BrokerComm::data("two"), BrokerComm::data(223)); 
	BrokerStore::insert(h, BrokerComm::data("myset"), BrokerComm::data(myset)); 
	BrokerStore::insert(h, BrokerComm::data("myvec"), BrokerComm::data(myvec)); 

	BrokerStore::increment(h, BrokerComm::data("one")); 
	BrokerStore::increment(h, BrokerComm::data("two")); 

	BrokerStore::add_to_set(h, BrokerComm::data("myset"), BrokerComm::data("e")); 
	BrokerStore::remove_from_set(h, BrokerComm::data("myset"), BrokerComm::data("b")); 
	
	BrokerStore::push_left(h, BrokerComm::data("myvec"), dv(BrokerComm::data("delta"))); 
	BrokerStore::push_right(h, BrokerComm::data("myvec"), dv(BrokerComm::data("omega"))); 

	when (local res = BrokerStore::size(h) )
	{
		print "master size", res; 
		event ready(); 
	} 	
	timeout 10 sec 
	{ print "timeout" ; } 
} 


event bro_init()
{
	BrokerComm::enable(); 
	BrokerComm::connect("127.0.0.1", broker_port, 1 secs); 
	BrokerComm::auto_event("bro/event/ready", ready); 

}
-------------- next part --------------
const broker_port: port 9999/tcp &redef; 
redef exit_only_after_terminate = T ; 

global h: opaque of BrokerStore::Handle; 
global expected_key_count = 4; 
global key_count = 0 ;

function do_lookup(key: string)
{

	when (local res = BrokerStore::lookup(h, BrokerComm::data(key)) ) 
	{ 
		++key_count ; 
		print "lookup", key, res ; 
	
		if (key_count == expected_key_count)
			terminate();
	}
	timeout 10 sec
	{
		print "timeout",key; 
	} 

}


event ready()
{
	h = BrokerStore::create_clone("mystore"); 
	
	when (local res = BrokerStore::keys(h) ) 
	{
		print "clone keys", res ; 
		do_lookup(BrokerComm::refine_to_string(BrokerComm::vector_lookup(res$result,0))) ; 
		do_lookup(BrokerComm::refine_to_string(BrokerComm::vector_lookup(res$result,1))) ; 
		do_lookup(BrokerComm::refine_to_string(BrokerComm::vector_lookup(res$result,2))) ; 
		do_lookup(BrokerComm::refine_to_string(BrokerComm::vector_lookup(res$result,3))) ; 
	}
	timeout 10 sec 	
	{ 	print "timeout"; } 

}

event bro_init()
{
	BrokerComm::enable();
	BrokerComm::subscribe_to_events("bro/event/ready"); 
	BrokerComm::listen(broker_port, "127.0.0.1");

}