Deprecate EventMgr::QueueEventFast() and update usages to Enqueue()

aux/bifcl

@@ -1 +1 @@
-Subproject commit 75f645ac9bdfd141f549b7e1a197459f2ad518be
+Subproject commit 3fefee1630269b96ea4f39021bf387b9d0abfd80

src/Anon.cc

@@ -421,12 +421,10 @@ ipaddr32_t anonymize_ip(ipaddr32_t ip, enum ip_addr_anonymization_class_t cl)
 void log_anonymization_mapping(ipaddr32_t input, ipaddr32_t output)
 	{
 	if ( anonymization_mapping )
-		{
-		mgr.QueueEventFast(anonymization_mapping, {
-			new AddrVal(input),
-			new AddrVal(output)
-		});
-		}
+		mgr.Enqueue(anonymization_mapping,
+			make_intrusive<AddrVal>(input),
+			make_intrusive<AddrVal>(output)
+		);
 	}
 
 #endif

src/DNS_Mgr.cc

@@ -704,7 +704,7 @@ void DNS_Mgr::Event(EventHandlerPtr e, DNS_Mapping* dm)
 	if ( ! e )
 		return;
 
-	mgr.QueueEventFast(e, {BuildMappingVal(dm).release()});
+	mgr.Enqueue(e, BuildMappingVal(dm));
 	}
 
 void DNS_Mgr::Event(EventHandlerPtr e, DNS_Mapping* dm,
@@ -713,11 +713,11 @@ void DNS_Mgr::Event(EventHandlerPtr e, DNS_Mapping* dm,
 	if ( ! e )
 		return;
 
-	mgr.QueueEventFast(e, {
-		BuildMappingVal(dm).release(),
-		l1->ConvertToSet(),
-		l2->ConvertToSet(),
-	});
+	mgr.Enqueue(e,
+		BuildMappingVal(dm),
+		IntrusivePtr{AdoptRef{}, l1->ConvertToSet()},
+		IntrusivePtr{AdoptRef{}, l2->ConvertToSet()}
+	);
 	}
 
 void DNS_Mgr::Event(EventHandlerPtr e, DNS_Mapping* old_dm, DNS_Mapping* new_dm)
@@ -725,10 +725,7 @@ void DNS_Mgr::Event(EventHandlerPtr e, DNS_Mapping* old_dm, DNS_Mapping* new_dm)
 	if ( ! e )
 		return;
 
-	mgr.QueueEventFast(e, {
-		BuildMappingVal(old_dm).release(),
-		BuildMappingVal(new_dm).release(),
-	});
+	mgr.Enqueue(e, BuildMappingVal(old_dm), BuildMappingVal(new_dm));
 	}
 
 IntrusivePtr<Val> DNS_Mgr::BuildMappingVal(DNS_Mapping* dm)

src/Event.cc

