Merge topic/actor-system throug a squashed commit.

src/logging/CMakeLists.txt

@@ -6,6 +6,9 @@ include_directories(BEFORE
                     ${CMAKE_CURRENT_BINARY_DIR}
 )
 
+include_directories(BEFORE ${CAF_INCLUDE_DIR_CORE})
+include_directories(BEFORE ${CAF_INCLUDE_DIR_IO})
+
 add_subdirectory(writers)
 
 set(logging_SRCS

src/logging/Manager.cc

@@ -8,6 +8,7 @@
 #include "../Net.h"
 #include "../Type.h"
 
+#include "broker/Manager.h"
 #include "threading/Manager.h"
 #include "threading/SerialTypes.h"
 
@@ -18,10 +19,6 @@
 #include "../plugin/Plugin.h"
 #include "../plugin/Manager.h"
 
-#ifdef ENABLE_BROKER
-#include "broker/Manager.h"
-#endif
-
 using namespace logging;
 
 struct Manager::Filter {
@@ -82,10 +79,7 @@ struct Manager::Stream {
 
 	WriterMap writers;	// Writers indexed by id/path pair.
 
-#ifdef ENABLE_BROKER
 	bool enable_remote;
-	int remote_flags;
-#endif
 
 	~Stream();
 	};
@@ -310,10 +304,7 @@ bool Manager::CreateStream(EnumVal* id, RecordVal* sval)
 	streams[idx]->event = event ? event_registry->Lookup(event->Name()) : 0;
 	streams[idx]->columns = columns->Ref()->AsRecordType();
 
-#ifdef ENABLE_BROKER
 	streams[idx]->enable_remote = internal_val("Log::enable_remote_logging")->AsBool();
-	streams[idx]->remote_flags = broker::PEERS;
-#endif
 
 	DBG_LOG(DBG_LOGGING, "Created new logging stream '%s', raising event %s",
 		streams[idx]->name.c_str(), event ? streams[idx]->event->Name() : "<none>");
@@ -1241,11 +1232,7 @@ WriterFrontend* Manager::CreateWriter(EnumVal* id, EnumVal* writer, WriterBacken
 	winfo->info->rotation_interval = winfo->interval;
 	winfo->info->rotation_base = parse_rotate_base_time(base_time);
 
-#ifdef ENABLE_BROKER
-	winfo->writer = new WriterFrontend(*winfo->info, id, writer, local, remote, stream->remote_flags);
-#else
-	winfo->writer = new WriterFrontend(*winfo->info, id, writer, local, remote, 0);
-#endif
+	winfo->writer = new WriterFrontend(*winfo->info, id, writer, local, remote);
 	winfo->writer->Init(num_fields, fields);
 
 	if ( ! from_remote )
@@ -1326,7 +1313,7 @@ void Manager::SendAllWritersTo(RemoteSerializer::PeerID peer)
 		{
 		Stream* stream = (*s);
 
-		if ( ! stream )
+		if ( ! (stream && stream->enable_remote) )
 			continue;
 
 		for ( Stream::WriterMap::iterator i = stream->writers.begin();
@@ -1344,14 +1331,13 @@ void Manager::SendAllWritersTo(RemoteSerializer::PeerID peer)
 		}
 	}
 
-void Manager::SendAllWritersTo(const string& peer)
+void Manager::SendAllWritersTo(const broker::endpoint_info& ei)
 	{
-#ifdef ENABLE_BROKER
 	for ( vector<Stream *>::iterator s = streams.begin(); s != streams.end(); ++s )
 		{
 		Stream* stream = (*s);
 
-		if ( ! stream )
+		if ( ! (stream && stream->enable_remote) )
 			continue;
 
 		for ( Stream::WriterMap::iterator i = stream->writers.begin();
@@ -1360,16 +1346,14 @@ void Manager::SendAllWritersTo(const string& peer)
 			WriterFrontend* writer = i->second->writer;
 
 			EnumVal writer_val(i->first.first, internal_type("Log::Writer")->AsEnumType());
-			broker_mgr->CreateLog((*s)->id,
-					      &writer_val,
-					      *i->second->info,
-					      writer->NumFields(),
-					      writer->Fields(),
-					      stream->remote_flags,
-					      peer);
+			broker_mgr->PublishLogCreate((*s)->id,
+						     &writer_val,
+						     *i->second->info,
+						     writer->NumFields(),
+						     writer->Fields(),
+						     ei);
 			}
 		}
-#endif
 	}
 
 bool Manager::SetBuf(EnumVal* id, bool enabled)
@@ -1418,9 +1402,7 @@ void Manager::Terminate()
 		}
 	}
 
-#ifdef ENABLE_BROKER
-
-bool Manager::EnableRemoteLogs(EnumVal* stream_id, int flags)
+bool Manager::EnableRemoteLogs(EnumVal* stream_id)
 	{
 	auto stream = FindStream(stream_id);
 
@@ -1428,7 +1410,6 @@ bool Manager::EnableRemoteLogs(EnumVal* stream_id, int flags)
 		return false;
 
 	stream->enable_remote = true;
-	stream->remote_flags = flags;
 	return true;
 	}
 
@@ -1463,8 +1444,6 @@ RecordType* Manager::StreamColumns(EnumVal* stream_id)
 	return stream->columns;
 	}
 
