Merge remote-tracking branch 'origin/topic/jsiwek/broker-misc'

* origin/topic/jsiwek/broker-misc:
  Add Broker::publish_and_relay BIF
  Document variable argument list BIFs using ellipsis
  Support unserializing broker data into type 'any'
  Fix a bug in broker data type-casting check
  Remove dead code in broker data/val conversion function

CHANGES

@@ -1,4 +1,43 @@
 
+2.5-641 | 2018-06-04 09:18:59 -0700
+
+  * Add Broker::publish_and_relay BIF
+
+    Like Broker::relay, except the relaying-node also calls event handlers. (Corelight)
+
+  * Document variable argument list BIFs using ellipsis. (Corelight).
+
+  * Support unserializing broker data into type 'any'
+
+    The receiver side will wrap the data as a Broker::Data value, which
+    can then be type-checked/cast via 'is' or 'as' operators to a specific
+    Bro type. For example:
+
+    Sender:
+
+        Broker::publish("topic", my_event, "hello")
+
+    Receiver:
+
+        event my_event(arg: any)
+            {
+            if ( arg is string )
+                print arg as string;
+            }
+
+     (Corelight)
+
+  * Fix a bug in broker data type-casting check (Corelight)
+
+  * Remove dead code in broker data/val conversion function (Corelight)
+
+  * SSH protocol now assesses the packet length at an earlier stage within binpac
+    (Andrew Woodford).
+
+  * Remove some UTF-8 characters that snuck into a few scripts. (Corelight)
+
+  * Decrypt the krb ticket and extract authentication data. (Julien Wallior)
+
 2.5-619 | 2018-06-01 11:29:15 -0500
 
   * Relocate temporary script coverage files (Corelight)

VERSION

@@ -1 +1 @@
-2.5-619
+2.5-641

aux/broker

@@ -1 +1 @@
-Subproject commit da4f84a4cf9921298910e5b61214f32de27c632d
+Subproject commit 81cf863bb26c39b88f6cf6d1d8439458a1586bee

src/Type.cc

@@ -1185,7 +1185,14 @@ void RecordType::DescribeFieldsReST(ODesc* d, bool func_args) const
 		if ( d->FindType(td->type) )
 			d->Add("<recursion>");
 		else
-			td->DescribeReST(d);
+			{
+			if ( num_fields == 1 && streq(td->id, "va_args") &&
+			     td->type->Tag() == TYPE_ANY )
+				// This was a BIF using variable argument list
+				d->Add("...");
+			else
+				td->DescribeReST(d);
+			}
 
 		if ( func_args )
 			continue;

src/broker/Data.cc

