Версия 2.1-alt5 (модифицированный LogServer и uniset-log)

Utilities/ULog/log.cc

@@ -76,6 +76,7 @@ int main( int argc, char** argv )
 	int cmdonly = 0;
 	timeout_t tout = 0;
 	timeout_t rdelay = 5000;
+	bool cmdlist = false;
 
 	try
 	{
@@ -132,14 +133,14 @@ int main( int argc, char** argv )
 				case 'l':
 				{
 					cmdonly = 1;
-					cmd = LogServerTypes::cmdList;
 					std::string filter("");
 					char* arg2 = checkArg(optind, argc, argv);
 
 					if( arg2 )
 						filter = string(arg2);
 
-					logfilter = filter;
+					vcmd.push_back( LogReader::Command(LogServerTypes::cmdList,0, filter) );
+					cmdlist = true;
 				}
 				break;
 
@@ -237,7 +238,16 @@ int main( int argc, char** argv )
 		lr.setCommandOnlyMode(cmdonly);
 		lr.setinTimeout(tout);
 		lr.setReconnectDelay(rdelay);
-		
+
+/*		
+		if( cmdlist && vcmd.size() == 1 )
+		{
+			cmdonly = 1;
+			lr.setReadCount(1);
+			lr.sendCommand(addr, port, vcmd, cmdonly, verb);
+			return 0;
+		}
+*/		
 		if( !vcmd.empty() )
 			lr.sendCommand(addr, port, vcmd, cmdonly, verb);
 

Utilities/ULog/logserver.cc

@@ -114,10 +114,16 @@ int main( int argc, char** argv )
 
 		}
 
+		la->add(dlog);
+
 		auto la2 = make_shared<LogAgregator>("la2");
 
 		auto dlog3 = la2->create("dlog3");
 		auto dlog4 = la2->create("dlog4");
+
+		la2->add(dlog);
+		la2->add(dlog2);
+		
 		la->add(la2);
 
 		if( la->getLog("la2/dlog3") == nullptr )
@@ -132,7 +138,7 @@ int main( int argc, char** argv )
 		auto dlog5 = la3->create("dlog5");
 		auto dlog6 = la3->create("dlog6");
 		la->add(la3);
-
+#if 0
 		for( int i = 0; i < 15; i++ )
 		{
 			ostringstream s;
@@ -145,7 +151,7 @@ int main( int argc, char** argv )
 
 			la->add(l);
 		}
-
+#endif
 
 #if 0
 		cout << la << endl;

conf/libuniset2.spec

@@ -13,7 +13,7 @@
 
 Name: libuniset2
 Version: 2.1
-Release: alt4
+Release: alt5
 
 Summary: UniSet - library for building distributed industrial control systems
 
@@ -455,6 +455,9 @@ mv -f %buildroot%python_sitelibdir_noarch/* %buildroot%python_sitelibdir/%oname
 # ..
 
 %changelog
+* Tue Jun 02 2015 Pavel Vainerman <pv@altlinux.ru> 2.1-alt5
+- (LogServer): refactoring
+
 * Mon Jun 01 2015 Pavel Vainerman <pv@altlinux.ru> 2.1-alt4
 - (LogServer): added suppport "filter mode"
 - (LogAgregator): refactoring, change show loglist format

include/LogAgregator.h

@@ -107,6 +107,10 @@
 	\note Полнота и формат поддерживаемых регулярных выражений зависит от поддержки компилятором стандарта с++11 (класс <regex>).
 */
 // -------------------------------------------------------------------------
+/* Т.к. в других агрегаторах может тоже встречаться такие же логи, приходится отдельно вести
+ * учёт подключений (conmap) и подключаться к потокам напрямую, а не к агрегатору
+ * иначе будет происходить дублирование информации на экране (логи смешиваются от разных агрегаторов)
+*/
 class LogAgregator:
 	public DebugStream
 {
@@ -135,6 +139,7 @@ class LogAgregator:
 
 		// найти лог..
 		std::shared_ptr<DebugStream> getLog( const std::string& logname );
+		bool logExist( std::shared_ptr<DebugStream>& l );
 
 		struct iLog
 		{
@@ -162,7 +167,7 @@ class LogAgregator:
 
 	protected:
 		void logOnEvent( const std::string& s );
-		void addLog( std::shared_ptr<DebugStream> l, const std::string& lname );
+		void addLog( std::shared_ptr<DebugStream> l, const std::string& lname, bool connect );
 		void addLogAgregator( std::shared_ptr<LogAgregator> la, const std::string& lname );
 
 		// поиск лога по составному логу.."agregator/agregator2/.../logname"
@@ -177,6 +182,9 @@ class LogAgregator:
 	private:
 		typedef std::unordered_map<std::string, std::shared_ptr<DebugStream>> LogMap;
 		LogMap lmap;
+
+		typedef std::unordered_map<std::shared_ptr<DebugStream>,sigc::connection> ConnectionMap;
+		ConnectionMap conmap;
 };
 // -------------------------------------------------------------------------
 #endif // LogAgregator_H_

src/Log/LogAgregator.cc

@@ -54,7 +54,8 @@ std::shared_ptr<DebugStream> LogAgregator::create( const std::string& logname )
 
 	auto l = std::make_shared<DebugStream>();
 	l->setLogName(logname);
-	l->signal_stream_event().connect( sigc::mem_fun(this, &LogAgregator::logOnEvent) );
+	auto conn = l->signal_stream_event().connect( sigc::mem_fun(this, &LogAgregator::logOnEvent) );
+	conmap.emplace(l,conn);
 	lmap[logname] = l;
 	return l;
 }