-#endif
-
 // Timer which on dispatching rotates the filter.
 class RotationTimer : public Timer {
 public:

src/logging/Manager.h

@@ -5,6 +5,8 @@
 #ifndef LOGGING_MANAGER_H
 #define LOGGING_MANAGER_H
 
+#include <broker/endpoint_info.hh>
+
 #include "../Val.h"
 #include "../Tag.h"
 #include "../EventHandler.h"
@@ -173,7 +175,7 @@ public:
 	/**
 	 * Announces all instantiated writers to a given Broker peer.
 	 */
-	void SendAllWritersTo(const string& peer);
+	void SendAllWritersTo(const broker::endpoint_info& ei);
 
 	/**
 	 * Sets log streams buffering state. This adjusts all associated
@@ -203,14 +205,12 @@ public:
 	 */
 	void Terminate();
 
-#ifdef ENABLE_BROKER
 	/**
 	 * Enable remote logs for a given stream.
 	 * @param stream_id the stream to enable remote logs for.
-	 * @param flags tune behavior of how log entries are sent to peer endpoints.
 	 * @return true if remote logs are enabled.
 	 */
-	bool EnableRemoteLogs(EnumVal* stream_id, int flags);
+	bool EnableRemoteLogs(EnumVal* stream_id);
 
 	/**
 	 * Disable remote logs for a given stream.
@@ -229,7 +229,6 @@ public:
 	 * a given log stream.
 	 */
 	RecordType* StreamColumns(EnumVal* stream_id);
-#endif
 
 protected:
 	friend class WriterFrontend;

src/logging/WriterBackend.cc

@@ -119,14 +119,8 @@ bool WriterBackend::WriterInfo::Write(SerializationFormat* fmt) const
 	return true;
 	}
 
-#ifdef ENABLE_BROKER
 broker::data WriterBackend::WriterInfo::ToBroker() const
 	{
-	auto bpath = broker::record::field(path);
-	auto brotation_base = broker::record::field(rotation_base);
-	auto brotation_interval = broker::record::field(rotation_interval);
-	auto bnetwork_time = broker::record::field(network_time);
-
 	auto t = broker::table();
 
 	for ( config_map::const_iterator i = config.begin(); i != config.end(); ++i )
@@ -136,23 +130,20 @@ broker::data WriterBackend::WriterInfo::ToBroker() const
 		t.insert(std::make_pair(key, value));
 		}
 
