Add Broker::publish_and_relay BIF

Like Broker::relay, except the relaying-node also calls event handlers.

aux/broker

@@ -1 +1 @@
-Subproject commit 7b84848bded443637fa34e76f7d8558bd1cafbee
+Subproject commit 11c9313582b5505c002a62eba28ffb93bae1e5b3

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_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_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