@@ -48,7 +48,6 @@ struct val_converter {
 	using result_type = Val*;
 
 	BroType* type;
-	bool require_log_attr;
 
 	result_type operator()(broker::none)
 		{
@@ -379,9 +378,6 @@ struct val_converter {
 
 			for ( auto i = 0u; i < static_cast<size_t>(rt->NumFields()); ++i )
 				{
-				if ( require_log_attr && ! rt->FieldDecl(i)->FindAttr(ATTR_LOG) )
-					continue;
-
 				if ( idx >= a.size() )
 					{
 					Unref(rval);
@@ -774,9 +770,12 @@ struct type_checker {
 		}
 };
 
-Val* bro_broker::data_to_val(broker::data d, BroType* type, bool require_log_attr)
+Val* bro_broker::data_to_val(broker::data d, BroType* type)
 	{
-	return broker::visit(val_converter{type, require_log_attr}, std::move(d));
+	if ( type->Tag() == TYPE_ANY )
+		return bro_broker::make_data_val(move(d));
+
+	return broker::visit(val_converter{type}, std::move(d));
 	}
 
 broker::expected<broker::data> bro_broker::val_to_data(Val* v)
@@ -1132,12 +1131,12 @@ broker::data& bro_broker::opaque_field_to_data(RecordVal* v, Frame* f)
 
 bool bro_broker::DataVal::canCastTo(BroType* t) const
 	{
-	return broker::visit(type_checker{type}, data);
+	return broker::visit(type_checker{t}, data);
 	}
 
 Val* bro_broker::DataVal::castTo(BroType* t)
 	{
-	return data_to_val(data, t, false);
+	return data_to_val(data, t);
 	}
 
 IMPLEMENT_SERIAL(bro_broker::DataVal, SER_COMM_DATA_VAL);

src/broker/Data.h

@@ -54,12 +54,10 @@ broker::expected<broker::data> val_to_data(Val* v);
  * Convert a Broker data value to a Bro value.
  * @param d a Broker data value.
  * @param type the expected type of the value to return.
- * @param require_log_attr if true, skip over record fields that don't have the
- * &log attribute.
  * @return a pointer to a new Bro value or a nullptr if the conversion was not
  * possible.
  */
-Val* data_to_val(broker::data d, BroType* type, bool require_log_attr = false);
+Val* data_to_val(broker::data d, BroType* type);
 
 /**
  * Convert a Bro threading::Value to a Broker data value.

src/broker/Manager.cc

@@ -341,7 +341,8 @@ bool Manager::PublishEvent(string topic, RecordVal* args)
 bool Manager::RelayEvent(std::string first_topic,
                          broker::set relay_topics,
                          std::string name,
-                         broker::vector args)
+                         broker::vector args,
+                         bool handle_on_relayer)
 	{
 	if ( bstate->endpoint.is_shutdown() )
 		return true;
@@ -349,18 +350,33 @@ bool Manager::RelayEvent(std::string first_topic,
 	if ( ! bstate->endpoint.peers().size() )
 		return true;
 
-	DBG_LOG(DBG_BROKER, "Publishing relay event: %s",
+	DBG_LOG(DBG_BROKER, "Publishing %s-relay event: %s",
+			handle_on_relayer ? "handle" : "",
 	        RenderEvent(first_topic, name, args).c_str());
-	broker::bro::RelayEvent msg(std::move(relay_topics), std::move(name),
+
-	                            std::move(args));
+	if ( handle_on_relayer )
-	bstate->endpoint.publish(std::move(first_topic), std::move(msg));
+		{
+		broker::bro::HandleAndRelayEvent msg(std::move(relay_topics),
+		                                     std::move(name),
+		                                     std::move(args));
+		bstate->endpoint.publish(std::move(first_topic), std::move(msg));
+		}
+	else
+		{
+		broker::bro::RelayEvent msg(std::move(relay_topics),
+		                            std::move(name),
+		                            std::move(args));
+		bstate->endpoint.publish(std::move(first_topic), std::move(msg));
+		}
+
 	++statistics.num_events_outgoing;
 	return true;
 	}
 
 bool Manager::RelayEvent(std::string first_topic,
                          std::set<std::string> relay_topics,
-                         RecordVal* args)
+                         RecordVal* args,
+                         bool handle_on_relayer)
 	{
 	if ( bstate->endpoint.is_shutdown() )
 		return true;
@@ -389,7 +405,7 @@ bool Manager::RelayEvent(std::string first_topic,
 		topic_set.emplace(std::move(t));
 
 	return RelayEvent(first_topic, std::move(topic_set), event_name,
-					  std::move(xs));
+					  std::move(xs), handle_on_relayer);
 	}
 
 bool Manager::PublishIdentifier(std::string topic, std::string id)
@@ -820,6 +836,10 @@ void Manager::DispatchMessage(broker::data msg)
 		ProcessRelayEvent(std::move(msg));
 		break;
 
+	case broker::bro::Message::Type::HandleAndRelayEvent:
+		ProcessHandleAndRelayEvent(std::move(msg));
+		break;
+
 	case broker::bro::Message::Type::LogCreate:
 		ProcessLogCreate(std::move(msg));
 		break;
@@ -907,23 +927,23 @@ void Manager::Process()
 	SetIdle(! had_input);
 	}
 
-void Manager::ProcessEvent(broker::bro::Event ev)
+
+void Manager::ProcessEvent(std::string name, broker::vector args)
 	{
-	DBG_LOG(DBG_BROKER, "Received event: %s", RenderMessage(ev).c_str());
+	DBG_LOG(DBG_BROKER, "Process event: %s %s",
-
+			name.data(), RenderMessage(args).data());
 	++statistics.num_events_incoming;
+	auto handler = event_registry->Lookup(name.data());
 
-	auto handler = event_registry->Lookup(ev.name().c_str());
 	if ( ! handler )
 		return;
 
-	auto& args = ev.args();
 	auto arg_types = handler->FType(false)->ArgTypes()->Types();
 
 	if ( static_cast<size_t>(arg_types->length()) != args.size() )
 		{
 		reporter->Warning("got event message '%s' with invalid # of args,"
-				  " got %zd, expected %d", ev.name().data(), args.size(),
+				  " got %zd, expected %d", name.data(), args.size(),
 				  arg_types->length());
 		return;
 		}
@@ -942,7 +962,7 @@ void Manager::ProcessEvent(broker::bro::Event ev)
 			{
 			reporter->Warning("failed to convert remote event '%s' arg #%d,"
 			                  " got %s, expected %s",
-			                  ev.name().data(), i, got_type,
+			                  name.data(), i, got_type,
 			                  type_name(expected_type->Tag()));
 			break;
 			}
@@ -954,6 +974,11 @@ void Manager::ProcessEvent(broker::bro::Event ev)
 		delete_vals(vl);
 	}
 
+void Manager::ProcessEvent(broker::bro::Event ev)
+	{
+	ProcessEvent(std::move(ev.name()), std::move(ev.args()));
+	}
+
 void Manager::ProcessRelayEvent(broker::bro::RelayEvent ev)
 	{
 	DBG_LOG(DBG_BROKER, "Received relay event: %s", RenderMessage(ev).c_str());
@@ -965,6 +990,18 @@ void Manager::ProcessRelayEvent(broker::bro::RelayEvent ev)
 		             std::move(ev.args()));
 	}
 
+void Manager::ProcessHandleAndRelayEvent(broker::bro::HandleAndRelayEvent ev)
+	{
+	DBG_LOG(DBG_BROKER, "Received handle-relay event: %s",
+	        RenderMessage(ev).c_str());
+	ProcessEvent(ev.name(), ev.args());
+
+	for ( auto& t : ev.topics() )
+		PublishEvent(std::move(broker::get<std::string>(t)),
+		             std::move(ev.name()),
+		             std::move(ev.args()));
+	}
+
 bool bro_broker::Manager::ProcessLogCreate(broker::bro::LogCreate lc)
 	{
 	DBG_LOG(DBG_BROKER, "Received log-create: %s", RenderMessage(lc).c_str());

src/broker/Manager.h

@@ -148,33 +148,41 @@ public:
 	bool PublishEvent(std::string topic, RecordVal* ev);
 
 	/**
-	 * Sends an event to any interested peers, who, upon receipt, immediately
+	 * Sends an event to any interested peers, who, upon receipt,
-	 * republish the event to a new set of topics.
+	 * republishes the event to a new set of topics and optionally
+	 * calls event handlers.
 	 * @param first_topic the first topic to use when publishing the event
 	 * @param relay_topics the set of topics the receivers will use to
 	 * republish the event.  The event is relayed at most a single hop.
 	 * @param name the name of the event
 	 * @param args the event's arguments
+	 * @param handle_on_relayer whether they relaying-node should call event
+	 * handlers.
 	 * @return true if the message is sent successfully.
 	 */
 	bool RelayEvent(std::string first_topic,
 					broker::set relay_topics,
 					std::string name,
-					broker::vector args);
+					broker::vector args,
+					bool handle_on_relayer);
 
 	/**
-	 * Sends an event to any interested peers, who, upon receipt, immediately
+	 * Sends an event to any interested peers, who, upon receipt,
-	 * republish the event to a new set of topics.
+	 * republishes the event to a new set of topics and optionally
+	 * calls event handlers.
 	 * @param first_topic the first topic to use when publishing the event
 	 * @param relay_topics the set of topics the receivers will use to
 	 * republish the event.  The event is relayed at most a single hop.
 	 * @param ev the event and its arguments to send to peers, in the form of
 	 * a Broker::Event record type.
+	 * @param handle_on_relayer whether they relaying-node should call event
+	 * handlers.
 	 * @return true if the message is sent successfully.
 	 */
 	bool RelayEvent(std::string first_topic,
 					std::set<std::string> relay_topics,
-					RecordVal* ev);
+					RecordVal* ev,
+					bool handle_on_relayer);
 
 	/**
 	 * Send a message to create a log stream to any interested peers.
@@ -340,8 +348,10 @@ private:
 	};
 
 	void DispatchMessage(broker::data msg);
+	void ProcessEvent(std::string name, broker::vector args);
 	void ProcessEvent(broker::bro::Event ev);
 	void ProcessRelayEvent(broker::bro::RelayEvent re);
+	void ProcessHandleAndRelayEvent(broker::bro::HandleAndRelayEvent ev);
 	bool ProcessLogCreate(broker::bro::LogCreate lc);
 	bool ProcessLogWrite(broker::bro::LogWrite lw);
 	bool ProcessIdentifierUpdate(broker::bro::IdentifierUpdate iu);

src/broker/messaging.bif

@@ -76,13 +76,13 @@ static bool relay_event_args(val_list& args, const BroString* topic,
 	if ( args[0]->Type()->Tag() == TYPE_RECORD )
 		rval = broker_mgr->RelayEvent(topic->CheckString(),
 									  std::move(topic_set),
-		                              args[0]->AsRecordVal());
+		                              args[0]->AsRecordVal(), false);
 	else
 		{
 		auto ev = broker_mgr->MakeEvent(&args, frame);
 		rval = broker_mgr->RelayEvent(topic->CheckString(),
 									  std::move(topic_set),
-									  ev);
+									  ev, false);
 		Unref(ev);
 		}
 
@@ -133,7 +133,7 @@ function Broker::publish%(topic: string, ...%): bool
 
 ## Publishes an event at a given topic, with any receivers automatically
 ## forwarding it to its peers with a different topic.  The event is relayed
-## at most a single hop.
+## at most a single hop and the relayer does not call any local event handlers.
 ##
 ## first_topic: the initial topic to use for publishing the event.
 ##
@@ -181,12 +181,74 @@ function Broker::relay%(first_topic: string, ...%): bool
 	if ( args[0]->Type()->Tag() == TYPE_RECORD )
 		rval = broker_mgr->RelayEvent(first_topic->CheckString(),
 									  std::move(topic_set),
-	                                  args[0]->AsRecordVal());
+	                                  args[0]->AsRecordVal(), false);
 	else
 		{
 		auto ev = broker_mgr->MakeEvent(&args, frame);
 		rval = broker_mgr->RelayEvent(first_topic->CheckString(),
-									  std::move(topic_set), ev);
+									  std::move(topic_set), ev, false);
+		Unref(ev);
+		}
+
+	return new Val(rval, TYPE_BOOL);
+	%}
+
+## Publishes an event at a given topic, with any receivers automatically
+## forwarding it to its peers with a different topic.  The event is relayed
+## at most a single hop and the relayer does call local event handlers.
+##
+## first_topic: the initial topic to use for publishing the event.
+##
+## args: the first member of the argument list may be either a string or
+##       a set of strings indicating the secondary topic that the first
+##       set of receivers will use to re-publish the event.  The remaining
+##       members of the argument list are either the return value of a
+##       previously-made call to :bro:see:`Broker::make_event` or the
+##       argument list that should be passed along to it, so that it can
+##       be called as part of executing this function.
+##
+## Returns: true if the message is sent.
+function Broker::publish_and_relay%(first_topic: string, ...%): bool
+	%{
+	bro_broker::Manager::ScriptScopeGuard ssg;
+	val_list* bif_args = @ARGS@;
+
+	if ( bif_args->length() < 3 )
+		{
+		builtin_error("Broker::publish_and_relay requires at least 3 arguments");
+		return new Val(false, TYPE_BOOL);
+		}
+
+	auto second_topic = (*bif_args)[1];
+
+	if ( second_topic->Type()->Tag() != TYPE_STRING &&
+		 ! is_string_set(second_topic->Type()) )
+		{
+		builtin_error("Broker::publish_and_relay requires a string or string_set as 2nd argument");
+		return new Val(false, TYPE_BOOL);
+		}
+
+	auto topic_set = val_to_topic_set(second_topic);
+
+	if ( topic_set.empty() )
+		return new Val(false, TYPE_BOOL);
+
+	val_list args(bif_args->length() - 2);
+
+	for ( auto i = 2; i < bif_args->length(); ++i )
+		args.append((*bif_args)[i]);
+
+	auto rval = false;
+
+	if ( args[0]->Type()->Tag() == TYPE_RECORD )
+		rval = broker_mgr->RelayEvent(first_topic->CheckString(),
+									  std::move(topic_set),
+	                                  args[0]->AsRecordVal(), true);
+	else
+		{
+		auto ev = broker_mgr->MakeEvent(&args, frame);
+		rval = broker_mgr->RelayEvent(first_topic->CheckString(),
+									  std::move(topic_set), ev, true);
 		Unref(ev);
 		}
 

testing/btest/Baseline/broker.remote_event_any/recv.recv.out

@@ -0,0 +1,12 @@
+receiver added peer: endpoint=127.0.0.1 msg=handshake successful
+is_remote should be T, and is, T
+receiver got ping: my-message, 1
+is_remote should be T, and is, T
+receiver got ping: my-message, 2
+is_remote should be T, and is, T
+receiver got ping: my-message, 3
+is_remote should be T, and is, T
+receiver got ping: my-message, 4
+is_remote should be T, and is, T
+receiver got ping: my-message, 5
+[num_peers=1, num_stores=0, num_pending_queries=0, num_events_incoming=5, num_events_outgoing=4, num_logs_incoming=0, num_logs_outgoing=1, num_ids_incoming=0, num_ids_outgoing=0]

testing/btest/Baseline/broker.remote_event_any/send.send.out

@@ -0,0 +1,11 @@
+is_remote should be F, and is, F
+sender added peer: endpoint=127.0.0.1 msg=received handshake from remote core
+is_remote should be T, and is, T
+sender got pong: my-message, 1
+is_remote should be T, and is, T
+sender got pong: my-message, 2
+is_remote should be T, and is, T
+sender got pong: my-message, 3
+is_remote should be T, and is, T
+sender got pong: my-message, 4
+sender lost peer: endpoint=127.0.0.1 msg=lost remote peer

testing/btest/Baseline/broker.remote_publish_and_relay_event/one.one.out

@@ -0,0 +1,3 @@
+sender added peer: endpoint=127.0.0.1 msg=received handshake from remote core
+got ready event
+sender lost peer: endpoint=127.0.0.1 msg=lost remote peer

testing/btest/Baseline/broker.remote_publish_and_relay_event/three.three.out

@@ -0,0 +1,2 @@
+receiver added peer: endpoint=127.0.0.1 msg=handshake successful
+got my_event, hello world

testing/btest/Baseline/broker.remote_publish_and_relay_event/two.two.out

@@ -0,0 +1,5 @@
+receiver added peer: endpoint=127.0.0.1 msg=received handshake from remote core
+receiver added peer: endpoint=127.0.0.1 msg=handshake successful
+sending ready event
+got my_event, hello world
+receiver lost peer: endpoint=127.0.0.1 msg=lost remote peer

testing/btest/broker/remote_event_any.bro

@@ -0,0 +1,107 @@
+# @TEST-SERIALIZE: comm
+#
+# @TEST-EXEC: btest-bg-run recv "bro -B broker -b ../recv.bro >recv.out"
+# @TEST-EXEC: btest-bg-run send "bro -B broker -b ../send.bro >send.out"
+#
+# @TEST-EXEC: btest-bg-wait 20
+# @TEST-EXEC: btest-diff recv/recv.out
+# @TEST-EXEC: btest-diff send/send.out
+
+@TEST-START-FILE send.bro
+
+redef Broker::default_connect_retry=1secs;
+redef Broker::default_listen_retry=1secs;
+redef exit_only_after_terminate = T;
+
+global event_count = 0;
+
+global ping: event(msg: string, c: any);
+
+event bro_init()
+    {
+    Broker::subscribe("bro/event/my_topic");
+    Broker::peer("127.0.0.1");
+    print "is_remote should be F, and is", is_remote_event();
+    }
+
+function send_event()
+    {
+    ++event_count;
+    local e = Broker::make_event(ping, "my-message", event_count);
+    Broker::publish("bro/event/my_topic", e);
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("sender added peer: endpoint=%s msg=%s",
+    endpoint$network$address, msg);
+    send_event();
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("sender lost peer: endpoint=%s msg=%s",
+    endpoint$network$address, msg);
+    terminate();
+    }
+
+event pong(msg: string, n: any)
+    {
+    print "is_remote should be T, and is", is_remote_event();
+
+    if ( n is count )
+        print fmt("sender got pong: %s, %s", msg, n as count);
+
+    send_event();
+    }
+
+@TEST-END-FILE
+
+
+@TEST-START-FILE recv.bro
+
+redef Broker::default_connect_retry=1secs;
+redef Broker::default_listen_retry=1secs;
+redef exit_only_after_terminate = T;
+
+const events_to_recv = 5;
+
+global handler: event(msg: string, c: count);
+global auto_handler: event(msg: string, c: count);
+
+global pong: event(msg: string, c: count);
+
+event bro_init()
+        {
+        Broker::subscribe("bro/event/my_topic");
+        Broker::listen("127.0.0.1");
+        }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+        {
+        print fmt("receiver added peer: endpoint=%s msg=%s", endpoint$network$address, msg);
+        }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+        {
+        print fmt("receiver lost peer: endpoint=%s msg=%s", endpoint$network$address, msg);
+        }
+
+event ping(msg: string, n: any)
+        {
+        print "is_remote should be T, and is", is_remote_event();
+
+        if ( n is count )
+        	print fmt("receiver got ping: %s, %s", msg, n as count);
+
+        if ( (n as count) == events_to_recv )
+                {
+		print get_broker_stats();
+                terminate();
+                return;
+                }
+
+        Broker::publish("bro/event/my_topic", pong, msg, n as count);
+        }
+
+@TEST-END-FILE

testing/btest/broker/remote_publish_and_relay_event.bro

@@ -0,0 +1,125 @@
+# @TEST-SERIALIZE: comm
+#
+# @TEST-EXEC: btest-bg-run three "bro -B broker -b ../three.bro >three.out"
+# @TEST-EXEC: btest-bg-run two "bro -B broker -b ../two.bro >two.out"
+# @TEST-EXEC: btest-bg-run one "bro -B broker -b ../one.bro >one.out"
+#
+# @TEST-EXEC: btest-bg-wait 20
+# @TEST-EXEC: btest-diff one/one.out
+# @TEST-EXEC: btest-diff two/two.out
+# @TEST-EXEC: btest-diff three/three.out
+
+@TEST-START-FILE one.bro
+
+redef Broker::default_connect_retry=1secs;
+redef Broker::default_listen_retry=1secs;
+redef exit_only_after_terminate = T;
+
+event my_event(s: string)
+	{
+	print "got my_event", s;
+	}
+
+event ready_event()
+	{
+	print "got ready event";
+
+	Broker::publish_and_relay("bro/event/pre-relay", "bro/event/post-relay",
+	                          my_event, "hello world");
+	}
+
+event bro_init()
+    {
+    Broker::subscribe("bro/event/ready");
+    Broker::peer("127.0.0.1", 10000/tcp);
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("sender added peer: endpoint=%s msg=%s",
+    		  endpoint$network$address, msg);
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("sender lost peer: endpoint=%s msg=%s",
+    		  endpoint$network$address, msg);
+    terminate();
+    }
+
+@TEST-END-FILE
+
+
+@TEST-START-FILE two.bro
+
+redef Broker::default_connect_retry=1secs;
+redef Broker::default_listen_retry=1secs;
+redef exit_only_after_terminate = T;
+
+global peers_added = 0;
+
+event my_event(s: string)
+	{
+	print "got my_event", s;
+	}
+
+event ready_event()
+	{
+	}
+
+event bro_init()
+    {
+    Broker::subscribe("bro/event/pre-relay");
+    Broker::listen("127.0.0.1", 10000/tcp);
+	Broker::peer("127.0.0.1", 9999/tcp);
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("receiver added peer: endpoint=%s msg=%s", endpoint$network$address, msg);
+    ++peers_added;
+
+    if ( peers_added == 2 )
+    	{
+		print "sending ready event";
+        Broker::publish("bro/event/ready", ready_event);
+    	}
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("receiver lost peer: endpoint=%s msg=%s", endpoint$network$address, msg);
+    terminate();
+    }
+
+@TEST-END-FILE
+
+@TEST-START-FILE three.bro
+
+redef Broker::default_connect_retry=1secs;
+redef Broker::default_listen_retry=1secs;
+redef exit_only_after_terminate = T;
+
+event my_event(s: string)
+	{
+	print "got my_event", s;
+	terminate();
+	}
+
+event bro_init()
+    {
+    Broker::subscribe("bro/event/post-relay");
+    Broker::listen("127.0.0.1", 9999/tcp);
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("receiver added peer: endpoint=%s msg=%s", endpoint$network$address, msg);
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print fmt("receiver lost peer: endpoint=%s msg=%s", endpoint$network$address, msg);
+    }
+
+@TEST-END-FILE