-	auto bconfig = broker::record::field(move(t));
-
-	return move(broker::record({bpath, brotation_base, brotation_interval, bnetwork_time, bconfig}));
+	return broker::vector({path, rotation_base, rotation_interval, network_time, std::move(t)});
 	}
 
 bool WriterBackend::WriterInfo::FromBroker(broker::data d)
 	{
-	auto r = broker::get<broker::record>(d);
-
-	if ( ! r )
+	if ( ! broker::is<broker::vector>(d) )
 		return false;
 
-	auto bpath = broker::get<std::string>(*r->get(0));
-	auto brotation_base = broker::get<double>(*r->get(1));
-	auto brotation_interval = broker::get<double>(*r->get(2));
-	auto bnetwork_time = broker::get<double>(*r->get(3));
-	auto bconfig = broker::get<broker::table>(*r->get(4));
+	auto v = broker::get<broker::vector>(d);
+	auto bpath = broker::get_if<std::string>(v[0]);
+	auto brotation_base = broker::get_if<double>(v[1]);
+	auto brotation_interval = broker::get_if<double>(v[2]);
+	auto bnetwork_time = broker::get_if<double>(v[3]);
+	auto bconfig = broker::get_if<broker::table>(v[4]);
 
 	if ( ! (bpath && brotation_base && brotation_interval && bnetwork_time && bconfig) )
 		return false;
@@ -164,8 +155,8 @@ bool WriterBackend::WriterInfo::FromBroker(broker::data d)
 
 	for ( auto i : *bconfig )
 		{
-		auto k = broker::get<std::string>(i.first);
-		auto v = broker::get<std::string>(i.second);
+		auto k = broker::get_if<std::string>(i.first);
+		auto v = broker::get_if<std::string>(i.second);
 
 		if ( ! (k && v) )
 			return false;
@@ -176,7 +167,6 @@ bool WriterBackend::WriterInfo::FromBroker(broker::data d)
 
 	return true;
 	}
-#endif
 
 WriterBackend::WriterBackend(WriterFrontend* arg_frontend) : MsgThread()
 	{

src/logging/WriterBackend.h

@@ -6,10 +6,7 @@
 #define LOGGING_WRITERBACKEND_H
 
 #include "threading/MsgThread.h"
-
-#ifdef ENABLE_BROKER
 #include "broker/Data.h"
-#endif
 
 #include "Component.h"
 
@@ -118,11 +115,8 @@ public:
 		// fields. They serialize/deserialize the struct.
 		bool Read(SerializationFormat* fmt);
 		bool Write(SerializationFormat* fmt) const;
-
-#ifdef ENABLE_BROKER
 		broker::data ToBroker() const;
 		bool FromBroker(broker::data d);
-#endif
 
 		private:
 		const WriterInfo& operator=(const WriterInfo& other); // Disable.

src/logging/WriterFrontend.cc

@@ -1,10 +1,7 @@
 
 #include "Net.h"
 #include "threading/SerialTypes.h"
-
-#ifdef ENABLE_BROKER
 #include "broker/Manager.h"
-#endif
 
 #include "Manager.h"
 #include "WriterFrontend.h"
@@ -101,7 +98,7 @@ private:
 
 using namespace logging;
 
-WriterFrontend::WriterFrontend(const WriterBackend::WriterInfo& arg_info, EnumVal* arg_stream, EnumVal* arg_writer, bool arg_local, bool arg_remote, int arg_remote_flags)
+WriterFrontend::WriterFrontend(const WriterBackend::WriterInfo& arg_info, EnumVal* arg_stream, EnumVal* arg_writer, bool arg_local, bool arg_remote)
 	{
 	stream = arg_stream;
 	writer = arg_writer;
@@ -112,7 +109,6 @@ WriterFrontend::WriterFrontend(const WriterBackend::WriterInfo& arg_info, EnumVa
 	buf = true;
 	local = arg_local;
 	remote = arg_remote;
-	remote_flags = arg_remote_flags;
 	write_buffer = 0;
 	write_buffer_pos = 0;
 	info = new WriterBackend::WriterInfo(arg_info);
@@ -179,14 +175,11 @@ void WriterFrontend::Init(int arg_num_fields, const Field* const * arg_fields)
 						       arg_num_fields,
 						       arg_fields);
 
-#ifdef ENABLE_BROKER
-		broker_mgr->CreateLog(stream,
-				      writer,
-				      *info,
-				      arg_num_fields,
-				      arg_fields,
-				      remote_flags);
-#endif
+		broker_mgr->PublishLogCreate(stream,
+					     writer,
+					     *info,
+					     arg_num_fields,
+					     arg_fields);
 		}
 
 	}
@@ -214,14 +207,11 @@ void WriterFrontend::Write(int arg_num_fields, Value** vals)
 						num_fields,
 						vals);
 
-#ifdef ENABLE_BROKER
-		broker_mgr->Log(stream,
+		broker_mgr->PublishLogWrite(stream,
 				writer,
 				info->path,
 				num_fields,
-				vals,
-				remote_flags);
-#endif
+				vals);
 		}
 
 	if ( ! backend )

src/logging/WriterFrontend.h

@@ -39,12 +39,9 @@ public:
 	 * remote: If true, the writer will forward logs to remote
 	 * clients.
 	 *
-	 * remote_flags: Broker flags controlling where remote logs are
-	 * propagated to.
-	 *
 	 * Frontends must only be instantiated by the main thread.
 	 */
-	WriterFrontend(const WriterBackend::WriterInfo& info, EnumVal* stream, EnumVal* writer, bool local, bool remote, int remote_flags);
+	WriterFrontend(const WriterBackend::WriterInfo& info, EnumVal* stream, EnumVal* writer, bool local, bool remote);
 
 	/**
 	 * Destructor.
@@ -202,7 +199,6 @@ protected:
 	bool buf;	// True if buffering is enabled (default).
 	bool local;	// True if logging locally.
 	bool remote;	// True if loggin remotely.
-	int remote_flags;	// Broker propagation flags.
 
 	const char* name;	// Descriptive name of the
 	WriterBackend::WriterInfo* info;	// The writer information.