@@ -164,7 +164,7 @@ void EventMgr::Dispatch(Event* event, bool no_remote)
 void EventMgr::Drain()
 	{
 	if ( event_queue_flush_point )
-		QueueEventFast(event_queue_flush_point, val_list{});
+		Enqueue(event_queue_flush_point, zeek::Args{});
 
 	SegmentProfiler prof(segment_logger, "draining-events");
 

src/Event.h

@@ -62,8 +62,7 @@ public:
 	// against the case where there's no handlers (one usually also does that
 	// because it would be a waste of effort to construct all the event
 	// arguments when there's no handlers to consume them).
-	// TODO: deprecate
-	/* [[deprecated("Remove in v4.1.  Use Enqueue() instead.")]] */
+	[[deprecated("Remove in v4.1.  Use Enqueue() instead.")]]
 	void QueueEventFast(const EventHandlerPtr &h, val_list vl,
 			SourceID src = SOURCE_LOCAL, analyzer::ID aid = 0,
 			TimerMgr* mgr = 0, BroObj* obj = 0);

src/Reporter.cc

@@ -480,26 +480,28 @@ void Reporter::DoLog(const char* prefix, EventHandlerPtr event, FILE* out,
 		auto vl_size = 1 + (bool)time + (bool)location + (bool)conn +
 		               (addl ? addl->length() : 0);
 
-		val_list vl(vl_size);
+		zeek::Args vl;
+		vl.reserve(vl_size);
 
 		if ( time )
-			vl.push_back(new Val(network_time ? network_time : current_time(), TYPE_TIME));
+			vl.emplace_back(make_intrusive<Val>(network_time ? network_time : current_time(), TYPE_TIME));
 
-		vl.push_back(new StringVal(buffer));
+		vl.emplace_back(make_intrusive<StringVal>(buffer));
 
 		if ( location )
-			vl.push_back(new StringVal(loc_str.c_str()));
+			vl.emplace_back(make_intrusive<StringVal>(loc_str.c_str()));
 
 		if ( conn )
-			vl.push_back(conn->BuildConnVal());
+			vl.emplace_back(AdoptRef{}, conn->BuildConnVal());
 
 		if ( addl )
-			std::copy(addl->begin(), addl->end(), std::back_inserter(vl));
+			for ( auto v : *addl )
+				vl.emplace_back(AdoptRef{}, v);
 
 		if ( conn )
-			conn->ConnectionEventFast(event, 0, std::move(vl));
+			conn->EnqueueEvent(event, std::move(vl));
 		else
-			mgr.QueueEventFast(event, std::move(vl));
+			mgr.Enqueue(event, std::move(vl));
 		}
 	else
 		{

src/RuleAction.cc

@@ -21,13 +21,11 @@ void RuleActionEvent::DoAction(const Rule* parent, RuleEndpointState* state,
 				const u_char* data, int len)
 	{
 	if ( signature_match )
-		{
-		mgr.QueueEventFast(signature_match, {
-			rule_matcher->BuildRuleStateValue(parent, state),
-			new StringVal(msg),
-			data ? new StringVal(len, (const char*)data) : val_mgr->GetEmptyString(),
-		});
-		}
+		mgr.Enqueue(signature_match,
+			IntrusivePtr{AdoptRef{}, rule_matcher->BuildRuleStateValue(parent, state)},
+			make_intrusive<StringVal>(msg),
+			data ? make_intrusive<StringVal>(len, (const char*)data) : IntrusivePtr{AdoptRef{}, val_mgr->GetEmptyString()}
+		);
 	}
 
 void RuleActionEvent::PrintDebug()

src/Sessions.cc

@@ -123,7 +123,7 @@ void NetSessions::NextPacket(double t, const Packet* pkt)
 	SegmentProfiler prof(segment_logger, "dispatching-packet");
 
 	if ( raw_packet )
-		mgr.QueueEventFast(raw_packet, {pkt->BuildPktHdrVal()});
+		mgr.Enqueue(raw_packet, IntrusivePtr{AdoptRef{}, pkt->BuildPktHdrVal()});
 
 	if ( pkt_profiler )
 		pkt_profiler->ProfilePkt(t, pkt->cap_len);
@@ -310,7 +310,7 @@ void NetSessions::DoNextPacket(double t, const Packet* pkt, const IP_Hdr* ip_hdr
 		{
 		dump_this_packet = 1;
 		if ( esp_packet )
-			mgr.QueueEventFast(esp_packet, {ip_hdr->BuildPktHdrVal()});
+			mgr.Enqueue(esp_packet, IntrusivePtr{AdoptRef{}, ip_hdr->BuildPktHdrVal()});
 
 		// Can't do more since upper-layer payloads are going to be encrypted.
 		return;

src/Stats.cc

@@ -372,11 +372,11 @@ void SampleLogger::SegmentProfile(const char* /* name */,
 					double dtime, int dmem)
 	{
 	if ( load_sample )
-		mgr.QueueEventFast(load_sample, {
-			load_samples->Ref(),
-			new IntervalVal(dtime, Seconds),
-			val_mgr->GetInt(dmem)
-		});
+		mgr.Enqueue(load_sample,
+			IntrusivePtr{NewRef{}, load_samples},
+			make_intrusive<IntervalVal>(dtime, Seconds),
+			IntrusivePtr{AdoptRef{}, val_mgr->GetInt(dmem)}
+		);
 	}
 
 void SegmentProfiler::Init()

src/analyzer/Analyzer.cc

@@ -688,13 +688,12 @@ void Analyzer::ProtocolConfirmation(Tag arg_tag)
 		return;
 
 	EnumVal* tval = arg_tag ? arg_tag.AsEnumVal() : tag.AsEnumVal();
-	Ref(tval);
 
-	mgr.QueueEventFast(protocol_confirmation, {
-		BuildConnVal(),
-		tval,
-		val_mgr->GetCount(id),
-	});
+	mgr.Enqueue(protocol_confirmation,
+		IntrusivePtr{AdoptRef{}, BuildConnVal()},
+		IntrusivePtr{NewRef{}, tval},
+		IntrusivePtr{AdoptRef{}, val_mgr->GetCount(id)}
+	);
 	}
 
 void Analyzer::ProtocolViolation(const char* reason, const char* data, int len)
@@ -716,14 +715,13 @@ void Analyzer::ProtocolViolation(const char* reason, const char* data, int len)
 		r = new StringVal(reason);
 
 	EnumVal* tval = tag.AsEnumVal();
-	Ref(tval);
 
-	mgr.QueueEventFast(protocol_violation, {
-		BuildConnVal(),
-		tval,
-		val_mgr->GetCount(id),
-		r,
-	});
+	mgr.Enqueue(protocol_violation,
+		IntrusivePtr{AdoptRef{}, BuildConnVal()},
+		IntrusivePtr{NewRef{}, tval},
+		IntrusivePtr{AdoptRef{}, val_mgr->GetCount(id)},
+		IntrusivePtr{AdoptRef{}, r}
+	);
 	}
 
 void Analyzer::AddTimer(analyzer_timer_func timer, double t,

src/analyzer/protocol/arp/ARP.cc

@@ -191,13 +191,13 @@ void ARP_Analyzer::BadARP(const struct arp_pkthdr* hdr, const char* msg)
 	if ( ! bad_arp )
 		return;
 
-	mgr.QueueEventFast(bad_arp, {
-		ConstructAddrVal(ar_spa(hdr)),
-		EthAddrToStr((const u_char*) ar_sha(hdr)),
-		ConstructAddrVal(ar_tpa(hdr)),
-		EthAddrToStr((const u_char*) ar_tha(hdr)),
-		new StringVal(msg),
-	});
+	mgr.Enqueue(bad_arp,
+		IntrusivePtr{AdoptRef{}, ConstructAddrVal(ar_spa(hdr))},
+		IntrusivePtr{AdoptRef{}, EthAddrToStr((const u_char*) ar_sha(hdr))},
+		IntrusivePtr{AdoptRef{}, ConstructAddrVal(ar_tpa(hdr))},
+		IntrusivePtr{AdoptRef{}, EthAddrToStr((const u_char*) ar_tha(hdr))},
+		make_intrusive<StringVal>(msg)
+	);
 	}
 
 void ARP_Analyzer::Corrupted(const char* msg)
@@ -213,14 +213,14 @@ void ARP_Analyzer::RREvent(EventHandlerPtr e,
 	if ( ! e )
 		return;
 
-	mgr.QueueEventFast(e, {
-		EthAddrToStr(src),
-		EthAddrToStr(dst),
-		ConstructAddrVal(spa),
-		EthAddrToStr((const u_char*) sha),
-		ConstructAddrVal(tpa),
-		EthAddrToStr((const u_char*) tha),
-	});
+	mgr.Enqueue(e,
+		IntrusivePtr{AdoptRef{}, EthAddrToStr(src)},
+		IntrusivePtr{AdoptRef{}, EthAddrToStr(dst)},
+		IntrusivePtr{AdoptRef{}, ConstructAddrVal(spa)},
+		IntrusivePtr{AdoptRef{}, EthAddrToStr((const u_char*) sha)},
+		IntrusivePtr{AdoptRef{}, ConstructAddrVal(tpa)},
+		IntrusivePtr{AdoptRef{}, EthAddrToStr((const u_char*) tha)}
+	);
 	}
 
 AddrVal* ARP_Analyzer::ConstructAddrVal(const void* addr)

src/analyzer/protocol/pia/PIA.cc

@@ -157,12 +157,11 @@ void PIA_UDP::ActivateAnalyzer(analyzer::Tag tag, const Rule* rule)
 			{
 			// Queue late match event
 			EnumVal *tval = tag ? tag.AsEnumVal() : GetAnalyzerTag().AsEnumVal();
-			Ref(tval);
 
-			mgr.QueueEventFast(protocol_late_match, {
-			    BuildConnVal(),
-			    tval,
-			});
+			mgr.Enqueue(protocol_late_match,
+			    IntrusivePtr{AdoptRef{}, BuildConnVal()},
+			    IntrusivePtr{NewRef{}, tval}
+			);
 			}
 
 		pkt_buffer.state = dpd_late_match_stop ? SKIPPING : MATCHING_ONLY;
@@ -306,12 +305,11 @@ void PIA_TCP::ActivateAnalyzer(analyzer::Tag tag, const Rule* rule)
 			{
 			// Queue late match event
 			EnumVal *tval = tag ? tag.AsEnumVal() : GetAnalyzerTag().AsEnumVal();
-			Ref(tval);
 
-			mgr.QueueEventFast(protocol_late_match, {
-			    BuildConnVal(),
-			    tval
-			});
+			mgr.Enqueue(protocol_late_match,
+			    IntrusivePtr{AdoptRef{}, BuildConnVal()},
+			    IntrusivePtr{NewRef{}, tval}
+			);
 			}
 
 		stream_buffer.state = dpd_late_match_stop ? SKIPPING : MATCHING_ONLY;

src/broker/Manager.cc

@@ -984,7 +984,8 @@ void Manager::ProcessEvent(const broker::topic& topic, broker::zeek::Event ev)
 		return;
 		}
 
-	val_list vl(args.size());
+	zeek::Args vl;
+	vl.reserve(args.size());
 
 	for ( auto i = 0u; i < args.size(); ++i )
 		{
@@ -993,7 +994,7 @@ void Manager::ProcessEvent(const broker::topic& topic, broker::zeek::Event ev)
 		auto val = data_to_val(std::move(args[i]), expected_type);
 
 		if ( val )
-			vl.push_back(val.release());
+			vl.emplace_back(std::move(val));
 		else
 			{
 			auto expected_name = type_name(expected_type->Tag());
@@ -1014,13 +1015,8 @@ void Manager::ProcessEvent(const broker::topic& topic, broker::zeek::Event ev)
 			}
 		}
 
-	if ( static_cast<size_t>(vl.length()) == args.size() )
-		mgr.QueueEventFast(handler, std::move(vl), SOURCE_BROKER);
-	else
-		{
-		for ( const auto& v : vl )
-			Unref(v);
-		}
+	if ( vl.size() == args.size() )
+		mgr.Enqueue(handler, std::move(vl), SOURCE_BROKER);
 	}
 
 bool bro_broker::Manager::ProcessLogCreate(broker::zeek::LogCreate lc)
