Allow serialization of closures over Broker.

anonymous-functions, their closures, can now be sent over broker.
In order to send an anonymous function the receiver must have parsed
a definition of the functon, but it need not to have been evaluated.
See testing/btest/language/closure-sending.zeek for an example of how
this can be done.

This also sends their closures as well as the closures of regular
functions.

src/Expr.cc

@@ -4322,17 +4322,48 @@ void CallExpr::ExprDescribe(ODesc* d) const
 		args->Describe(d);
 	}
 
-LambdaExpr::LambdaExpr(std::unique_ptr<function_ingredients> ingredients,
+LambdaExpr::LambdaExpr(std::unique_ptr<function_ingredients> ing,
 		       std::shared_ptr<id_list> outer_ids) : Expr(EXPR_LAMBDA)
 	{
-	this->ingredients = std::move(ingredients);
+	ingredients = std::move(ing);
 	this->outer_ids = std::move(outer_ids);
 
-	SetType(this->ingredients->id->Type()->Ref());
+	SetType(ingredients->id->Type()->Ref());
+
+	// Install a dummy version of the function globally for use only
+	// when broker provides a closure.
+	BroFunc* dummy_func = new BroFunc(
+		ingredients->id,
+		ingredients->body,
+		ingredients->inits,
+		ingredients->frame_size,
+		ingredients->priority);
+
+	dummy_func->SetOuterIDs(this->outer_ids);
+
+	// Get the body's "string" representation.
+	ODesc d;
+	dummy_func->Describe(&d);
+	const char* desc = d.Description();
+
+	// Install that in the global_scope
+	ID* id = install_ID(desc, current_module.c_str(), true, false);
+
+	// Update lamb's name
+	dummy_func->SetName(desc);
+
+	// When the id goes away it will unref v.
+	Val* v = new Val(dummy_func);
+
+	// id will unref v when its done.
+	id->SetVal(v);
+	id->SetType(ingredients->id->Type()->Ref());
+	id->SetConst();
 	}
 
 Val* LambdaExpr::Eval(Frame* f) const
 	{
+
 	BroFunc* lamb = new BroFunc(
 		ingredients->id,
 		ingredients->body,
@@ -4342,7 +4373,15 @@ Val* LambdaExpr::Eval(Frame* f) const
 
 	lamb->AddClosure(outer_ids, f);
 
-	return (new Val(lamb));
+	ODesc d;
+	lamb->Describe(&d);
+	const char* desc = d.Description();
+
+	// Set name to corresponding dummy func.
+	// Allows for lookups by the receiver.
+	lamb->SetName(desc);
+	
+	return new Val(lamb);
 	}
 
 void LambdaExpr::ExprDescribe(ODesc* d) const

src/Frame.cc

@@ -2,13 +2,16 @@
 
 #include "zeek-config.h"
 
+#include <string>
+#include <algorithm> // std::any_of
+
 #include "Frame.h"
 #include "Stmt.h"
 #include "Func.h"
 #include "Trigger.h"
 
-#include <string>
+#include "broker/Data.h"
-#include <algorithm> // std::any_of
+#include <broker/error.hh>
 
 vector<Frame*> g_frame_stack;
 
@@ -27,7 +30,7 @@ Frame::Frame(int arg_size, const BroFunc* func, const val_list* fn_args)
 	call = 0;
 	delayed = false;
 
-        is_view = false;
+	is_view = false;
 
 	Clear();
 	}
@@ -75,16 +78,23 @@ void Frame::SetElement(int n, Val* v)
 	{
 	Unref(frame[n]);
 	frame[n] = v;
-      	}
+	}
+
 void Frame::SetElement(const ID* id, Val* v)
 	{
 	SetElement(id->Offset(), v);
-       	}
+	}
 
 
-Val* Frame::GetElement(ID* id) const
+Val* Frame::GetElement(const ID* id) const
 	{
-	return this->frame[id->Offset()];
+	if (HasOuterIDs())
+		{
+		auto where = offset_map.find(std::string(id->Name()));
+		if (where != offset_map.end())
+			return frame[ where->second ];
+		}
+	return frame[id->Offset()];
 	}
 
 void Frame::AddElement(ID* id, Val* v)
@@ -155,7 +165,7 @@ Frame* Frame::Clone()
 	}
 
 Frame* Frame::SelectiveClone(id_list* selection)