@@ -66,7 +67,7 @@ void LogAgregator::add( std::shared_ptr<DebugStream> l, const std::string& lname
 	if( lag )
 		addLogAgregator(lag, (lname.empty() ? l->getLogName() : lname) );
 	else
-		addLog(l, (lname.empty() ? l->getLogName() : lname));
+		addLog(l, (lname.empty() ? l->getLogName() : lname), true);
 }
 // ------------------------------------------------------------------------
 void LogAgregator::add( std::shared_ptr<LogAgregator> loga, const std::string& lname  )
@@ -81,33 +82,37 @@ void LogAgregator::addLogAgregator( std::shared_ptr<LogAgregator> la, const std:
 	if( i != lmap.end() )
 		return;
 
-	addLog(la, lname);
+	auto lst = la->getLogList();
+	for( auto&& l: lst )
+	{
+		auto c = conmap.find(l.log);
+		if( c == conmap.end() )
+		{
+			auto conn = l.log->signal_stream_event().connect( sigc::mem_fun(this, &LogAgregator::logOnEvent) );
+			conmap.emplace(l.log,conn);
+		}
+	}
+
+	addLog(la, lname, false);
 }
 // ------------------------------------------------------------------------
-void LogAgregator::addLog( std::shared_ptr<DebugStream> l, const std::string& lname )
+void LogAgregator::addLog( std::shared_ptr<DebugStream> l, const std::string& lname, bool connect )
 {
 	auto i = lmap.find(lname);
 
 	if( i != lmap.end() )
 		return;
 
-	bool have = false;
-
-	// пока делаем "не оптимальный поиск" / пробегаем по всему списку, зато не храним дополнительный map /
-	// на то, чтобы не подключаться к одному логу дважды,
-	// иначе будет дублироваие при выводе потом (на экран)
-	for( const auto& i : lmap )
+	if( connect )
 	{
-		if( i.second == l )
+		auto c = conmap.find(l);
+		if( c == conmap.end() )
 		{
-			have = true;
-			break;
+			auto conn = l->signal_stream_event().connect( sigc::mem_fun(this, &LogAgregator::logOnEvent) );
+			conmap.emplace(l,conn);
 		}
 	}
 
-	if( !have )
-		l->signal_stream_event().connect( sigc::mem_fun(this, &LogAgregator::logOnEvent) );
-
 	lmap[lname] = l;
 }
 // ------------------------------------------------------------------------
@@ -224,6 +229,25 @@ std::vector<std::string> LogAgregator::splitFirst( const std::string& lname, con
 	return std::move(v);
 }
 // -------------------------------------------------------------------------
+bool LogAgregator::logExist( std::shared_ptr<DebugStream>& log )
+{
+	for( const auto& l: lmap )
+	{
+		if( l.second == log )
+			return true;
+
+		bool res = false;
+		auto ag = dynamic_pointer_cast<LogAgregator>(l.second);
+		if( ag )
+			res = ag->logExist(log);
+
+		if( res )
+			return true;
+	}
+
+	return false;
+}
+// -------------------------------------------------------------------------
 std::shared_ptr<DebugStream> LogAgregator::findLog( const std::string& lname )
 {
 	auto v = splitFirst(lname, sep);

src/Log/LogReader.cc

@@ -132,6 +132,8 @@ void LogReader::sendCommand( const std::string& _addr, ost::tpport_t _port, std:
 	if( outTimeout == 0 )
 		outTimeout = TIMEOUT_INF;
 
+	std::string listfilter("");
+
 	for( const auto& c : vcmd )
 	{
 		if( c.cmd == LogServerTypes::cmdNOP )
@@ -175,6 +177,13 @@ void LogReader::sendCommand( const std::string& _addr, ost::tpport_t _port, std:
 					continue;
 				}
 
+				if( c.cmd == LogServerTypes::cmdList )
+				{
+					listfilter = c.logfilter;
+					n=0;
+					continue;
+				}
+
 				sendCommand(msg, verbose);
 				break;
 			}
@@ -193,37 +202,31 @@ void LogReader::sendCommand( const std::string& _addr, ost::tpport_t _port, std:
 
 	// после команд.. выводим список текущий..
 
-	timeout_t reply_timeout = 4000; // TIMEOUT_INF;
+	timeout_t reply_timeout = 1000; // TIMEOUT_INF;
 
 	LogServerTypes::lsMessage msg;
 	msg.cmd = LogServerTypes::cmdList;
 	msg.data = 0;
-	msg.setLogName("ALL");
+	msg.setLogName( (listfilter.empty() ? "ALL" : listfilter) );
 	sendCommand(msg, verbose);
 
 	// теперь ждём ответ..
 	try
 	{
-		int a = 3;
+		int a = 1;
 
 		while( a > 0 && tcp->isPending(ost::Socket::pendingInput, reply_timeout) )
 		{
-			int n = 0;
-
-			do
+			int n = tcp->peek( buf, sizeof(buf) - 1 );
+			if( n > 0 )
 			{
-				n = tcp->peek( buf, sizeof(buf) - 1 );
-
-				if( n > 0 )
-				{
-					tcp->read(buf, n);
-					buf[n] = '\0';
+				tcp->read(buf, n);
+				buf[n] = '\0';
 
-					log << buf;
-					a--;
-				}
+				log << buf;
 			}
-			while( n > 0 );
+
+			a--;
 		}
 
 		// rlog.warn() << "(LogReader): ...wait reply timeout..." << endl;