@@ -1243,7 +1239,7 @@ void Manager::ProcessStatus(broker::status stat)
 		return;
 
 	auto ei = internal_type("Broker::EndpointInfo")->AsRecordType();
-	auto endpoint_info = new RecordVal(ei);
+	auto endpoint_info = make_intrusive<RecordVal>(ei);
 
 	if ( ctx )
 		{
@@ -1268,9 +1264,9 @@ void Manager::ProcessStatus(broker::status stat)
 		}
 
 	auto str = stat.message();
-	auto msg = new StringVal(str ? *str : "");
+	auto msg = make_intrusive<StringVal>(str ? *str : "");
 
-	mgr.QueueEventFast(event, {endpoint_info, msg});
+	mgr.Enqueue(event, std::move(endpoint_info), std::move(msg));
 	}
 
 void Manager::ProcessError(broker::error err)
@@ -1347,10 +1343,10 @@ void Manager::ProcessError(broker::error err)
 		msg = fmt("[%s] %s", caf::to_string(err.category()).c_str(), caf::to_string(err.context()).c_str());
 		}
 
-	mgr.QueueEventFast(Broker::error, {
-		BifType::Enum::Broker::ErrorCode->GetVal(ec).release(),
-		new StringVal(msg),
-	});
+	mgr.Enqueue(Broker::error,
+		BifType::Enum::Broker::ErrorCode->GetVal(ec),
+		make_intrusive<StringVal>(msg)
+	);
 	}
 
 void Manager::ProcessStoreResponse(StoreHandleVal* s, broker::store::response response)