-        {
+	{
 	Frame* other = new Frame(size, function, func_args);
 
 	loop_over_list(*selection, i)
@@ -164,8 +174,148 @@ Frame* Frame::SelectiveClone(id_list* selection)
 	  other->frame[current->Offset()] = this->frame[current->Offset()];
 	  }
 
-        return other;
+	return other;
-        }
+	}
+
+broker::expected<broker::data> Frame::Serialize() const
+	{
+	broker::vector rval;
+	rval.emplace_back(std::string("Frame"));
+
+	auto om = SerializeOffsetMap();
+	if ( ! om ) return broker::ec::invalid_data;
+	rval.emplace_back( *om );
+
+	for (int i = 0; i < size; ++i)
+		{
+		if ( ! frame[i] )
+			{
+			// data
+			rval.emplace_back(broker::none());
+			// type
+			rval.emplace_back(broker::none());
+			}
+		else
+			{
+			auto expected = bro_broker::val_to_data(frame[i]);
+			if ( ! expected )
+				return broker::ec::invalid_data;
+			else
+				{
+				// data
+				rval.emplace_back(std::move(*expected));
+				// type
+				rval.emplace_back(static_cast<broker::integer>(frame[i]->Type()->Tag()));
+				}
+			}
+		}
+
+	return {std::move(rval)};
+	}
+
+std::pair<bool, Frame*> Frame::Unserialize(const broker::vector& data)
+	{
+	#define FAIL std::make_pair(false, nullptr)
+	#define GET_OR_RETURN(type, name, index)							\
+		if (auto __##name##__ = broker::get_if<type>(data[index]))		\
+			name = *__##name##__;										\
+		else															\
+			return FAIL;												\
+
+	std::string pivot;
+	GET_OR_RETURN(std::string, pivot, 0)
+
+	if (pivot == "Frame")
+		{
+
+		int frame_size = (data.size() - 2) / 2;
+		// Cool -> We serialized a function with a null frame.
+		if (frame_size == 0) return std::make_pair(true, nullptr);
+
+		// Unserialize the offset map.
+		broker::vector o_map;
+		GET_OR_RETURN(broker::vector, o_map, 1)
+
+		std::unordered_map<std::string, int> offset_map;
+		bool status = ClosureFrame::UnserializeIntoOffsetMap(o_map, offset_map);
+
+		// Function / arg information updated later as needed.
+		Frame* f = new Frame(frame_size, nullptr, nullptr);
+		f->offset_map = std::move(offset_map);
+
+		for (int i = 0, j = 2; i < frame_size; ++i, j += 2)
+			{
+			// Null values in the serialized frame are stored as broker::none.
+			if ( ! broker::get_if<broker::none>(data[j]) )
+				{
+				broker::integer g;
+				GET_OR_RETURN(broker::integer, g, (j+1))
+
+				BroType t( static_cast<TypeTag>(g) );
+
+				auto val = bro_broker::data_to_val(std::move(data[j]), &t);
+				if ( ! val ) return FAIL;
+
+				f->frame[i] = val;
+				}
+			}
+
+		return std::make_pair(true, f);
+		}
+
+	else if (pivot == "ClosureFrame")
+		{
+
+		broker::vector o_map;
+		broker::vector v_closure;
+		broker::vector v_body;
+
+		GET_OR_RETURN(broker::vector, o_map, 1)
+		GET_OR_RETURN(broker::vector, v_closure, 2)
+		GET_OR_RETURN(broker::vector, v_body, 3)
+
+		std::unordered_map<std::string, int> offset_map;
+		bool status = ClosureFrame::UnserializeIntoOffsetMap(o_map, offset_map);
+
+		if ( ! status ) return FAIL;
+
+		auto result = Frame::Unserialize(v_closure);
+		if ( ! result.first )
+			return FAIL;
+		Frame* closure = result.second;
+
+		result = Frame::Unserialize(v_body);
+		if ( ! result.first )
+			return FAIL;
+		Frame* body = result.second;
+
+		ClosureFrame* c = new ClosureFrame(closure, body, nullptr);
+		c->offset_map = std::move(offset_map);
+
+		return std::make_pair(true, c);
+		}
+
+	return FAIL;
+	#undef GET_OR_RETURN
+	#undef FAIL
+	}
+
+void Frame::SetOuterIDs (std::shared_ptr<id_list> outer_ids)
+	{
+	// When cloning we bypass this step and just directly copy over the map,
+	// hence the check.
+	if ( ! outer_ids ) return;
+
+	if (offset_map.size()) return;
+
+	id_list tmp = *(outer_ids.get());
+	loop_over_list(tmp, i)
+	  {
+	    ID* id = tmp[i];
+	    if (id)
+	      offset_map.emplace(id->Name(), id->Offset());
+	  }
+	}
 
 void Frame::SetTrigger(Trigger* arg_trigger)
 	{
@@ -183,8 +333,14 @@ void Frame::ClearTrigger()
 	trigger = 0;
 	}
 
+bool Frame::CaptureContains(const ID* i) const
+	{
+	auto where = offset_map.find(std::string(i->Name()));
+	return where != offset_map.end();
+	}
+
 ClosureFrame::ClosureFrame(Frame* closure, Frame* not_closure,
-			   std::shared_ptr<id_list> outer_ids) : Frame(not_closure, true)
+	std::shared_ptr<id_list> outer_ids) : Frame(not_closure, true)
 	{
 	assert(closure);
 	assert(outer_ids);
@@ -192,19 +348,7 @@ ClosureFrame::ClosureFrame(Frame* closure, Frame* not_closure,
 	this->closure = closure;
 	body = not_closure;
 	
-	// To clone a ClosureFrame we null outer_ids and then copy
+	SetOuterIDs(outer_ids);
-	// the set over directly, hence the check.
-	if (outer_ids)
-		{
-		// Install the closure IDs
-		id_list* tmp = outer_ids.get();
-		loop_over_list(*tmp, i)
-			{
-			ID* id = (*tmp)[i];
-			if (id)
-			  closure_elements.push_back(id->Name());
-			}
-		}
 	}
 
 ClosureFrame::~ClosureFrame()
@@ -215,18 +359,23 @@ ClosureFrame::~ClosureFrame()
 	Unref(body);
 	}
 
-Val* ClosureFrame::GetElement(ID* id) const
+Val* ClosureFrame::GetElement(const ID* id) const
 	{
 	if ( CaptureContains(id) )
-	      return ClosureFrame::GatherFromClosure(this, id);
+	  {
-
+	  int my_offset = offset_map.at(std::string(id->Name()));
+	  return ClosureFrame::GatherFromClosure(this, id, my_offset);
+	  }
 	return this->NthElement(id->Offset());
 	}
 
 void ClosureFrame::SetElement(const ID* id, Val* v)
 	{
 	if ( CaptureContains(id) )
-	  ClosureFrame::SetInClosure(this, id, v);
+	  {
+	  int my_offset = offset_map.at(std::string(id->Name()));
+	  ClosureFrame::SetInClosure(this, id, v, my_offset);
+	  }
 	else
 	  this->Frame::SetElement(id->Offset(), v);
 	}
@@ -237,12 +386,12 @@ Frame* ClosureFrame::Clone()
 	Frame* new_regular = body->Clone();
 
 	ClosureFrame* cf = new ClosureFrame(new_closure, new_regular, nullptr);
-	cf->closure_elements = closure_elements;
+	cf->offset_map = offset_map;
 	return cf;
 	}
 
 Frame* ClosureFrame::SelectiveClone(id_list* choose)
