libtransport.git Changeset - 8b7923f3e3f4

Changeset - 8b7923f3e3f4

Parent rev.

Child rev.

[Not reviewed]

0 8 0

Jan Kaluza - 13 years ago 2012-07-18 15:20:06
hanzz.k@gmail.com

Much more useful spectrum2_manager

8 files changed with 250 insertions and 84 deletions:

include/transport/admininterface.h

include/transport/networkpluginserver.h

include/transport/protocol.proto

spectrum/src/main.cpp

spectrum_manager/src/CMakeLists.txt

spectrum_manager/src/main.cpp

184

src/admininterface.cpp

src/networkpluginserver.cpp

0 comments (0 inline, 0 general)

include/transport/admininterface.h

➞

Show inline comments

@@ @@ -38,6 +38,8 @@ class AdminInterface { @@
 		~AdminInterface();
 		void handleQuery(Swift::Message::ref message);
 	private:
 		void handleMessageReceived(Swift::Message::ref message);

include/transport/networkpluginserver.h

➞

Show inline comments

@@ @@ -42,6 +42,7 @@ class VCardResponder; @@
 class RosterResponder;
 class BlockResponder;
 class DummyReadBytestream;
 class AdminInterface;
 class NetworkPluginServer {
 	public:
@@ @@ -63,6 +64,10 @@ class NetworkPluginServer { @@
 		virtual ~NetworkPluginServer();
 		void setAdminInterface(AdminInterface *adminInterface) {
 			m_adminInterface = adminInterface;
+		}
 		int getBackendCount() {
 			return m_clients.size();
+		}
@@ @@ -101,7 +106,8 @@ class NetworkPluginServer { @@
 		void handleStatsPayload(Backend *c, const std::string &payload);
 		void handleFTStartPayload(const std::string &payload);
 		void handleFTFinishPayload(const std::string &payload);
 		void handleFTDataPayload(Backend *b ,const std::string &payload);
 		void handleFTDataPayload(Backend *b, const std::string &payload);
 		void handleQueryPayload(Backend *b, const std::string &payload);
 		void handleUserCreated(User *user);
 		void handleRoomJoined(User *user, const Swift::JID &who, const std::string &room, const std::string &nickname, const std::string &password);
@@ @@ -145,6 +151,7 @@ class NetworkPluginServer { @@
 		std::map<unsigned long, FileTransferManager::Transfer> m_filetransfers;
 		FileTransferManager *m_ftManager;
 		std::vector<std::string> m_crashedBackends;
 		AdminInterface *m_adminInterface;
 };
+}

include/transport/protocol.proto

➞

Show inline comments

@@ @@ -158,6 +158,7 @@ message WrapperMessage { @@
 		TYPE_FT_CONTINUE			= 28;
 		TYPE_EXIT					= 29;
 		TYPE_BACKEND_CONFIG			= 30;
 		TYPE_QUERY					= 31;
+	}
 	required Type type = 1;
 	optional bytes payload = 2;

spectrum/src/main.cpp

➞

Show inline comments

@@ @@ -334,6 +334,8 @@ int main(int argc, char **argv) @@
 	NetworkPluginServer plugin(&transport, &config, &userManager, &ftManager);
 	AdminInterface adminInterface(&transport, &userManager, &plugin, storageBackend, userRegistration);
 	plugin.setAdminInterface(&adminInterface);
 	StatsResponder statsResponder(&transport, &userManager, &plugin, storageBackend);
 	statsResponder.start();

spectrum_manager/src/CMakeLists.txt

➞

Show inline comments

 cmake_minimum_required(VERSION 2.6)
 FILE(GLOB SRC *.cpp)
 ADD_EXECUTABLE(spectrum2_manager ${SRC} ../../src/config.cpp ../../src/util.cpp)
+ADD_EXECUTABLE(spectrum2_manager ${SRC} ../../src/config.cpp ../../src/util.cpp ${CMAKE_CURRENT_SOURCE_DIR}/../../include/transport/protocol.pb.cc)
 target_link_libraries(spectrum2_manager ${SWIFTEN_LIBRARY})
+target_link_libraries(spectrum2_manager ${SWIFTEN_LIBRARY} ${PROTOBUF_LIBRARIES})
 INSTALL(TARGETS spectrum2_manager RUNTIME DESTINATION bin)

spectrum_manager/src/main.cpp

➞

Show inline comments

 #include "managerconfig.h"
 #include "transport/config.h"
 #include "transport/protocol.pb.h"
 #include "Swiften/Swiften.h"
 #include "Swiften/EventLoop/SimpleEventLoop.h"
@@ @@ -12,6 +13,11 @@ @@
 #include "signal.h"
 #include "sys/wait.h"
 #define WRAP(MESSAGE, TYPE) 	pbnetwork::WrapperMessage wrap; \
 	wrap.set_type(TYPE); \
 	wrap.set_payload(MESSAGE); \
 	wrap.SerializeToString(&MESSAGE);
 using namespace Transport;
@@ @@ -19,33 +25,7 @@ using namespace boost::filesystem; @@
 using namespace boost;
 static int finished;
 static std::string *m;
 static void handleDisconnected(Swift::Client *client, const boost::optional<Swift::ClientError> &, const std::string &server) {
 	std::cout << "[ DISCONNECTED ] " << server << "\n";
 	if (--finished == 0) {
 		exit(0);
+	}
+}
 static void handleConnected(Swift::Client *client, const std::string &server) {
 	boost::shared_ptr<Swift::Message> message(new Swift::Message());
 	message->setTo(server);
 	message->setFrom(client->getJID());
 	message->setBody(*m);
 	client->sendMessage(message);
+}
 static void handleMessageReceived(Swift::Client *client, Swift::Message::ref message, const std::string &server) {
 	std::string body = message->getBody();
 	boost::replace_all(body, "\n", "\n[      OK      ] " + server + ": ");
 	std::cout << "[      OK      ] " << server << ": " << body <<  "\n";
 	if (--finished == 0) {
 		exit(0);
+	}
+}
 std::string _data;
 static std::string searchForBinary(const std::string &binary) {
 	std::vector<std::string> path_list;
 	return (unsigned long) pid;
+}
 static int getPort(const std::string &portfile) {
 	path p(portfile);
 	if (!exists(p) || is_directory(p)) {
 		return 0;
+	}
 	std::ifstream f(p.string().c_str(), std::ios_base::in);
 	std::string port;
 	f >> port;
 	if (port.empty())
 		return 0;
 	return boost::lexical_cast<int>(port);
+}
 static int isRunning(const std::string &pidfile) {
 	path p(pidfile);
 	if (!exists(p) || is_directory(p)) {
@@ @@ -288,7 +284,75 @@ static int show_status(ManagerConfig *config) { @@
 	return ret;
+}
 static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactories &networkFactories, const std::string &message) {
 static void handleDataRead(boost::shared_ptr<Swift::Connection> m_conn, boost::shared_ptr<Swift::SafeByteArray> data) {
 	_data += std::string(data->begin(), data->end());
 	// Parse data while there are some
 	while (_data.size() != 0) {
 		// expected_size of wrapper message
 		unsigned int expected_size;
 		// if data is >= 4, we have whole header and we can
 		// read expected_size.
 		if (_data.size() >= 4) {
 			expected_size = *((unsigned int*) &_data[0]);
 			expected_size = ntohl(expected_size);
 			// If we don't have whole wrapper message, wait for next
 			// handleDataRead call.
 			if (_data.size() - 4 < expected_size)
 				return;
+		}
 		else {
 			return;
+		}
 		// Parse wrapper message and erase it from buffer.
 		pbnetwork::WrapperMessage wrapper;
 		if (wrapper.ParseFromArray(&_data[4], expected_size) == false) {
 			std::cout << "PARSING ERROR " << expected_size << "\n";
 			_data.erase(_data.begin(), _data.begin() + 4 + expected_size);
 			continue;
+		}
 		_data.erase(_data.begin(), _data.begin() + 4 + expected_size);
 		if (wrapper.type() == pbnetwork::WrapperMessage_Type_TYPE_QUERY) {
 			pbnetwork::BackendConfig payload;
 			if (payload.ParseFromString(wrapper.payload()) == false) {
 				std::cout << "PARSING ERROR\n";
 				// TODO: ERROR
 				continue;
+			}
 			std::cout << payload.config() << "\n";
 			exit(0);
+		}
+	}
+}
 static void handleConnected(boost::shared_ptr<Swift::Connection> m_conn, const std::string &msg, bool error) {
 	if (error) {
 		std::cerr << "Can't connect the server\n";
 		exit(50);
+	}
 	else {
 		pbnetwork::BackendConfig m;
 		m.set_config(msg);
 		std::string message;
 		m.SerializeToString(&message);
 		WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_QUERY);
 		uint32_t size = htonl(message.size());
 		char *header = (char *) &size;
 		// send header together with wrapper message
 		m_conn->write(Swift::createSafeByteArray(std::string(header, 4) + message));
+	}
+}
 static void ask_local_server(ManagerConfig *config, Swift::BoostNetworkFactories &networkFactories, const std::string &jid, const std::string &message) {
 	path p(CONFIG_STRING(config, "service.config_directory"));
 	try {
 			exit(7);
+		}
 		bool found = false;
 		directory_iterator end_itr;
 		for (directory_iterator itr(p); itr != end_itr; ++itr) {
 			if (is_regular(itr->path()) && extension(itr->path()) == ".cfg") {
 					continue;
+				}
 				if (CONFIG_VECTOR(&cfg, "service.admin_jid").empty() || CONFIG_STRING(&cfg, "service.admin_password").empty()) {
 					std::cerr << itr->path().string() << ": service.admin_jid or service.admin_password empty. This server can't be queried over XMPP.\n";
 				if (CONFIG_STRING(&cfg, "service.jid") != jid) {
 					continue;
+				}
 				finished++;
 				Swift::Client *client = new Swift::Client(CONFIG_VECTOR(&cfg, "service.admin_jid")[0], CONFIG_STRING(&cfg, "service.admin_password"), &networkFactories);
 				client->setAlwaysTrustCertificates();
 				client->onConnected.connect(boost::bind(&handleConnected, client, CONFIG_STRING(&cfg, "service.jid")));
 				client->onDisconnected.connect(bind(&handleDisconnected, client, _1, CONFIG_STRING(&cfg, "service.jid")));
 				client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, CONFIG_STRING(&cfg, "service.jid")));
 				Swift::ClientOptions opt;
 				opt.allowPLAINWithoutTLS = true;
 				client->connect(opt);
 				found = true;
 				boost::shared_ptr<Swift::Connection> m_conn;
 				m_conn = networkFactories.getConnectionFactory()->createConnection();
 				m_conn->onDataRead.connect(boost::bind(&handleDataRead, m_conn, _1));
 				m_conn->onConnectFinished.connect(boost::bind(&handleConnected, m_conn, message, _1));
 				m_conn->connect(Swift::HostAddressPort(Swift::HostAddress(CONFIG_STRING(&cfg, "service.backend_host")), getPort(CONFIG_STRING(&cfg, "service.portfile"))));
 // 				finished++;
 // 				Swift::Client *client = new Swift::Client(CONFIG_VECTOR(&cfg, "service.admin_jid")[0], CONFIG_STRING(&cfg, "service.admin_password"), &networkFactories);
 // 				client->setAlwaysTrustCertificates();
 // 				client->onConnected.connect(boost::bind(&handleConnected, client, CONFIG_STRING(&cfg, "service.jid")));
 // 				client->onDisconnected.connect(bind(&handleDisconnected, client, _1, CONFIG_STRING(&cfg, "service.jid")));
 // 				client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, CONFIG_STRING(&cfg, "service.jid")));
 // 				Swift::ClientOptions opt;
 // 				opt.allowPLAINWithoutTLS = true;
 // 				client->connect(opt);
+			}
+		}
 		if (!found) {
 			std::cerr << "Config file for Spectrum instance with this JID was not found\n";
 			exit(20)
+		}
+	}
 	catch (const filesystem_error& ex) {
 		std::cerr << "boost filesystem error\n";
+	}
+}
 // static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactories &networkFactories, const std::string &message) {
 // 	path p(CONFIG_STRING(config, "service.config_directory"));
 //
 // 	try {
 // 		if (!exists(p)) {
 // 			std::cerr << "Config directory " << CONFIG_STRING(config, "service.config_directory") << " does not exist\n";
 // 			exit(6);
 // 		}
 //
 // 		if (!is_directory(p)) {
 // 			std::cerr << "Config directory " << CONFIG_STRING(config, "service.config_directory") << " does not exist\n";
 // 			exit(7);
 // 		}
 //
 // 		directory_iterator end_itr;
 // 		for (directory_iterator itr(p); itr != end_itr; ++itr) {
 // 			if (is_regular(itr->path()) && extension(itr->path()) == ".cfg") {
 // 				Config cfg;
 // 				if (cfg.load(itr->path().string()) == false) {
 // 					std::cerr << "Can't load config file " << itr->path().string() << ". Skipping...\n";
 // 					continue;
 // 				}
 //
 // 				if (CONFIG_VECTOR(&cfg, "service.admin_jid").empty() || CONFIG_STRING(&cfg, "service.admin_password").empty()) {
 // 					std::cerr << itr->path().string() << ": service.admin_jid or service.admin_password empty. This server can't be queried over XMPP.\n";
 // 					continue;
 // 				}
 //
 // 				finished++;
 // 				Swift::Client *client = new Swift::Client(CONFIG_VECTOR(&cfg, "service.admin_jid")[0], CONFIG_STRING(&cfg, "service.admin_password"), &networkFactories);
 // 				client->setAlwaysTrustCertificates();
 // 				client->onConnected.connect(boost::bind(&handleConnected, client, CONFIG_STRING(&cfg, "service.jid")));
 // 				client->onDisconnected.connect(bind(&handleDisconnected, client, _1, CONFIG_STRING(&cfg, "service.jid")));
 // 				client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, CONFIG_STRING(&cfg, "service.jid")));
 // 				Swift::ClientOptions opt;
 // 				opt.allowPLAINWithoutTLS = true;
 // 				client->connect(opt);
 // 			}
 // 		}
 // 	}
 // 	catch (const filesystem_error& ex) {
 // 		std::cerr << "boost filesystem error\n";
 // 		exit(5);
 // 	}
 // }
 int main(int argc, char **argv)
+{
 	ManagerConfig config;
 	std::string config_file;
 	std::string command;
+	std::vector<std::string> command;
 	boost::program_options::variables_map vm;
 	boost::program_options::options_description desc("Usage: spectrum [OPTIONS] <COMMAND>\nCommands:\n"
 	boost::program_options::options_description desc("Usage: spectrum [OPTIONS] <COMMAND>\n"
 													 "       spectrum [OPTIONS] <instance_JID> <other>\nCommands:\n"
 													 " start - start all local Spectrum2 instances\n"
-													 " stop  - stop all  local Spectrum2 instances\n"
 													 " stop  - stop all local Spectrum2 instances\n"
 													 " status - status of local Spectrum2 instances\n"
-													 " <other> - send command to all local + remote Spectrum2 instances and print output\n"
+													 " <other> - send command to local Spectrum2 instance and print output\n"
 													 "Allowed options");
 	desc.add_options()
 		("help,h", "Show help output")
 		("config,c", boost::program_options::value<std::string>(&config_file)->default_value("/etc/spectrum2/spectrum_manager.cfg"), "Spectrum manager config file")
-		("command", boost::program_options::value<std::string>(&command)->default_value(""), "Command")
+		("command", boost::program_options::value<std::vector<std::string> >(&command), "Command")
+		;
 	try
+	{
@@ @@ -388,37 +512,32 @@ int main(int argc, char **argv) @@
 		return 1;
+	}
 	if (command == "start") {
+	if (command[0] == "start") {
 		start_all_instances(&config);
+	}
 	else if (command == "stop") {
+	else if (command[0] == "stop") {
 		stop_all_instances(&config);
+	}
 	else if (command == "status") {
+	else if (command[0] == "status") {
 		return show_status(&config);
+	}
 	else {
 		if (command.size() < 2) {
 			std::cout << desc << "\n";
 			return 11;
+		}
 		Swift::SimpleEventLoop eventLoop;
 		Swift::BoostNetworkFactories networkFactories(&eventLoop);
 		std::string message = command;
 		m = &message;
 		ask_local_servers(&config, networkFactories, message);
 		std::vector<std::string> servers = CONFIG_VECTOR(&config, "servers.server");
 		for (std::vector<std::string>::const_iterator it = servers.begin(); it != servers.end(); it++) {
 			finished++;
 			Swift::Client *client = new Swift::Client(CONFIG_STRING(&config, "service.admin_username") + "@" + *it, CONFIG_STRING(&config, "service.admin_password"), &networkFactories);
 			client->setAlwaysTrustCertificates();
 			client->onConnected.connect(boost::bind(&handleConnected, client, *it));
 			client->onDisconnected.connect(bind(&handleDisconnected, client, _1, *it));
 			client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, *it));
 			Swift::ClientOptions opt;
 			opt.allowPLAINWithoutTLS = true;
 			client->connect(opt);
 	// 		std::cout << *it << "\n";
+		}
 		std::string jid = command[0];
 		command.erase(command.begin());
 		std::string cmd = boost::algorithm::join(command, " ");
 		ask_local_server(&config, networkFactories, jid, cmd);
 // 		std::string message = command;
 // 		m = &message;
 // 		ask_local_server(&config, networkFactories, message);
 		eventLoop.run();
+	}

src/admininterface.cpp

➞

Show inline comments

 AdminInterface::~AdminInterface() {
+}
 void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
 	if (!message->getTo().getNode().empty())
 		return;
 	std::vector<std::string> const &x = CONFIG_VECTOR(m_component->getConfig(),"service.admin_jid");
 	if (std::find(x.begin(), x.end(), message->getFrom().toBare().toString()) == x.end()) {
 	    LOG4CXX_WARN(logger, "Message not from admin user, but from " << message->getFrom().toBare().toString());
 	    return;
+	}
 	// Ignore empty messages
 	if (message->getBody().empty()) {
 		return;
+	}
 void AdminInterface::handleQuery(Swift::Message::ref message) {
 	LOG4CXX_INFO(logger, "Message from admin received");
 	message->setTo(message->getFrom());
 	message->setFrom(m_component->getJID());
 	else {
 		message->setBody("Unknown command. Try \"help\"");
+	}
+}
 void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
 	if (!message->getTo().getNode().empty())
 		return;
 	std::vector<std::string> const &x = CONFIG_VECTOR(m_component->getConfig(),"service.admin_jid");
 	if (std::find(x.begin(), x.end(), message->getFrom().toBare().toString()) == x.end()) {
 	    LOG4CXX_WARN(logger, "Message not from admin user, but from " << message->getFrom().toBare().toString());
 	    return;
+	}
 	// Ignore empty messages
 	if (message->getBody().empty()) {
 		return;
+	}
 	handleQuery(message);
 	m_component->getStanzaChannel()->sendMessage(message);
+}

src/networkpluginserver.cpp

➞

Show inline comments

@@ @@ -32,6 +32,7 @@ @@
 #include "transport/rosterresponder.h"
 #include "transport/memoryreadbytestream.h"
 #include "transport/logging.h"
 #include "transport/admininterface.h"
 #include "blockresponder.h"
 #include "Swiften/Swiften.h"
 #include "Swiften/Server/ServerStanzaChannel.h"
 	m_config = config;
 	m_component = component;
 	m_isNextLongRun = false;
 	m_adminInterface = NULL;
 	m_component->m_factory = new NetworkFactory(this);
 	m_userManager->onUserCreated.connect(boost::bind(&NetworkPluginServer::handleUserCreated, this, _1));
 	m_userManager->onUserDestroyed.connect(boost::bind(&NetworkPluginServer::handleUserDestroyed, this, _1));
@@ @@ -763,6 +765,32 @@ void NetworkPluginServer::handlePongReceived(Backend *c) { @@
 	c->pongReceived = true;
+}
 void NetworkPluginServer::handleQueryPayload(Backend *b, const std::string &data) {
 	pbnetwork::BackendConfig payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	if (!m_adminInterface) {
 		return;
+	}
 	boost::shared_ptr<Swift::Message> msg(new Swift::Message());
 	msg->setBody(payload.config());
 	m_adminInterface->handleQuery(msg);
 	pbnetwork::BackendConfig vcard;
 	vcard.set_config(msg->getBody());
 	std::string message;
 	vcard.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_QUERY);
 	send(b->connection, message);
+}
 void NetworkPluginServer::handleDataRead(Backend *c, boost::shared_ptr<Swift::SafeByteArray> data) {
 	// Append data to buffer
 	c->data.insert(c->data.end(), data->begin(), data->end());
 			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_REMOVED:
 				handleBuddyRemovedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_QUERY:
 				handleQueryPayload(c, wrapper.payload());
 				break;
 			default:
 				return;
+		}

0 comments (0 inline, 0 general)