src/file_analysis/File.cc

@@ -622,18 +622,23 @@ void File::FileEvent(EventHandlerPtr h)
 	if ( ! FileEventAvailable(h) )
 		return;
 
-	FileEvent(h, {val->Ref()});
+	FileEvent(h, zeek::Args{{NewRef{}, val}});
 	}
 
 void File::FileEvent(EventHandlerPtr h, val_list* vl)
 	{
-	FileEvent(h, std::move(*vl));
+	FileEvent(h, zeek::val_list_to_args(vl));
 	delete vl;
 	}
 
 void File::FileEvent(EventHandlerPtr h, val_list vl)
 	{
-	mgr.QueueEventFast(h, std::move(vl));
+	FileEvent(h, zeek::val_list_to_args(&vl));
+	}
+
+void File::FileEvent(EventHandlerPtr h, zeek::Args args)
+	{
+	mgr.Enqueue(h, std::move(args));
 
 	if ( h == file_new || h == file_over_new_connection ||
 	     h == file_sniff ||

src/file_analysis/File.h

@@ -10,6 +10,7 @@
 #include "AnalyzerSet.h"
 #include "BroString.h"
 #include "BroList.h" // for val_list
+#include "ZeekArgs.h"
 #include "WeirdState.h"
 
 using std::string;
@@ -175,6 +176,7 @@ public:
 	 * @param h pointer to an event handler.
 	 * @param vl list of argument values to pass to event call.
 	 */
+	// TODO: deprecate
 	void FileEvent(EventHandlerPtr h, val_list* vl);
 
 	/**
@@ -182,8 +184,16 @@ public:
 	 * @param h pointer to an event handler.
 	 * @param vl list of argument values to pass to event call.
 	 */
+	// TODO: deprecate
 	void FileEvent(EventHandlerPtr h, val_list vl);
 
+	/**
+	 * Raises an event related to the file's life-cycle.
+	 * @param h pointer to an event handler.
+	 * @param args list of argument values to pass to event call.
+	 */
+	void FileEvent(EventHandlerPtr h, zeek::Args args);
+
 	/**
 	 * Sets the MIME type for a file to a specific value.
 	 *

src/file_analysis/Manager.cc

@@ -432,13 +432,12 @@ string Manager::GetFileID(const analyzer::Tag& tag, Connection* c, bool is_orig)
 		analyzer_mgr->GetComponentName(tag).c_str());
 
 	EnumVal* tagval = tag.AsEnumVal();
-	Ref(tagval);
 
-	mgr.QueueEventFast(get_file_handle, {
-		tagval,
-		c->BuildConnVal(),
-		val_mgr->GetBool(is_orig),
-	});
+	mgr.Enqueue(get_file_handle,
+		IntrusivePtr{NewRef{}, tagval},
+		IntrusivePtr{AdoptRef{}, c->BuildConnVal()},
+		IntrusivePtr{AdoptRef{}, val_mgr->GetBool(is_orig)}
+	);
 	mgr.Drain(); // need file handle immediately so we don't have to buffer data
 	return current_file_id;
 	}

src/file_analysis/analyzer/data_event/DataEvent.cc

@@ -42,11 +42,11 @@ bool DataEvent::DeliverChunk(const u_char* data, uint64_t len, uint64_t offset)
 	{
 	if ( ! chunk_event ) return true;
 
-	mgr.QueueEventFast(chunk_event, {
-		GetFile()->GetVal()->Ref(),
-		new StringVal(new BroString(data, len, 0)),
-		val_mgr->GetCount(offset),
-	});
+	mgr.Enqueue(chunk_event,
+		IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+		make_intrusive<StringVal>(new BroString(data, len, 0)),
+		IntrusivePtr{AdoptRef{}, val_mgr->GetCount(offset)}
+	);
 
 	return true;
 	}
@@ -55,10 +55,10 @@ bool DataEvent::DeliverStream(const u_char* data, uint64_t len)
 	{
 	if ( ! stream_event ) return true;
 
-	mgr.QueueEventFast(stream_event, {
-		GetFile()->GetVal()->Ref(),
-		new StringVal(new BroString(data, len, 0)),
-	});
+	mgr.Enqueue(stream_event,
+		IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+		make_intrusive<StringVal>(new BroString(data, len, 0))
+	);
 
 	return true;
 	}

src/file_analysis/analyzer/entropy/Entropy.cc

@@ -60,15 +60,15 @@ void Entropy::Finalize()
 	montepi = scc = ent = mean = chisq = 0.0;
 	entropy->Get(&ent, &chisq, &mean, &montepi, &scc);
 
-	RecordVal* ent_result = new RecordVal(entropy_test_result);
+	auto ent_result = make_intrusive<RecordVal>(entropy_test_result);
 	ent_result->Assign(0, make_intrusive<Val>(ent,     TYPE_DOUBLE));
 	ent_result->Assign(1, make_intrusive<Val>(chisq,   TYPE_DOUBLE));
 	ent_result->Assign(2, make_intrusive<Val>(mean,    TYPE_DOUBLE));
 	ent_result->Assign(3, make_intrusive<Val>(montepi, TYPE_DOUBLE));
 	ent_result->Assign(4, make_intrusive<Val>(scc,     TYPE_DOUBLE));
 
-	mgr.QueueEventFast(file_entropy, {
-		GetFile()->GetVal()->Ref(),
-		ent_result,
-	});
+	mgr.Enqueue(file_entropy,
+		IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+		std::move(ent_result)
+	);
 	}

src/file_analysis/analyzer/hash/Hash.cc

@@ -51,9 +51,9 @@ void Hash::Finalize()
 	if ( ! file_hash )
 		return;
 
-	mgr.QueueEventFast(file_hash, {
-		GetFile()->GetVal()->Ref(),
-		new StringVal(kind),
-		hash->Get().release(),
-	});
+	mgr.Enqueue(file_hash,
+		IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+		make_intrusive<StringVal>(kind),
+		hash->Get()
+	);
 	}

src/file_analysis/analyzer/pe/pe-analyzer.pac

@@ -42,7 +42,7 @@ refine flow File += {
 		%{
 		if ( pe_dos_header )
 			{
-			RecordVal* dh = new RecordVal(BifType::Record::PE::DOSHeader);
+			auto dh = make_intrusive<RecordVal>(BifType::Record::PE::DOSHeader);
 			dh->Assign(0, make_intrusive<StringVal>(${h.signature}.length(), (const char*) ${h.signature}.data()));
 			dh->Assign(1, val_mgr->GetCount(${h.UsedBytesInTheLastPage}));
 			dh->Assign(2, val_mgr->GetCount(${h.FileSizeInPages}));
@@ -61,10 +61,9 @@ refine flow File += {
 			dh->Assign(15, val_mgr->GetCount(${h.OEMinfo}));
 			dh->Assign(16, val_mgr->GetCount(${h.AddressOfNewExeHeader}));
 
-			mgr.QueueEventFast(pe_dos_header, {
-			    connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-			    dh
-			    });
+			mgr.Enqueue(pe_dos_header,
+			    IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+			    std::move(dh));
 			}
 		return true;
 		%}
@@ -72,12 +71,10 @@ refine flow File += {
 	function proc_dos_code(code: bytestring): bool
 		%{
 		if ( pe_dos_code )
-			{
-			mgr.QueueEventFast(pe_dos_code, {
-			    connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-			    new StringVal(code.length(), (const char*) code.data())
-			    });
-			}
+			mgr.Enqueue(pe_dos_code,
+			    IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+			    make_intrusive<StringVal>(code.length(), (const char*) code.data())
+			    );
 		return true;
 		%}
 
@@ -95,7 +92,7 @@ refine flow File += {
 		%{
 		if ( pe_file_header )
 			{
-			RecordVal* fh = new RecordVal(BifType::Record::PE::FileHeader);
+			auto fh = make_intrusive<RecordVal>(BifType::Record::PE::FileHeader);
 			fh->Assign(0, val_mgr->GetCount(${h.Machine}));
 			fh->Assign(1, make_intrusive<Val>(static_cast<double>(${h.TimeDateStamp}), TYPE_TIME));
 			fh->Assign(2, val_mgr->GetCount(${h.PointerToSymbolTable}));
@@ -103,10 +100,9 @@ refine flow File += {
 			fh->Assign(4, val_mgr->GetCount(${h.SizeOfOptionalHeader}));
 			fh->Assign(5, characteristics_to_bro(${h.Characteristics}, 16));
 
-			mgr.QueueEventFast(pe_file_header, {
-			    connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-			    fh
-			    });
+			mgr.Enqueue(pe_file_header,
+			    IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+			    std::move(fh));
 			}
 
 		return true;
@@ -124,7 +120,7 @@ refine flow File += {
 
 		if ( pe_optional_header )
 			{
-			RecordVal* oh = new RecordVal(BifType::Record::PE::OptionalHeader);
+			auto oh = make_intrusive<RecordVal>(BifType::Record::PE::OptionalHeader);
 
 			oh->Assign(0, val_mgr->GetCount(${h.magic}));
 			oh->Assign(1, val_mgr->GetCount(${h.major_linker_version}));
@@ -155,10 +151,9 @@ refine flow File += {
 
 			oh->Assign(23, process_rvas(${h.rvas}));
 
-			mgr.QueueEventFast(pe_optional_header, {
-			    connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-			    oh
-			    });
+			mgr.Enqueue(pe_optional_header,
+			    IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+			    std::move(oh));
 			}
 		return true;
 		%}
@@ -167,7 +162,7 @@ refine flow File += {
 		%{
 		if ( pe_section_header )
 			{
-			RecordVal* section_header = new RecordVal(BifType::Record::PE::SectionHeader);
+			auto section_header = make_intrusive<RecordVal>(BifType::Record::PE::SectionHeader);
 
 			// Strip null characters from the end of the section name.
 			u_char* first_null = (u_char*) memchr(${h.name}.data(), 0, ${h.name}.length());
@@ -188,10 +183,10 @@ refine flow File += {
 			section_header->Assign(8, val_mgr->GetCount(${h.non_used_num_of_line_nums}));
 			section_header->Assign(9, characteristics_to_bro(${h.characteristics}, 32));
 
-			mgr.QueueEventFast(pe_section_header, {
-			    connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-			    section_header
-			    });
+			mgr.Enqueue(pe_section_header,
+			    IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+			    std::move(section_header)
+			    );
 			}
 		return true;
 		%}

src/file_analysis/analyzer/unified2/unified2-analyzer.pac

@@ -66,7 +66,7 @@ refine flow Flow += {
 		%{
 		if ( ::unified2_event )
 			{
-			RecordVal* ids_event = new RecordVal(BifType::Record::Unified2::IDSEvent);
+			auto ids_event = make_intrusive<RecordVal>(BifType::Record::Unified2::IDSEvent);
 			ids_event->Assign(0, val_mgr->GetCount(${ev.sensor_id}));
 			ids_event->Assign(1, val_mgr->GetCount(${ev.event_id}));
 			ids_event->Assign(2, make_intrusive<Val>(ts_to_double(${ev.ts}), TYPE_TIME));
@@ -81,11 +81,9 @@ refine flow Flow += {
 			ids_event->Assign(11, to_port(${ev.dst_p}, ${ev.protocol}));
 			ids_event->Assign(17, val_mgr->GetCount(${ev.packet_action}));
 
-			mgr.QueueEventFast(::unified2_event, {
-					connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-					ids_event,
-					},
-				SOURCE_LOCAL);
+			mgr.Enqueue(::unified2_event,
+					IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+					std::move(ids_event));
 			}
 		return true;
 		%}
@@ -94,7 +92,7 @@ refine flow Flow += {
 		%{
 		if ( ::unified2_event )
 			{
-			RecordVal* ids_event = new RecordVal(BifType::Record::Unified2::IDSEvent);
+			auto ids_event = make_intrusive<RecordVal>(BifType::Record::Unified2::IDSEvent);
 			ids_event->Assign(0, val_mgr->GetCount(${ev.sensor_id}));
 			ids_event->Assign(1, val_mgr->GetCount(${ev.event_id}));
 			ids_event->Assign(2, make_intrusive<Val>(ts_to_double(${ev.ts}), TYPE_TIME));
@@ -113,11 +111,9 @@ refine flow Flow += {
 			ids_event->Assign(15, val_mgr->GetCount(${ev.mpls_label}));
 			ids_event->Assign(16, val_mgr->GetCount(${ev.vlan_id}));
 
-			mgr.QueueEventFast(::unified2_event, {
-					connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-					ids_event,
-					},
-				SOURCE_LOCAL);
+			mgr.Enqueue(::unified2_event,
+					IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+					std::move(ids_event));
 			}
 
 		return true;
@@ -127,7 +123,7 @@ refine flow Flow += {
 		%{
 		if ( ::unified2_packet )
 			{
-			RecordVal* packet = new RecordVal(BifType::Record::Unified2::Packet);
+			auto packet = make_intrusive<RecordVal>(BifType::Record::Unified2::Packet);
 			packet->Assign(0, val_mgr->GetCount(${pkt.sensor_id}));
 			packet->Assign(1, val_mgr->GetCount(${pkt.event_id}));
 			packet->Assign(2, val_mgr->GetCount(${pkt.event_second}));
@@ -135,11 +131,9 @@ refine flow Flow += {
 			packet->Assign(4, val_mgr->GetCount(${pkt.link_type}));
 			packet->Assign(5, bytestring_to_val(${pkt.packet_data}));
 
-			mgr.QueueEventFast(::unified2_packet, {
-					connection()->bro_analyzer()->GetFile()->GetVal()->Ref(),
-					packet,
-					},
-				SOURCE_LOCAL);
+			mgr.Enqueue(::unified2_packet,
+					IntrusivePtr{NewRef{}, connection()->bro_analyzer()->GetFile()->GetVal()},
+					std::move(packet));
 			}
 
 		return true;

src/file_analysis/analyzer/x509/OCSP.cc

@@ -420,10 +420,10 @@ void file_analysis::OCSP::ParseRequest(OCSP_REQUEST* req)
 #endif
 
 	if ( ocsp_request )
-		mgr.QueueEventFast(ocsp_request, {
-			GetFile()->GetVal()->Ref(),
-			val_mgr->GetCount(version),
-		});
+		mgr.Enqueue(ocsp_request,
+			IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+			IntrusivePtr{AdoptRef{}, val_mgr->GetCount(version)}
+		);
 
 	BIO *bio = BIO_new(BIO_s_mem());
 
@@ -466,10 +466,10 @@ void file_analysis::OCSP::ParseResponse(OCSP_RESPONSE *resp)
 	StringVal* status_val = new StringVal(strlen(status_str), status_str);
 
 	if ( ocsp_response_status )
-		mgr.QueueEventFast(ocsp_response_status, {
-			GetFile()->GetVal()->Ref(),
-			status_val->Ref(),
-		});
+		mgr.Enqueue(ocsp_response_status,
+			IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+			IntrusivePtr{NewRef{}, status_val}
+		);
 
 	//if (!resp_bytes)
 	//	{

src/file_analysis/analyzer/x509/X509.cc

@@ -289,16 +289,16 @@ void file_analysis::X509::ParseBasicConstraints(X509_EXTENSION* ex)
 		{
 		if ( x509_ext_basic_constraints )
 			{
-			RecordVal* pBasicConstraint = new RecordVal(BifType::Record::X509::BasicConstraints);
+			auto pBasicConstraint = make_intrusive<RecordVal>(BifType::Record::X509::BasicConstraints);
 			pBasicConstraint->Assign(0, val_mgr->GetBool(constr->ca ? 1 : 0));
 
 			if ( constr->pathlen )
 				pBasicConstraint->Assign(1, val_mgr->GetCount((int32_t) ASN1_INTEGER_get(constr->pathlen)));
 
-			mgr.QueueEventFast(x509_ext_basic_constraints, {
-				GetFile()->GetVal()->Ref(),
-				pBasicConstraint,
-			});
+			mgr.Enqueue(x509_ext_basic_constraints,
+				IntrusivePtr{NewRef{}, GetFile()->GetVal()},
+				std::move(pBasicConstraint)
+			);
 			}
 
 		BASIC_CONSTRAINTS_free(constr);

src/file_analysis/analyzer/x509/x509-extension.pac

@@ -38,15 +38,15 @@ refine connection MockConnection += {
 		if ( ! x509_ocsp_ext_signed_certificate_timestamp )
 			return true;
 
-		mgr.QueueEventFast(x509_ocsp_ext_signed_certificate_timestamp, {
-			bro_analyzer()->GetFile()->GetVal()->Ref(),
-			val_mgr->GetCount(version),
-			new StringVal(logid.length(), reinterpret_cast<const char*>(logid.begin())),
-			val_mgr->GetCount(timestamp),
-			val_mgr->GetCount(digitally_signed_algorithms->HashAlgorithm()),
-			val_mgr->GetCount(digitally_signed_algorithms->SignatureAlgorithm()),
-			new StringVal(digitally_signed_signature.length(), reinterpret_cast<const char*>(digitally_signed_signature.begin()))
-			});
+		mgr.Enqueue(x509_ocsp_ext_signed_certificate_timestamp,
+			IntrusivePtr{NewRef{}, bro_analyzer()->GetFile()->GetVal()},
+			IntrusivePtr{AdoptRef{}, val_mgr->GetCount(version)},
+			make_intrusive<StringVal>(logid.length(), reinterpret_cast<const char*>(logid.begin())),
+			IntrusivePtr{AdoptRef{}, val_mgr->GetCount(timestamp)},
+			IntrusivePtr{AdoptRef{}, val_mgr->GetCount(digitally_signed_algorithms->HashAlgorithm())},
+			IntrusivePtr{AdoptRef{}, val_mgr->GetCount(digitally_signed_algorithms->SignatureAlgorithm())},
+			make_intrusive<StringVal>(digitally_signed_signature.length(), reinterpret_cast<const char*>(digitally_signed_signature.begin()))
+			);
 
 		return true;
 		%}

src/iosource/pcap/Source.cc

@@ -48,7 +48,7 @@ void PcapSource::Close()
 	Closed();
 
 	if ( Pcap::file_done )
-		mgr.QueueEventFast(Pcap::file_done, {new StringVal(props.path)});
+		mgr.Enqueue(Pcap::file_done, make_intrusive<StringVal>(props.path));
 	}
 
 void PcapSource::OpenLive()

src/logging/Manager.cc

@@ -710,7 +710,7 @@ bool Manager::Write(EnumVal* id, RecordVal* columns_arg)
 
 	// Raise the log event.
 	if ( stream->event )
-		mgr.QueueEventFast(stream->event, {columns->Ref()}, SOURCE_LOCAL);
+		mgr.Enqueue(stream->event, columns);
 
 	// Send to each of our filters.
 	for ( list<Filter*>::iterator i = stream->filters.begin();

src/main.cc

@@ -269,7 +269,7 @@ void terminate_bro()
 
 	EventHandlerPtr zeek_done = internal_handler("zeek_done");
 	if ( zeek_done )
-		mgr.QueueEventFast(zeek_done, val_list{});
+		mgr.Enqueue(zeek_done, zeek::Args{});
 
 	timer_mgr->Expire();
 	mgr.Drain();
@@ -826,7 +826,7 @@ int main(int argc, char** argv)
 
 	EventHandlerPtr zeek_init = internal_handler("zeek_init");
 	if ( zeek_init )	//### this should be a function
-		mgr.QueueEventFast(zeek_init, val_list{});
+		mgr.Enqueue(zeek_init, zeek::Args{});
 
 	EventRegistry::string_list dead_handlers =
 		event_registry->UnusedHandlers();
@@ -873,10 +873,10 @@ int main(int argc, char** argv)
 			if ( i->skipped )
 				continue;
 
-			mgr.QueueEventFast(zeek_script_loaded, {
-				new StringVal(i->name.c_str()),
-				val_mgr->GetCount(i->include_level),
-			});
+			mgr.Enqueue(zeek_script_loaded,
+				make_intrusive<StringVal>(i->name.c_str()),
+				IntrusivePtr{AdoptRef{}, val_mgr->GetCount(i->include_level)}
+			);
 			}
 		}