-        {
+	{
 	id_list us;
 	// and
 	id_list them;
@@ -261,50 +410,102 @@ Frame* ClosureFrame::SelectiveClone(id_list* choose)
 	Frame* you  = this->body->SelectiveClone(&them);
 
 	ClosureFrame* who = new ClosureFrame(me, you, nullptr);
-	who->closure_elements = closure_elements;
+	who->offset_map = offset_map;
 
 	return who;
 	}
 
+broker::expected<broker::data> ClosureFrame::Serialize() const
+	{
+	broker::vector rval;
+	rval.emplace_back(std::string("ClosureFrame"));
+
+	auto om = SerializeOffsetMap();
+	if ( ! om ) return broker::ec::invalid_data;
+	rval.emplace_back( *om );
+
+	auto cl = closure->Serialize();
+	if ( ! cl ) broker::ec::invalid_data;
+	rval.emplace_back( *cl );
+
+	auto bo = body->Serialize();
+	if ( ! bo ) broker::ec::invalid_data;
+	rval.emplace_back( *bo );
+
+	return {std::move(rval)};
+	}
+
+broker::expected<broker::data> Frame::SerializeOffsetMap() const
+	{
+	broker::vector rval;
+
+	std::for_each(offset_map.begin(), offset_map.end(), 
+		[&rval] (const std::pair<std::string, int>& e) 
+			{ rval.emplace_back(e.first); rval.emplace_back(e.second);});
+
+	return {std::move(rval)};
+	}
+
+bool ClosureFrame::UnserializeIntoOffsetMap(const broker::vector& data, std::unordered_map<std::string, int>& target)
+	{
+	#define GET_OR_RETURN(type, name, index)							\
+		if (auto __##name##__ = broker::get_if<type>(data[index]))		\
+			name = *__##name##__;										\
+		else															\
+			return false;												\
+
+	assert(target.size() == 0);
+
+	std::unordered_map<std::string, int> rval;
+
+	for (broker::vector::size_type i = 0; i < data.size(); i += 2)
+		{
+		std::string key;
+		int offset;
+		GET_OR_RETURN(std::string, key, i)
+		GET_OR_RETURN(broker::integer, offset, i+1)
+		target.insert( {std::move(key), std::move(offset)} );
+		}
+
+	return true;
+	#undef GET_OR_RETURN
+	}
+
 // Each ClosureFrame knows all of the outer IDs that are used inside of it. This is known at
 // parse time. These leverage that. If frame_1 encloses frame_2 then the location of a lookup
 // for an outer id in frame_2 can be determined by checking if that id is also an outer id in
 // frame_2. If it is not, then frame_2 owns the id and the lookup is done there, otherwise, 
 // go deeper.
-
+Val* ClosureFrame::GatherFromClosure(const Frame* start, const ID* id, const int offset)
-// Note the useage of dynamic_cast.
-
-
-Val* ClosureFrame::GatherFromClosure(const Frame* start, const ID* id)
 	{
 	const ClosureFrame* conductor = dynamic_cast<const ClosureFrame*>(start);
 
+	// If a subframe has outer IDs then it was serialized and passed around before this frame
+	// was born. We differ to its maping as it is older and wiser. Otherwise, we use our own.
 	if ( ! conductor )
-		return start->NthElement(id->Offset());
+		{
+		if (start->HasOuterIDs())
+			return start->GetElement(id);
+
+		return start->NthElement(offset);
+		}
 
 	if (conductor->CaptureContains(id))
-		return ClosureFrame::GatherFromClosure(conductor->closure, id);
+		return ClosureFrame::GatherFromClosure(conductor->closure, id, offset);
 
-	return conductor->NthElement(id->Offset());
+	return conductor->NthElement(offset);
-        }
+	}
 
-void ClosureFrame::SetInClosure(Frame* start, const ID* id, Val* val)
+void ClosureFrame::SetInClosure(Frame* start, const ID* id, Val* val, const int offset)
-        {
+	{
 	ClosureFrame* conductor = dynamic_cast<ClosureFrame*>(start);
 
 	if ( ! conductor )
-	  start->SetElement(id->Offset(), val);
+	  start->SetElement(offset, val);
 
 	else if (conductor->CaptureContains(id))
-	  ClosureFrame::SetInClosure(conductor->closure, id, val);
+	  ClosureFrame::SetInClosure(conductor->closure, id, val, offset);
 
 	else
-	  conductor->Frame::SetElement(id->Offset(), val);
+	  conductor->Frame::SetElement(offset, val);
 	}
-
-bool ClosureFrame::CaptureContains(const ID* i) const
-  {
-    const char* target = i->Name();
-    return std::any_of(closure_elements.begin(), closure_elements.end(),
-		      [target](const char* in) { return strcmp(target, in) == 0; });
-  }

src/Frame.h

@@ -4,30 +4,35 @@
 #define frame_h
 
 #include <vector>
+#include <unordered_map>
+#include <string>
 #include <memory> // std::shared_ptr
+#include <utility> // std::pair
+
+#include <broker/data.hh>
+#include <broker/expected.hh>
 
 #include "Val.h"
 
-using namespace std;
-
 class BroFunc;
 class Trigger;
 class CallExpr;
 class Val;
 
 class Frame : public BroObj {
+friend class BroFunc;
 public:
 	Frame(int size, const BroFunc* func, const val_list *fn_args);
-        // Constructs a copy or view of other. If a view is constructed the
+	// Constructs a copy or view of other. If a view is constructed the
-        // destructor will not change other's state on deletion.
+	// destructor will not change other's state on deletion.
-        Frame(const Frame* other, bool is_view = false);
+	Frame(const Frame* other, bool is_view = false);
 	~Frame() override;
 
 	Val* NthElement(int n) const { return frame[n]; }
-        void SetElement(int n, Val* v);
+	void SetElement(int n, Val* v);
-        virtual void SetElement(const ID* id, Val* v);
+	virtual void SetElement(const ID* id, Val* v);
 
-	virtual Val* GetElement(ID* id) const;
+	virtual Val* GetElement(const ID* id) const;
 	void AddElement(ID* id, Val* v);
 
 	void Reset(int startIdx);
@@ -56,9 +61,48 @@ public:
 
 	// Deep-copies values.
 	virtual Frame* Clone();
-	// Only deep-copies values corresponding to requested IDs.
+
+	/**
+	 * Clones this frame, only copying values corresponding to IDs in
+	 * *selection*. All other values are null.
+	 *
+	 * @param selection a list of IDs that will be cloned into the new
+	 * frame.
+	 * @return a new frame with the requested values and ref count +1
+	 */
 	virtual Frame* SelectiveClone(id_list* selection);
 
+	/**
+	 * Serializes the Frame into a Broker representation.
+	 *
+	 * @return the broker representaton, or an error if the serialization
+	 * failed.
+	*/
+	virtual broker::expected<broker::data> Serialize() const;
+
+	/**
+	 * Instantiates a Frame from a serialized one.
+	 *
+	 * @return a pair. the first item is the status of the serialization,
+	 * the second is the Unserialized frame with reference count +1
+	 */
+	static std::pair<bool, Frame*> Unserialize(const broker::vector& data);
+
+	/**
+	 * Installs *outer_ids* in this Frame's offset_map. 
+	 * 
+	 * Note: This needs to be done before serializing a Frame to guarantee that 
+	 * the unserialized frame will perform lookups properly.
+	 * 
+	 * @param outer_ids the ids that this frame holds
+	 */
+	void SetOuterIDs(std::shared_ptr<id_list> outer_ids);
+
+	/**
+	 * @return does this frame have an initialized offset_map?
+	 */
+	bool HasOuterIDs() const { return offset_map.size(); }
+
 	// If the frame is run in the context of a trigger condition evaluation,
 	// the trigger needs to be registered.
 	void SetTrigger(Trigger* arg_trigger);
@@ -75,6 +119,21 @@ public:
 protected:
 	void Clear();
 
+	/**
+	 * Does offset_map contain an offset corresponding to *i*?
+	 *
+	 * @param i the ID to check for.
+	 * @return true of offset_map has an offset for i, false otherwise.
+	 */
+	bool CaptureContains(const ID* i) const;
+
+	/**
+	 * Serializes this Frame's offset map.
+	 * 
+	 * @return a serialized version of the offset map.
+	 */
+	broker::expected<broker::data> SerializeOffsetMap() const;
+
 	Val** frame;
 	int size;
 
@@ -89,10 +148,23 @@ protected:
 	const CallExpr* call;
 	bool delayed;
 
-        // For ClosureFrames, we don't want a Frame as much as we want a frame that
+	/** 
-        // is a view to another underlying one. Rather or not a Frame is a view
+	 * Maps ID names to the offsets they had when passed into the frame.
-        // impacts how the Frame handles deleting itself.
+	 * 
-        bool is_view;
+	 * A frame that has been serialized maintains its own map between IDs and
+	 * their offsets. This is because a serialized frame is not guaranteed to 
+	 * be unserialized somewhere where the offsets for the IDs that it contains
+	 * are the same.
+	 */
+	std::unordered_map<std::string, int> offset_map;
+
+private:
+
+	/** 
+	 * Rather or not this frame is a view of another one. Frames that
+	 * are views do not delete their underlying frame on deletion.
+	 */
+	bool is_view;
 };
 
 
@@ -103,28 +175,57 @@ protected:
  */
 class ClosureFrame : public Frame {
 public:
-	ClosureFrame(Frame* closure, Frame* body,
+	/**
-		std::shared_ptr<id_list> outer_ids);
+	 * Constructs a closure Frame from a closure and body frame, and a list of ids
+	 * that this frame should refer to its closure to for values. For non closure
+	 * related operations the ClosureFrame is just a view of the body frame.
+	 *
+	 * @param closure the frame that holds IDs in *outer_ids*.
+	 * @param body the frame to refer to for all non-closure actions.
+	 * @param outer_ids a list of ids that have been captured by the ClosureFrame.
+	 * These inform the closure on where to refer get and set operations.
+	 */
+	ClosureFrame(Frame* closure, Frame* body, std::shared_ptr<id_list> outer_ids);
 	~ClosureFrame() override;
-	Val* GetElement(ID* id) const override;
+
+	Val* GetElement(const ID* id) const override;
 	void SetElement(const ID* id, Val* v) override;
+
 	Frame* Clone() override;
-        Frame* SelectiveClone(id_list* selection) override;
+	Frame* SelectiveClone(id_list* selection) override;
+
+	broker::expected<broker::data> Serialize() const override;
+	static bool UnserializeIntoOffsetMap
+		(const broker::vector& data, std::unordered_map<std::string, int>& target);
 
 private:
+
+	/**
+	 * Finds the Value corresponding to *id* in the closure of *start*.
+	 * 
+	 * @param start the frame to begin the search from
+	 * @param id the ID whose corresponding value is to be collected.
+	 * @param offset the offset at which to look for id's value when its
+	 * frame has been found.
+	 * @return the Value corresponding to *id*.
+	 */
+	static Val* GatherFromClosure(const Frame* start, const ID* id, const int offset);
+
+	/**
+	 * Sets the Value corresponding to *id* in the closure of *start* to *val*
+	 *
+	 * @param start the frame to begin the search from
+	 * @param val the Value to associate with *id* in the closure.
+	 * @param id the ID whose corresponding value is to be updated.
+	 * @param offset the offset at which to look for id's value when its
+	 * frame has been found.
+	 */
+	static void SetInClosure(Frame* start, const ID* id, Val* val, const int offset);
+
 	Frame* closure;
 	Frame* body;
-
-        // Both of these assume that it has already been verified that id is
-        // in the start frame.
-        static Val* GatherFromClosure(const Frame* start, const ID* id);
-        static void SetInClosure(Frame* start, const ID* id, Val* val);
-
-        bool CaptureContains(const ID* i) const;
-
-        std::vector<const char*> closure_elements;
 };
 
-extern vector<Frame*> g_frame_stack;
+extern std::vector<Frame*> g_frame_stack;
 
 #endif

src/Func.cc

@@ -28,6 +28,7 @@
 #include <signal.h>
 
 #include <algorithm>
+#include <broker/error.hh>
 
 #include "Base64.h"
 #include "Stmt.h"
@@ -266,8 +267,7 @@ std::pair<bool, Val*> Func::HandlePluginResult(std::pair<bool, Val*> plugin_resu
 	}
 
 BroFunc::BroFunc(ID* arg_id, Stmt* arg_body, id_list* aggr_inits,
-		int arg_frame_size, int priority)
+		int arg_frame_size, int priority) : Func(BRO_FUNC)
-: Func(BRO_FUNC)
 	{
 	name = arg_id->Name();
 	type = arg_id->Type()->Ref();
@@ -285,14 +285,13 @@ BroFunc::BroFunc(ID* arg_id, Stmt* arg_body, id_list* aggr_inits,
 BroFunc::~BroFunc()
 	{
 	std::for_each(bodies.begin(), bodies.end(), [](Body& b) { Unref(b.stmts); });
-
 	Unref(closure);
 	}
 
 int BroFunc::IsPure() const
 	{
 	return std::all_of(bodies.begin(), bodies.end(),
-			   [](const Body& b) { return b.stmts->IsPure(); });
+		[](const Body& b) { return b.stmts->IsPure(); });
 	}
 
 Val* BroFunc::Call(val_list* args, Frame* parent) const
@@ -328,7 +327,9 @@ Val* BroFunc::Call(val_list* args, Frame* parent) const
 
 	Frame* f = new Frame(frame_size, this, args);
 	if ( closure )
+		{
 		f = new ClosureFrame(this->closure, f, this->outer_ids);
+		}
 
 	// Hand down any trigger.
 	if ( parent )
@@ -495,6 +496,22 @@ void BroFunc::SetClosureFrame(Frame* f)
 	closure = f ? f->SelectiveClone( outer_ids.get() ) : nullptr;
 	}
 
+bool BroFunc::UpdateClosure(const broker::vector& data)
+	{
+	auto result = Frame::Unserialize(data);
+	if ( ! result.first ) return false;
+
+	Frame* new_closure = result.second;
+
+	if (new_closure)
+		new_closure->function = this;
+
+	if (closure) Unref(closure);
+	closure = new_closure;
+
+	return true;
+	}
+
 Func* BroFunc::DoClone()
 	{
 	// A BroFunc could hold a closure. In this case a clone of it must
@@ -518,6 +535,21 @@ Func* BroFunc::DoClone()
 		}
 	}
 
+broker::expected<broker::data> BroFunc::SerializeClosure() const
+	{
+	if (! closure) 
+		{
+		return broker::vector({"Frame"});
+		}
+
+	closure->SetOuterIDs(outer_ids);
+
+	auto e = closure->Serialize();
+
+	if ( !e ) return broker::ec::invalid_data;
+	return *e;
+	}
+
 void BroFunc::Describe(ODesc* d) const
 	{
 	d->Add(Name());

src/Func.h

@@ -6,11 +6,13 @@
 #include <utility>
 #include <memory> // std::shared_ptr, std::unique_ptr
 
+#include <broker/data.hh>
+#include <broker/expected.hh>
+
 #include "BroList.h"
 #include "Obj.h"
 #include "Debug.h"
 #include "Frame.h"
-// #include "Val.h"
 
 class Val;
 class ListExpr;
@@ -20,8 +22,6 @@ class Frame;
 class ID;
 class CallExpr;
 
-// struct CloneState;
-
 class Func : public BroObj {
 public:
 
@@ -74,8 +74,8 @@ public:
 
 protected:
 	Func();
-        // Copies this functions state into other.
+	// Copies this functions state into other.
-        void CopyStateInto(Func* other) const;
+	void CopyStateInto(Func* other) const;
 
 	// Helper function for handling result of plugin hook.
 	std::pair<bool, Val*> HandlePluginResult(std::pair<bool, Val*> plugin_result, val_list* args, function_flavor flavor) const;
@@ -96,15 +96,43 @@ public:
 	~BroFunc() override;
 
 	int IsPure() const override;
+
 	Val* Call(val_list* args, Frame* parent) const override;
 
-        void AddClosure(std::shared_ptr<id_list> ids, Frame* f);
+	/**
+	 * Adds adds a closure to the function. Closures are cloned and 
+	 * future calls to BroFunc will not modify *f*
+	 * 
+	 * @param ids IDs that are captured by the closure.
+	 * @param f the closure to be captured.
+	 */
+	void AddClosure(std::shared_ptr<id_list> ids, Frame* f);
+
+	/**
+	 * Replaces the current closure with one built from *data*
+	 * 
+	 * @param data a serialized closure
+	 */
+	bool UpdateClosure(const broker::vector& data);
+
 	void AddBody(Stmt* new_body, id_list* new_inits, int new_frame_size,
 			int priority) override;
 
+	/**
+	 * Clones this function along with its closures.
+	 */
 	Func* DoClone() override;
 
-	int FrameSize() const {	return frame_size; }
+	/**
+	 * Serializes this function's closure.
+	 * 
+	 * @return a serialized version of the function's closure.
+	 */
+	broker::expected<broker::data> SerializeClosure() const;
+
+	/** Sets this function's outer_id list. */
+	void SetOuterIDs(std::shared_ptr<id_list> ids)
+		{ outer_ids = std::move(ids); }
 
 	void Describe(ODesc* d) const override;
 
@@ -115,18 +143,17 @@ protected:
 	int frame_size;
 
 private:
-	// Makes a deep copy of the input frame and captures it.
+	/**
+	 * Performs a selective clone of *f* using the IDs that were
+	 * captured in the function's closure.
+	 * 
+	 * @param f the frame to be cloned.
+	 */
 	void SetClosureFrame(Frame* f);
 
-	void SetOuterIDs(std::shared_ptr<id_list> ids)
+	// List of the outer IDs used in the function.
-		{ outer_ids = std::move(ids); }
-
-        // List of the outer IDs used in the function. Shared becase other instances
-        // would like to use it as well.
 	std::shared_ptr<id_list> outer_ids = nullptr;
-	// The frame the Func was initialized in. This is not guaranteed to be
+	// The frame the BroFunc was initialized in.
-	// initialized and should be handled with care.
-        // A BroFunc Unrefs its closure on deletion.
 	Frame* closure = nullptr;
 };
 

src/Scope.cc

@@ -172,7 +172,7 @@ ID* install_ID(const char* name, const char* module_name,
 	if ( is_export || ! module_name ||
 	     (is_global &&
 	      normalized_module_name(module_name) == GLOBAL_MODULE_NAME) )
-		scope = SCOPE_GLOBAL;
+		    scope = SCOPE_GLOBAL;
 	else if ( is_global )
 		scope = SCOPE_MODULE;
 	else

src/broker/Data.cc

@@ -104,28 +104,6 @@ struct val_converter {
 
 			return nullptr;
 			}
-		case TYPE_FUNC:
-			{
-			auto id = global_scope()->Lookup(a.data());
-
-			if ( ! id )
-				return nullptr;
-
-			auto rval = id->ID_Val();
-
-			if ( ! rval )
-				return nullptr;
-
-			auto t = rval->Type();
-
-			if ( ! t )
-				return nullptr;
-
-			if ( t->Tag() != TYPE_FUNC )
-				return nullptr;
-
-			return rval->Ref();
-			}
 		default:
 			return nullptr;
 		}
@@ -364,6 +342,55 @@ struct val_converter {
 
 			return rval;
 			}
+		else if (type->Tag() == TYPE_FUNC)
+			{
+			#define GET_OR_RETURN(type, name, index)							\
+				if (auto __##name##__ = broker::get_if<type>(a[index]))			\
+					name = *__##name##__;										\
+				else															\
+					return nullptr;												\
+
+			if (a.size() > 2)
+				return nullptr;
+			
+			std::string name;
+			GET_OR_RETURN(std::string, name, 0);
+
+			auto id = global_scope()->Lookup(name.c_str());
+
+			if ( ! id )
+				return nullptr;
+
+			auto rval = id->ID_Val();
+
+			if ( ! rval )
+				return nullptr;
+
+			auto t = rval->Type();
+
+			if ( ! t )
+				return nullptr;
+
+
+			if ( t->Tag() != TYPE_FUNC )
+				return nullptr;
+
+			if (a.size() == 2) // We have a closure.
+				{
+				broker::vector frame;
+				GET_OR_RETURN(broker::vector, frame, 1);
+
+				bool status = false;
+				if ( BroFunc* b = dynamic_cast<BroFunc*>(rval->AsFunc()))
+					{
+					status = b->UpdateClosure(frame);
+					}
+				if ( ! status ) return nullptr;
+				}
+
+			return rval->Ref();
+			#undef GET_OR_RETURN
+			}
 		else if ( type->Tag() == TYPE_RECORD )
 			{
 			auto rt = type->AsRecordType();
@@ -479,28 +506,6 @@ struct type_checker {
 			return true;
 		case TYPE_FILE:
 			return true;
-		case TYPE_FUNC:
-			{
-			auto id = global_scope()->Lookup(a.data());
-
-			if ( ! id )
-				return false;
-
-			auto rval = id->ID_Val();
-
-			if ( ! rval )
-				return false;
-
-			auto t = rval->Type();
-
-			if ( ! t )
-				return false;
-
-			if ( t->Tag() != TYPE_FUNC )
-				return false;
-
-			return true;
-			}
 		default:
 			return false;
 		}
@@ -696,6 +701,40 @@ struct type_checker {
 					return false;
 				}
 
+			return true;
+			}
+		else if ( type->Tag() == TYPE_FUNC )
+			{
+			#define GET_OR_RETURN(type, name, index)							\
+				if (auto __##name##__ = broker::get_if<type>(a[index]))			\
+					name = *__##name##__;										\
+				else															\
+					return false;												\
+
+			if (a.size() > 2)
+				return false;
+			
+			std::string name;
+			GET_OR_RETURN(std::string, name, 0);
+
+			auto id = global_scope()->Lookup(name.c_str());
+
+			if ( ! id )
+				return false;
+
+			auto rval = id->ID_Val();
+
+			if ( ! rval )
+				return false;
+
+			auto t = rval->Type();
+
+			if ( ! t )
+				return false;
+
+			if ( t->Tag() != TYPE_FUNC )
+				return false;
+
 			return true;
 			}
 		else if ( type->Tag() == TYPE_RECORD )
@@ -818,13 +857,13 @@ broker::expected<broker::data> bro_broker::val_to_data(Val* v)
 		return {v->AsDouble()};
 	case TYPE_TIME:
 		{
-	  auto secs = broker::fractional_seconds{v->AsTime()};
+		auto secs = broker::fractional_seconds{v->AsTime()};
-	  auto since_epoch = std::chrono::duration_cast<broker::timespan>(secs);
+		auto since_epoch = std::chrono::duration_cast<broker::timespan>(secs);
 		return {broker::timestamp{since_epoch}};
 		}
 	case TYPE_INTERVAL:
 		{
-	  auto secs = broker::fractional_seconds{v->AsInterval()};
+		auto secs = broker::fractional_seconds{v->AsInterval()};
 		return {std::chrono::duration_cast<broker::timespan>(secs)};
 		}
 	case TYPE_ENUM:
@@ -841,7 +880,31 @@ broker::expected<broker::data> bro_broker::val_to_data(Val* v)
 	case TYPE_FILE:
 		return {string(v->AsFile()->Name())};
 	case TYPE_FUNC:
-		return {string(v->AsFunc()->Name())};
+	        {
+			Func* f = v->AsFunc();
+			std::string name(f->Name());
+			broker::vector rval;
+			rval.push_back(name);
+
+			auto starts_with = [](const std::string& s ,const std::string& in) -> bool 
+				{ return (0 == s.find(in)); };
+
+			if ( starts_with(name, "anonymous-function") )
+				{
+				// Only BroFuncs have closures.
+				if (BroFunc* b = dynamic_cast<BroFunc*>(f))
+					{
+					auto bc = dynamic_cast<BroFunc*>(f)->SerializeClosure();
+					if ( ! bc )
+						return broker::ec::invalid_data;
+					rval.emplace_back(std::move(*bc));
+					}
+				else
+					return broker::ec::invalid_data;
+				}
+
+			return {std::move(rval)};
+			}
 	case TYPE_TABLE:
 		{
 		auto is_set = v->Type()->IsSet();
@@ -995,6 +1058,8 @@ RecordVal* bro_broker::make_data_val(Val* v)
 	auto rval = new RecordVal(BifType::Record::Broker::Data);
 	auto data = val_to_data(v);
 
+	if  ( ! data ) reporter->Warning("Didn't get a value from val_to_data");
+
 	if ( data )
 		rval->Assign(0, new DataVal(move(*data)));
 

src/broker/Manager.cc

@@ -1042,10 +1042,20 @@ void Manager::ProcessEvent(const broker::topic& topic, broker::zeek::Event ev)
 			vl.append(val);
 		else
 			{
+			const char* expected_name = type_name(expected_type->Tag());
+
 			reporter->Warning("failed to convert remote event '%s' arg #%d,"
-			                  " got %s, expected %s",
+						" got %s, expected %s",
-			                  name.data(), i, got_type,
+						name.data(), i, got_type,
-			                  type_name(expected_type->Tag()));
+						expected_name);
+
+			// If we got a vector and expected a function this is possibly because of a mismatch
+			// between anonymous-function bodies.
+			if ( (strcmp( expected_name, "func") == 0) && (strcmp("vector", got_type) ==  0) )
+				{
+				reporter->Warning("when sending functions the receiver must have access to a"
+				" version of that function.\nFor anonymous functions, that function must have the same body.");
+				}
 			break;
 			}
 		}

src/parse.y

@@ -1234,11 +1234,8 @@ anonymous_function:
 
 		'}'
 			{
-			// Every time a new LambdaExpr is evaluated it must return a new instance
+			// Gather the ingredients for a BroFunc from the current scope
-			// of a BroFunc. Here, we collect the ingredients for a function and give
-			// it to our LambdaExpr.
 			std::unique_ptr<function_ingredients> ingredients = gather_function_ingredients($5);
-			
 			std::shared_ptr<id_list> outer_ids = gather_outer_ids(pop_scope(), $5);
 
 			$$ = new LambdaExpr(std::move(ingredients), std::move(outer_ids));

testing/btest/Baseline/language.closure-sending-naming/recv.recv.error

@@ -0,0 +1,4 @@
+warning: failed to convert remote event 'ping' arg #1, got vector, expected func
+warning: when sending functions the receiver must have access to a version of that function.
+For anonymous functions, that function must have the same body.
+received termination signal

testing/btest/Baseline/language.closure-sending-naming/recv.recv.out

@@ -0,0 +1 @@
+peer added

testing/btest/Baseline/language.closure-sending-naming/send.send.out

@@ -0,0 +1,2 @@
+peer added
+peer lost

testing/btest/Baseline/language.closure-sending/recv.recv.out

@@ -0,0 +1,12 @@
+peer added
+receiver got ping: function 2
+begin: 100 | base_step: 1
+begin: 100 | base_step: 1 | step: 76
+177
+receiver got ping: function 1
+inside: 2 | outside: 11 | global: 100
+78
+receiver got ping: function 2
+begin: 100 | base_step: 3
+begin: 100 | base_step: 3 | step: 76
+179

testing/btest/Baseline/language.closure-sending/send.send.out

@@ -0,0 +1,12 @@
+peer added
+begin: 100 | base_step: 50
+sender got pong: function 2
+begin: 100 | base_step: 1
+begin: 100 | base_step: 1 | step: 76
+177
+begin: 100 | base_step: 50
+sender got pong: function 1
+inside: 2 | outside: 11 | global: 10
+78
+begin: 100 | base_step: 50
+peer lost

testing/btest/Baseline/language.function-sending/recv.recv.out

@@ -0,0 +1,11 @@
+peer added
+receiver got ping: my-message, myfunc\x0a{ \x0aprint fmt(myfunc(%s), c);\x0a}
+myfunc(1)
+receiver got ping: my-message, myfunc\x0a{ \x0aprint fmt(myfunc(%s), c);\x0a}
+myfunc(2)
+receiver got ping: my-message, myfunc\x0a{ \x0aprint fmt(myfunc(%s), c);\x0a}
+myfunc(3)
+receiver got ping: my-message, myfunc\x0a{ \x0aprint fmt(myfunc(%s), c);\x0a}
+myfunc(4)
+receiver got ping: my-message, myfunc\x0a{ \x0aprint fmt(myfunc(%s), c);\x0a}
+myfunc(5)

testing/btest/Baseline/language.function-sending/send.send.out

@@ -0,0 +1,10 @@
+peer added
+sender got pong: my-message, myfunc\x0a{ \x0aprint fmt(bodiesdontsend(%s), c);\x0a}
+bodiesdontsend(1)
+sender got pong: my-message, myfunc\x0a{ \x0aprint fmt(bodiesdontsend(%s), c);\x0a}
+bodiesdontsend(2)
+sender got pong: my-message, myfunc\x0a{ \x0aprint fmt(bodiesdontsend(%s), c);\x0a}
+bodiesdontsend(3)
+sender got pong: my-message, myfunc\x0a{ \x0aprint fmt(bodiesdontsend(%s), c);\x0a}
+bodiesdontsend(4)
+peer lost

testing/btest/language/closure-sending-naming.zeek

@@ -0,0 +1,127 @@
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-EXEC: btest-bg-run recv "zeek -B broker -b ../recv.zeek >recv.out 2>recv.error"
+# @TEST-EXEC: btest-bg-run send "zeek -B broker -b ../send.zeek >send.out"
+#
+# @TEST-EXEC: btest-bg-wait 20
+# @TEST-EXEC: btest-diff recv/recv.error
+# @TEST-EXEC: btest-diff recv/recv.out
+# @TEST-EXEC: btest-diff send/send.out
+
+@TEST-START-FILE send.zeek
+
+redef exit_only_after_terminate = T;
+type myfunctype: function(c: count) : function(d: count) : count;
+
+global global_with_same_name = 10;
+
+global ping: event(msg: string, f: myfunctype);
+
+event zeek_init()
+    {
+    Broker::subscribe("zeek/event/my_topic");
+    Broker::peer("127.0.0.1", 9999/tcp);
+    }
+
+global n = 0;
+
+function send_event()
+    {
+    # log fails to be looked up because of a missing print statment
+    local log : myfunctype = function(c: count) : function(d: count) : count
+        {
+        # print fmt("inside: %s | outside: %s | global: %s", c, event_count, global_with_same_name);
+        return function(d: count) : count { return d + c; };
+        };
+
+        local e2 = Broker::make_event(ping, "function 1", log);
+        Broker::publish("zeek/event/my_topic", e2);
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer added";
+    send_event();
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer lost";
+    terminate();
+    }
+
+event pong(msg: string, f: myfunctype)
+    {
+    print fmt("sender got pong: %s", msg);
+    local adder = f(n);
+    print adder(76);
+    send_event();
+    }
+
+@TEST-END-FILE
+
+@TEST-START-FILE recv.zeek
+
+redef exit_only_after_terminate = T;
+const events_to_recv = 1;
+type myfunctype: function(c: count) : function(d: count) : count;
+
+global global_with_same_name = 100;
+
+global pong: event(msg: string, f: myfunctype);
+
+# This is one, of many, ways to declare your functions that you plan to receive.
+# All you are doing is giving the parser a version of their body, so they can be
+# anywhere. This seems to work quite nicely because it keeps them scoped and stops
+# them from ever being evaluated.
+function my_funcs()
+    {
+    return;
+
+    local event_count = 11;
+
+    local l : myfunctype = function(c: count) : function(d: count) : count
+    {
+    print fmt("inside: %s | outside: %s | global: %s", c, event_count, global_with_same_name);
+    return function(d: count) : count { return d + c; };
+    };
+    }
+
+event die() { terminate(); }
+
+event zeek_init()
+    {
+    Broker::subscribe("zeek/event/my_topic");
+    Broker::listen("127.0.0.1", 9999/tcp);
+    schedule 5sec { die() };
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer added";
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer lost";
+    }
+
+global n = 0;
+
+event ping(msg: string, f: myfunctype)
+    {
+    print fmt("receiver got ping: %s", msg);
+    ++n;
+    local adder = f(n);
+    print adder(76);
+
+    if ( n == events_to_recv )
+        terminate();
+    else
+        {
+        local e = Broker::make_event(pong, msg, f);
+        Broker::publish("zeek/event/my_topic", e);
+        }
+    }
+
+@TEST-END-FILE

testing/btest/language/closure-sending.zeek

@@ -0,0 +1,161 @@
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-EXEC: btest-bg-run recv "zeek -B broker -b ../recv.zeek >recv.out"
+# @TEST-EXEC: btest-bg-run send "zeek -B broker -b ../send.zeek >send.out"
+#
+# @TEST-EXEC: btest-bg-wait 20
+# @TEST-EXEC: btest-diff send/send.out
+# @TEST-EXEC: btest-diff recv/recv.out
+
+@TEST-START-FILE send.zeek
+
+redef exit_only_after_terminate = T;
+type myfunctype: function(c: count) : function(d: count) : count;
+
+global global_with_same_name = 10;
+
+global ping: event(msg: string, f: myfunctype);
+
+event zeek_init()
+    {
+    Broker::subscribe("zeek/event/my_topic");
+    Broker::peer("127.0.0.1", 9999/tcp);
+    }
+
+global n = 0;
+
+function send_event()
+    {
+    # in this frame event_count has an offset of three.
+    # in the receiving frame it has an offset of one.
+    # this tests to ensure that id lookups are being routed properly.
+    local dog = 0;
+    local not_dog = 1;
+    local event_count = 11;
+
+    local log : myfunctype = function(c: count) : function(d: count) : count
+        {
+        print fmt("inside: %s | outside: %s | global: %s", c, event_count, global_with_same_name);
+        return function(d: count) : count { return d + c; };
+        };
+
+	local two_part_adder_maker = function (begin : count) : function (base_step : count) : function ( step : count) : count
+		{
+		return function (base_step : count) : function (step : count) : count
+			{
+                print fmt("begin: %s | base_step: %s", begin, base_step);
+				return function (step : count) : count
+					{
+                    print fmt("begin: %s | base_step: %s | step: %s", begin, base_step, step);
+					return (begin += base_step + step); }; }; };
+	
+	local l = two_part_adder_maker(100);
+	local stepper = l(50);
+
+    ++n;
+    if ( n % 2 == 0)
+        {
+        local e2 = Broker::make_event(ping, "function 1", log);
+        Broker::publish("zeek/event/my_topic", e2);
+        }
+    else
+        {
+        local e = Broker::make_event(ping, "function 2", l);
+        Broker::publish("zeek/event/my_topic", e);
+        }
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer added";
+    send_event();
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer lost";
+    terminate();
+    }
+
+event pong(msg: string, f: myfunctype)
+    {
+    print fmt("sender got pong: %s", msg);
+    local adder = f(n);
+    print adder(76);
+    send_event();
+    }
+
+@TEST-END-FILE
+
+@TEST-START-FILE recv.zeek
+
+redef exit_only_after_terminate = T;
+const events_to_recv = 3;
+type myfunctype: function(c: count) : function(d: count) : count;
+# type myfunctype: function(c: count);
+
+global global_with_same_name = 100;
+
+global pong: event(msg: string, f: myfunctype);
+
+# This is one, of many, ways to declare your functions that you plan to receive.
+# All you are doing is giving the parser a version of their body, so they can be
+# anywhere. This seems to work quite nicely because it keeps them scoped and stops
+# them from ever being evaluated.
+function my_funcs()
+    {
+    return;
+
+    local begin = 100;
+    local event_count = begin;
+
+    local l : myfunctype = function(c: count) : function(d: count) : count
+    {
+    print fmt("inside: %s | outside: %s | global: %s", c, event_count, global_with_same_name);
+    return function(d: count) : count { return d + c; };
+    };
+
+    local dog_fish = function (base_step : count) : function (step : count) : count
+        {
+print fmt("begin: %s | base_step: %s", begin, base_step); # actual formatting doesn't matter for name resolution.
+return function (step : count) : count
+    {
+                print fmt("begin: %s | base_step: %s | step: %s", begin, base_step, step);
+                return (begin += base_step + step); }; };
+    }
+
+event zeek_init()
+    {
+    Broker::subscribe("zeek/event/my_topic");
+    Broker::listen("127.0.0.1", 9999/tcp);
+    }
+
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer added";
+    }
+
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer lost";
+    }
+
+global n = 0;
+
+event ping(msg: string, f: myfunctype)
+    {
+    print fmt("receiver got ping: %s", msg);
+    ++n;
+    local adder = f(n);
+    print adder(76);
+
+    if ( n == events_to_recv )
+        terminate();
+    else
+        {
+        local e = Broker::make_event(pong, msg, f);
+        Broker::publish("zeek/event/my_topic", e);
+        }
+    }
+
+@TEST-END-FILE

testing/btest/language/function-sending.zeek

@@ -0,0 +1,88 @@
+# @TEST-PORT: BROKER_PORT
+#
+# @TEST-EXEC: btest-bg-run recv "zeek -B broker -b ../recv.zeek >recv.out"
+# @TEST-EXEC: btest-bg-run send "zeek -B broker -b ../send.zeek >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.zeek
+
+redef exit_only_after_terminate = T;
+global event_count = 0;
+type myfunctype: function(c: count);
+function myfunc(c: count)
+    {
+    print fmt("bodiesdontsend(%s)", c);
+    }
+global ping: event(msg: string, f: myfunctype);
+event zeek_init()
+    {
+    Broker::subscribe("zeek/event/my_topic");
+    Broker::peer("127.0.0.1", 9999/tcp);
+    }
+function send_event()
+    {
+    ++event_count;
+    local e = Broker::make_event(ping, "my-message", myfunc);
+    Broker::publish("zeek/event/my_topic", e);
+    }
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer added";
+    send_event();
+    }
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer lost";
+    terminate();
+    }
+event pong(msg: string, f: myfunctype)
+    {
+    print fmt("sender got pong: %s, %s", msg, f);
+    f(event_count);
+    send_event();
+    }
+
+@TEST-END-FILE
+
+@TEST-START-FILE recv.zeek
+
+redef exit_only_after_terminate = T;
+const events_to_recv = 5;
+type myfunctype: function(c: count);
+function myfunc(c: count)
+    {
+    print fmt("myfunc(%s)", c);
+    }
+global pong: event(msg: string, f: myfunctype);
+event zeek_init()
+    {
+    Broker::subscribe("zeek/event/my_topic");
+    Broker::listen("127.0.0.1", 9999/tcp);
+    }
+event Broker::peer_added(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer added";
+    }
+event Broker::peer_lost(endpoint: Broker::EndpointInfo, msg: string)
+    {
+    print "peer lost";
+    }
+global n = 0;
+event ping(msg: string, f: myfunctype)
+    {
+    print fmt("receiver got ping: %s, %s", msg, f);
+    ++n;
+    f(n);
+    if ( n == events_to_recv )
+        terminate();
+    else
+        {
+        local e = Broker::make_event(pong, msg, f);
+        Broker::publish("zeek/event/my_topic", e);
+        }
+    }
+
+@TEST-END-FILE