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 246 insertions and 80 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

181

src/admininterface.cpp

src/networkpluginserver.cpp

0 comments (0 inline, 0 general)

include/transport/admininterface.h

➞

Show inline comments

 /**
  * libtransport -- C++ library for easy XMPP Transports development
+ *
  * Copyright (C) 2011, Jan Kaluza <hanzz.k@gmail.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
  * (at your option) any later version.
+ *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
+ *
  * You should have received a copy of the GNU General Public License
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111-1301  USA
  */
 #pragma once
 #include <string>
 #include <map>
 #include "Swiften/Swiften.h"
 namespace Transport {
 class Component;
 class StorageBackend;
 class UserManager;
 class NetworkPluginServer;
 class UserRegistration;
 class AdminInterface {
 	public:
 		AdminInterface(Component *component, UserManager *userManager, NetworkPluginServer *server = NULL, StorageBackend *storageBackend = NULL, UserRegistration *userRegistration = NULL);
 		~AdminInterface();
 		void handleQuery(Swift::Message::ref message);
 	private:
 		void handleMessageReceived(Swift::Message::ref message);
 		Component *m_component;
 		StorageBackend *m_storageBackend;
 		UserManager *m_userManager;
 		NetworkPluginServer *m_server;
 		UserRegistration *m_userRegistration;
 };
+}

include/transport/networkpluginserver.h

➞

Show inline comments

 /**
  * libtransport -- C++ library for easy XMPP Transports development
+ *
  * Copyright (C) 2011, Jan Kaluza <hanzz.k@gmail.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
  * (at your option) any later version.
+ *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
+ *
  * You should have received a copy of the GNU General Public License
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111-1301  USA
  */
 #pragma once
 #include <time.h>
 #include "Swiften/Swiften.h"
 #include "Swiften/Presence/PresenceOracle.h"
 #include "Swiften/Disco/EntityCapsManager.h"
 #include "Swiften/Network/BoostConnectionServer.h"
 #include "Swiften/Network/Connection.h"
 #include "storagebackend.h"
 #include "transport/filetransfermanager.h"
 namespace Transport {
 class UserManager;
 class User;
 class Component;
 class Buddy;
 class LocalBuddy;
 class Config;
 class NetworkConversation;
 class VCardResponder;
 class RosterResponder;
 class BlockResponder;
 class DummyReadBytestream;
 class AdminInterface;
 class NetworkPluginServer {
 	public:
 		struct Backend {
 			int pongReceived;
 			std::list<User *> users;
 			Swift::SafeByteArray data;
 			boost::shared_ptr<Swift::Connection> connection;
 			unsigned long res;
 			unsigned long init_res;
 			unsigned long shared;
 			bool acceptUsers;
 			bool longRun;
 			bool willDie;
 			std::string id;
 		};
 		NetworkPluginServer(Component *component, Config *config, UserManager *userManager, FileTransferManager *ftManager);
 		virtual ~NetworkPluginServer();
 		void setAdminInterface(AdminInterface *adminInterface) {
 			m_adminInterface = adminInterface;
+		}
 		int getBackendCount() {
 			return m_clients.size();
+		}
 		const std::list<Backend *> &getBackends() {
 			return m_clients;
+		}
 		const std::vector<std::string> &getCrashedBackends() {
 			return m_crashedBackends;
+		}
 		void collectBackend();
 		bool moveToLongRunBackend(User *user);
 		void handleMessageReceived(NetworkConversation *conv, boost::shared_ptr<Swift::Message> &message);
 	private:
 		void handleNewClientConnection(boost::shared_ptr<Swift::Connection> c);
 		void handleSessionFinished(Backend *c);
 		void handlePongReceived(Backend *c);
 		void handleDataRead(Backend *c, boost::shared_ptr<Swift::SafeByteArray> data);
 		void handleConnectedPayload(const std::string &payload);
 		void handleDisconnectedPayload(const std::string &payload);
 		void handleBuddyChangedPayload(const std::string &payload);
 		void handleBuddyRemovedPayload(const std::string &payload);
 		void handleConvMessagePayload(const std::string &payload, bool subject = false);
 		void handleParticipantChangedPayload(const std::string &payload);
 		void handleRoomChangedPayload(const std::string &payload);
 		void handleVCardPayload(const std::string &payload);
 		void handleChatStatePayload(const std::string &payload, Swift::ChatState::ChatStateType type);
 		void handleAuthorizationPayload(const std::string &payload);
 		void handleAttentionPayload(const std::string &payload);
 		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 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);
 		void handleRoomLeft(User *user, const std::string &room);
 		void handleUserReadyToConnect(User *user);
 		void handleUserPresenceChanged(User *user, Swift::Presence::ref presence);
 		void handleUserDestroyed(User *user);
 		void handleBuddyUpdated(Buddy *buddy, const Swift::RosterItemPayload &item);
 		void handleBuddyRemoved(Buddy *buddy);
 		void handleBuddyAdded(Buddy *buddy, const Swift::RosterItemPayload &item);
 		void handleBlockToggled(Buddy *buddy);
 		void handleVCardUpdated(User *user, boost::shared_ptr<Swift::VCard> vcard);
 		void handleVCardRequired(User *user, const std::string &name, unsigned int id);
 		void handleFTStateChanged(Swift::FileTransfer::State state, const std::string &userName, const std::string &buddyName, const std::string &fileName, unsigned long size, unsigned long id);
 		void handleFTAccepted(User *user, const std::string &buddyName, const std::string &fileName, unsigned long size, unsigned long ftID);
 		void handleFTRejected(User *user, const std::string &buddyName, const std::string &fileName, unsigned long size);
 		void handleFTDataNeeded(Backend *b, unsigned long ftid);
 		void send(boost::shared_ptr<Swift::Connection> &, const std::string &data);
 		void pingTimeout();
 		void sendPing(Backend *c);
 		Backend *getFreeClient(bool acceptUsers = true, bool longRun = false);
 		UserManager *m_userManager;
 		VCardResponder *m_vcardResponder;
 		RosterResponder *m_rosterResponder;
 		BlockResponder *m_blockResponder;
 		Config *m_config;
 		boost::shared_ptr<Swift::ConnectionServer> m_server;
 		std::list<Backend *>  m_clients;
 		Swift::Timer::ref m_pingTimer;
 		Swift::Timer::ref m_collectTimer;
 		Component *m_component;
 		std::list<User *> m_waitingUsers;
 		bool m_isNextLongRun;
 		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

 package pbnetwork;
 enum ConnectionError {
 	CONNECTION_ERROR_NETWORK_ERROR = 0;
 	CONNECTION_ERROR_INVALID_USERNAME = 1;
 	CONNECTION_ERROR_AUTHENTICATION_FAILED = 2;
 	CONNECTION_ERROR_AUTHENTICATION_IMPOSSIBLE = 3;
 	CONNECTION_ERROR_NO_SSL_SUPPORT = 4;
 	CONNECTION_ERROR_ENCRYPTION_ERROR = 5;
 	CONNECTION_ERROR_NAME_IN_USE = 6;
 	CONNECTION_ERROR_INVALID_SETTINGS = 7;
 	CONNECTION_ERROR_CERT_NOT_PROVIDED = 8;
 	CONNECTION_ERROR_CERT_UNTRUSTED = 9;
 	CONNECTION_ERROR_CERT_EXPIRED = 10;
 	CONNECTION_ERROR_CERT_NOT_ACTIVATED = 11;
 	CONNECTION_ERROR_CERT_HOSTNAME_MISMATCH = 12;
 	CONNECTION_ERROR_CERT_FINGERPRINT_MISMATCH = 13;
 	CONNECTION_ERROR_CERT_SELF_SIGNED = 14;
 	CONNECTION_ERROR_CERT_OTHER_ERROR = 15;
 	CONNECTION_ERROR_OTHER_ERROR = 16;
+}
 enum StatusType {
 	STATUS_ONLINE		= 0;
 	STATUS_AWAY			= 1;
 	STATUS_FFC			= 2;
 	STATUS_XA			= 3;
 	STATUS_DND			= 4;
 	STATUS_NONE			= 5;
 	STATUS_INVISIBLE	= 6;
+}
 message Connected {
 	required string user = 1;
+}
 message Disconnected {
 	required string user = 1;
 	required int32 error = 2;
 	optional string message = 3;
+}
 message Login {
 	required string user = 1;
 	required string legacyName = 2;
 	required string password = 3;
 	repeated string extraFields = 4;
+}
 message Logout {
 	required string user = 1;
 	required string legacyName = 2;
+}
 message Buddy {
 	required string userName = 1;
 	required string buddyName = 2;
 	optional string alias = 3;
 	repeated string group = 4;
 	optional StatusType status = 5;
 	optional string statusMessage = 6;
 	optional string iconHash = 7;
 	optional bool blocked = 8;
+}
 message ConversationMessage {
 	required string userName = 1;
 	required string buddyName = 2;
 	required string message = 3;
 	optional string nickname = 4;
 	optional string xhtml = 5;
+}
 message Room {
 	required string userName = 1;
 	required string nickname = 2;
 	required string room = 3;
 	optional string password = 4;
+}
 message Participant {
 	required string userName = 1;
 	required string room = 2;
 	required string nickname = 3;
 	required int32 flag = 4;
 	required StatusType status = 5;
 	optional string statusMessage = 6;
 	optional string newname = 7;
+}
 message VCard {
 	required string userName = 1;
 	required string buddyName = 2;
 	required int32 id = 3;
 	optional string fullname = 4;
 	optional string nickname = 5;
 	optional bytes photo = 6;
+}
 message Status {
 	required string userName = 1;
 	required StatusType status = 3;
 	optional string statusMessage = 4;
+}
 message Stats {
 	required int32 res = 1;
 	required int32 init_res = 2;
 	required int32 shared = 3;
 	required string id = 4;
+}
 message File {
 	required string userName = 1;
 	required string buddyName = 2;
 	required string fileName = 3;
 	required int32 size = 4;
 	optional int32 ftID = 5;
+}
 message FileTransferData {
 	required int32 ftID = 1;
 	required bytes data = 2;
+}
 message BackendConfig {
 	required string config = 1;
+}
 message WrapperMessage {
 	enum Type {
 		TYPE_CONNECTED 				= 1;
 		TYPE_DISCONNECTED 			= 2;
 		TYPE_LOGIN 					= 3;
 		TYPE_LOGOUT 				= 4;
 		TYPE_BUDDY_CHANGED			= 6;
 		TYPE_BUDDY_REMOVED			= 7;
 		TYPE_CONV_MESSAGE			= 8;
 		TYPE_PING					= 9;
 		TYPE_PONG					= 10;
 		TYPE_JOIN_ROOM				= 11;
 		TYPE_LEAVE_ROOM				= 12;
 		TYPE_PARTICIPANT_CHANGED	= 13;
 		TYPE_ROOM_NICKNAME_CHANGED	= 14;
 		TYPE_ROOM_SUBJECT_CHANGED	= 15;
 		TYPE_VCARD					= 16;
 		TYPE_STATUS_CHANGED			= 17;
 		TYPE_BUDDY_TYPING			= 18;
 		TYPE_BUDDY_STOPPED_TYPING	= 19;
 		TYPE_BUDDY_TYPED			= 20;
 		TYPE_AUTH_REQUEST			= 21;
 		TYPE_ATTENTION				= 22;
 		TYPE_STATS					= 23;
 		TYPE_FT_START				= 24;
 		TYPE_FT_FINISH				= 25;
 		TYPE_FT_DATA				= 26;
 		TYPE_FT_PAUSE				= 27;
 		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

@@ @@ -145,224 +145,226 @@ int main(int argc, char **argv) @@
 		("config", boost::program_options::value<std::string>(&config_file)->default_value(""), "Config file")
 		("version,v", "Shows Spectrum version")
+		;
 	try
+	{
 		boost::program_options::positional_options_description p;
 		p.add("config", -1);
 		boost::program_options::store(boost::program_options::command_line_parser(argc, argv).
           options(desc).positional(p).allow_unregistered().run(), vm);
 		boost::program_options::notify(vm);
 		if (vm.count("version")) {
 			std::cout << SPECTRUM_VERSION << "\n";
 			return 0;
+		}
 		if(vm.count("help"))
+		{
 			std::cout << desc << "\n";
 			return 1;
+		}
 		if(vm.count("config") == 0) {
 			std::cout << desc << "\n";
 			return 1;
+		}
 		if(vm.count("no-daemonize")) {
 			no_daemon = true;
+		}
+	}
 	catch (std::runtime_error& e)
+	{
 		std::cout << desc << "\n";
 		return 1;
+	}
 	catch (...)
+	{
 		std::cout << desc << "\n";
 		return 1;
+	}
 	if (!config.load(vm["config"].as<std::string>(), jid)) {
 		std::cerr << "Can't load configuration file.\n";
 		return 1;
+	}
 	// create directories
 	try {
 		boost::filesystem::create_directories(
 			boost::filesystem::path(CONFIG_STRING(&config, "service.pidfile")).parent_path().string()
 		);
+	}
 	catch (...) {
 		std::cerr << "Can't create service.pidfile directory " << boost::filesystem::path(CONFIG_STRING(&config, "service.pidfile")).parent_path().string() << ".\n";
 		return 1;
+	}
 	// create directories
 	try {
 		boost::filesystem::create_directories(CONFIG_STRING(&config, "service.working_dir"));
+	}
 	catch (...) {
 		std::cerr << "Can't create service.working_dir directory " << CONFIG_STRING(&config, "service.working_dir") << ".\n";
 		return 1;
+	}
 	// create directories
 	try {
 		boost::filesystem::create_directories(
 			boost::filesystem::path(CONFIG_STRING(&config, "service.portfile")).parent_path().string()
 		);
+	}
 	catch (...) {
 		std::cerr << "Can't create service.portfile directory " << CONFIG_STRING(&config, "service.portfile") << ".\n";
 		return 1;
+	}
 #ifndef WIN32
 	if (!CONFIG_STRING(&config, "service.group").empty() ||!CONFIG_STRING(&config, "service.user").empty() ) {
 		struct group *gr;
 		if ((gr = getgrnam(CONFIG_STRING(&config, "service.group").c_str())) == NULL) {
 			std::cerr << "Invalid service.group name " << CONFIG_STRING(&config, "service.group") << "\n";
 			return 1;
+		}
 		struct passwd *pw;
 		if ((pw = getpwnam(CONFIG_STRING(&config, "service.user").c_str())) == NULL) {
 			std::cerr << "Invalid service.user name " << CONFIG_STRING(&config, "service.user") << "\n";
 			return 1;
+		}
 		chown(CONFIG_STRING(&config, "service.working_dir").c_str(), pw->pw_uid, gr->gr_gid);
+	}
 	char backendport[20];
 	FILE* port_file_f;
 	port_file_f = fopen(CONFIG_STRING(&config, "service.portfile").c_str(), "w+");
 	if (port_file_f == NULL) {
 		std::cerr << "Cannot create port_file file " << CONFIG_STRING(&config, "service.portfile").c_str() << ". Exiting\n";
 		exit(1);
+	}
 	sprintf(backendport,"%s\n",CONFIG_STRING(&config, "service.backend_port").c_str());
 	if (fwrite(backendport,1,strlen(backendport),port_file_f) < strlen(backendport)) {
 		std::cerr << "Cannot write to port file " << CONFIG_STRING(&config, "service.portfile") << ". Exiting\n";
 		exit(1);
+	}
 	fclose(port_file_f);
 	if (!no_daemon) {
 		// daemonize
 		daemonize(CONFIG_STRING(&config, "service.working_dir").c_str(), CONFIG_STRING(&config, "service.pidfile").c_str());
 // 		removeOldIcons(CONFIG_STRING(&config, "service.working_dir") + "/icons");
+    }
 #endif
 	Logging::initMainLogging(&config);
 #ifndef WIN32
 	if (!CONFIG_STRING(&config, "service.group").empty() ||!CONFIG_STRING(&config, "service.user").empty() ) {
 		struct rlimit limit;
 		getrlimit(RLIMIT_CORE, &limit);
 		if (!CONFIG_STRING(&config, "service.group").empty()) {
 			struct group *gr;
 			if ((gr = getgrnam(CONFIG_STRING(&config, "service.group").c_str())) == NULL) {
 				std::cerr << "Invalid service.group name " << CONFIG_STRING(&config, "service.group") << "\n";
 				return 1;
+			}
 			if (((setgid(gr->gr_gid)) != 0) || (initgroups(CONFIG_STRING(&config, "service.user").c_str(), gr->gr_gid) != 0)) {
 				std::cerr << "Failed to set service.group name " << CONFIG_STRING(&config, "service.group") << " - " << gr->gr_gid << ":" << strerror(errno) << "\n";
 				return 1;
+			}
+		}
 		if (!CONFIG_STRING(&config, "service.user").empty()) {
 			struct passwd *pw;
 			if ((pw = getpwnam(CONFIG_STRING(&config, "service.user").c_str())) == NULL) {
 				std::cerr << "Invalid service.user name " << CONFIG_STRING(&config, "service.user") << "\n";
 				return 1;
+			}
 			if ((setuid(pw->pw_uid)) != 0) {
 				std::cerr << "Failed to set service.user name " << CONFIG_STRING(&config, "service.user") << " - " << pw->pw_uid << ":" << strerror(errno) << "\n";
 				return 1;
+			}
+		}
 		setrlimit(RLIMIT_CORE, &limit);
+	}
 	struct rlimit limit;
 	limit.rlim_max = RLIM_INFINITY;
 	limit.rlim_cur = RLIM_INFINITY;
 	setrlimit(RLIMIT_CORE, &limit);
 #endif
 	Swift::SimpleEventLoop eventLoop;
 	Swift::BoostNetworkFactories *factories = new Swift::BoostNetworkFactories(&eventLoop);
 	UserRegistry userRegistry(&config, factories);
 	Component transport(&eventLoop, factories, &config, NULL, &userRegistry);
 	component_ = &transport;
 // 	Logger logger(&transport);
 	std::string error;
 	StorageBackend *storageBackend = StorageBackend::createBackend(&config, error);
 	if (storageBackend == NULL) {
 		if (!error.empty()) {
 			std::cerr << error << "\n";
 			return -2;
+		}
+	}
 	else if (!storageBackend->connect()) {
 		std::cerr << "Can't connect to database. Check the log to find out the reason.\n";
 		return -1;
+	}
 	UserManager userManager(&transport, &userRegistry, storageBackend);
 	userManager_ = &userManager;
 	UserRegistration *userRegistration = NULL;
 	UsersReconnecter *usersReconnecter = NULL;
 	if (storageBackend) {
 		userRegistration = new UserRegistration(&transport, &userManager, storageBackend);
 		userRegistration->start();
 		usersReconnecter = new UsersReconnecter(&transport, storageBackend);
+	}
 	FileTransferManager ftManager(&transport, &userManager);
 	NetworkPluginServer plugin(&transport, &config, &userManager, &ftManager);
 	AdminInterface adminInterface(&transport, &userManager, &plugin, storageBackend, userRegistration);
 	plugin.setAdminInterface(&adminInterface);
 	StatsResponder statsResponder(&transport, &userManager, &plugin, storageBackend);
 	statsResponder.start();
 	GatewayResponder gatewayResponder(transport.getIQRouter(), &userManager);
 	gatewayResponder.start();
 	DiscoItemsResponder discoItemsResponder(&transport);
 	discoItemsResponder.start();
 	AdHocManager adhocmanager(&transport, &discoItemsResponder);
 	adhocmanager.start();
 	SettingsAdHocCommandFactory settings;
 	adhocmanager.addAdHocCommand(&settings);
 	eventLoop_ = &eventLoop;
 	eventLoop.run();
 	if (userRegistration) {
 		userRegistration->stop();
 		delete userRegistration;
+	}
 	if (usersReconnecter) {
 		delete usersReconnecter;
+	}
 	delete storageBackend;
 	delete factories;
 	return 0;
+}

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)
 INSTALL(FILES
 	spectrum_manager.cfg
 	DESTINATION /etc/spectrum2
+	)

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"
 #include <boost/foreach.hpp>
 #include <iostream>
 #include <fstream>
 #include <iterator>
 #include <algorithm>
 #include <boost/filesystem.hpp>
 #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;
 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;
 	char * env_path = getenv("PATH");
 	if (env_path != NULL) {
 		std::string buffer = "";
 		for (int s = 0; s < strlen(env_path); s++) {
 			if (env_path[s] == ':') {
 				path_list.insert(path_list.end(), std::string(buffer));
 				buffer = "";
+			}
 			else {
 				buffer += env_path[s];
+			}
+		}
 		if (buffer != "") {
 			path_list.insert(path_list.end(), std::string(buffer));
 			buffer = "";
+		}
 		for (std::vector<std::string>::iterator dit = path_list.begin(); dit < path_list.end(); dit++) {
 			std::string bpath = *dit;
 			bpath += "/";
 			bpath += binary;
 			path p(bpath);
 			if (exists(p) && !is_directory(p)) {
 				return bpath;
+			}
+		}
+	}
 	return "";
+}
 // Executes new backend
 static unsigned long exec_(std::string path, std::string config, std::string jid = "") {
 	// fork and exec
 	pid_t pid = fork();
 	if ( pid == 0 ) {
 		// child process
 		if (jid.empty()) {
 			exit(execl(path.c_str(), path.c_str(), config.c_str(), NULL));
+		}
 		else {
 			exit(execl(path.c_str(), path.c_str(), "-j", jid.c_str(), config.c_str(), NULL));
+		}
 	} else if ( pid < 0 ) {
 		// fork failed
+	}
 	else {
 		waitpid(pid, 0, 0);
+	}
 	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)) {
 		return 0;
+	}
 	std::ifstream f(p.string().c_str(), std::ios_base::in);
 	std::string pid;
 	f >> pid;
 	if (pid.empty())
 		return 0;
 	if (kill(boost::lexical_cast<int>(pid), 0) != 0)
 		return 0;
 	return boost::lexical_cast<int>(pid);
+}
 static void start_all_instances(ManagerConfig *config) {
 	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);
+		}
 		std::string spectrum2_binary = searchForBinary("spectrum2");
 		if (spectrum2_binary.empty()) {
 			std::cerr << "spectrum2 binary not found in PATH\n";
 			exit(8);
+		}
 		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;
+				}
 				std::vector<std::string> vhosts;
 				if (CONFIG_HAS_KEY(&cfg, "vhosts.vhost"))
 					vhosts = CONFIG_VECTOR(&cfg, "vhosts.vhost");
 				vhosts.push_back(CONFIG_STRING(&cfg, "service.jid"));
 				BOOST_FOREACH(std::string &vhost, vhosts) {
 					Config vhostCfg;
 					if (vhostCfg.load(itr->path().string(), vhost) == false) {
 						std::cerr << "Can't load config file " << itr->path().string() << ". Skipping...\n";
 						continue;
+					}
 					int pid = isRunning(CONFIG_STRING(&vhostCfg, "service.pidfile"));
 					if (pid == 0) {
 						std::cout << "Starting " << itr->path() << ": OK\n";
 						exec_(spectrum2_binary, itr->path().string(), vhost);
+					}
 					else {
 						std::cout << "Starting " << itr->path() << ": Already started (PID=" << pid << ")\n";
+					}
+				}
+			}
+		}
+	}
 	catch (const filesystem_error& ex) {
 		std::cerr << "boost filesystem error\n";
 		exit(5);
+	}
+}
 static void stop_all_instances(ManagerConfig *config) {
 	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";
+				}
 				std::vector<std::string> vhosts;
 				if (CONFIG_HAS_KEY(&cfg, "vhosts.vhost"))
 					vhosts = CONFIG_VECTOR(&cfg, "vhosts.vhost");
 				vhosts.push_back(CONFIG_STRING(&cfg, "service.jid"));
 				BOOST_FOREACH(std::string &vhost, vhosts) {
 					Config vhostCfg;
 					if (vhostCfg.load(itr->path().string(), vhost) == false) {
 						std::cerr << "Can't load config file " << itr->path().string() << ". Skipping...\n";
 						continue;
+					}
 					int pid = isRunning(CONFIG_STRING(&vhostCfg, "service.pidfile"));
 					if (pid) {
 						std::cout << "Stopping " << itr->path() << ": OK\n";
 						kill(pid, SIGTERM);
+					}
 					else {
 						std::cout << "Stopping " << itr->path() << ": Not running\n";
+					}
+				}
+			}
+		}
+	}
 	catch (const filesystem_error& ex) {
 		std::cerr << "boost filesystem error\n";
 		exit(5);
+	}
+}
 static int show_status(ManagerConfig *config) {
 	int ret = 0;
 	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";
+				}
 				std::vector<std::string> vhosts;
 				if (CONFIG_HAS_KEY(&cfg, "vhosts.vhost"))
 					vhosts = CONFIG_VECTOR(&cfg, "vhosts.vhost");
 				vhosts.push_back(CONFIG_STRING(&cfg, "service.jid"));
 				BOOST_FOREACH(std::string &vhost, vhosts) {
 					Config vhostCfg;
 					if (vhostCfg.load(itr->path().string(), vhost) == false) {
 						std::cerr << "Can't load config file " << itr->path().string() << ". Skipping...\n";
 						continue;
+					}
 					int pid = isRunning(CONFIG_STRING(&vhostCfg, "service.pidfile"));
 					if (pid) {
 						std::cout << itr->path() << ": " << vhost << " Running\n";
+					}
 					else {
 						ret = 3;
 						std::cout << itr->path() << ": " << vhost << " Stopped\n";
+					}
+				}
+			}
+		}
+	}
 	catch (const filesystem_error& ex) {
 		std::cerr << "boost filesystem error\n";
 		exit(5);
+	}
 	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 {
 		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);
+		}
 		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") {
 				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";
 				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";
 		exit(5);
+	}
+}
 // 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"
 													 " 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
+	{
 		boost::program_options::positional_options_description p;
 		p.add("command", -1);
 		boost::program_options::store(boost::program_options::command_line_parser(argc, argv).
           options(desc).positional(p).run(), vm);
 		boost::program_options::notify(vm);
 		if(vm.count("help"))
+		{
 			std::cout << desc << "\n";
 			return 1;
+		}
+	}
 	catch (std::runtime_error& e)
+	{
 		std::cout << desc << "\n";
 		return 2;
+	}
 	catch (...)
+	{
 		std::cout << desc << "\n";
 		return 3;
+	}
 	if (!config.load(config_file)) {
 		std::cerr << "Can't load configuration file.\n";
 		return 4;
+	}
 	if (command.empty()) {
 		std::cout << desc << "\n";
 		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;
 		std::string jid = command[0];
 		command.erase(command.begin());
 		std::string cmd = boost::algorithm::join(command, " ");
 		ask_local_servers(&config, networkFactories, message);
 		ask_local_server(&config, networkFactories, jid, cmd);
 // 		std::string message = command;
 // 		m = &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";
+		}
 // 		ask_local_server(&config, networkFactories, message);
 		eventLoop.run();
+	}
+}

src/admininterface.cpp

➞

Show inline comments

 /**
  * libtransport -- C++ library for easy XMPP Transports development
+ *
  * Copyright (C) 2011, Jan Kaluza <hanzz.k@gmail.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
  * (at your option) any later version.
+ *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
+ *
  * You should have received a copy of the GNU General Public License
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111-1301  USA
  */
 #include "transport/admininterface.h"
 #include "transport/user.h"
 #include "transport/transport.h"
 #include "transport/storagebackend.h"
 #include "transport/conversationmanager.h"
 #include "transport/rostermanager.h"
 #include "transport/usermanager.h"
 #include "transport/networkpluginserver.h"
 #include "transport/logging.h"
 #include "transport/userregistration.h"
 #include "storageresponder.h"
 #include "transport/memoryusage.h"
 #include <boost/foreach.hpp>
 namespace Transport {
 DEFINE_LOGGER(logger, "AdminInterface");
 static std::string getArg(const std::string &body) {
 	std::string ret;
 	if (body.find(" ") == std::string::npos)
 		return ret;
 	return body.substr(body.find(" ") + 1);
+}
 AdminInterface::AdminInterface(Component *component, UserManager *userManager, NetworkPluginServer *server, StorageBackend *storageBackend, UserRegistration *userRegistration) {
 	m_component = component;
 	m_storageBackend = storageBackend;
 	m_userManager = userManager;
 	m_server = server;
 	m_userRegistration = userRegistration;
 	m_component->getStanzaChannel()->onMessageReceived.connect(bind(&AdminInterface::handleMessageReceived, this, _1));
+}
 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());
 	if (message->getBody() == "status") {
 		int users = m_userManager->getUserCount();
 		int backends = m_server->getBackendCount();
 		message->setBody("Running (" + boost::lexical_cast<std::string>(users) + " users connected using " + boost::lexical_cast<std::string>(backends) + " backends)");
+	}
 	else if (message->getBody() == "online_users") {
 		std::string lst;
 		const std::map<std::string, User *> &users = m_userManager->getUsers();
 		if (users.size() == 0)
 			lst = "0";
 		for (std::map<std::string, User *>::const_iterator it = users.begin(); it != users.end(); it ++) {
 			lst += (*it).first + "\n";
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody() == "online_users_count") {
 		int users = m_userManager->getUserCount();
 		message->setBody(boost::lexical_cast<std::string>(users));
+	}
 	else if (message->getBody() == "reload") {
 		bool done = m_component->getConfig()->reload();
 		if (done) {
 			message->setBody("Config reloaded");
+		}
 		else {
 			message->setBody("Error during config reload");
+		}
+	}
 	else if (message->getBody() == "online_users_per_backend") {
 		std::string lst;
 		int id = 1;
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		for (std::list <NetworkPluginServer::Backend *>::const_iterator b = backends.begin(); b != backends.end(); b++) {
 			NetworkPluginServer::Backend *backend = *b;
 			lst += "Backend " + boost::lexical_cast<std::string>(id) + " (ID=" + backend->id + ")";
 			lst += backend->acceptUsers ? "" : " - not-accepting";
 			lst += backend->longRun ? " - long-running" : "";
 			lst += ":\n";
 			if (backend->users.size() == 0) {
 				lst += "   waiting for users\n";
+			}
 			else {
 				time_t now = time(NULL);
 				for (std::list<User *>::const_iterator u = backend->users.begin(); u != backend->users.end(); u++) {
 					User *user = *u;
 					lst += "   " + user->getJID().toBare().toString();
 					lst += " - non-active for " + boost::lexical_cast<std::string>(now - user->getLastActivity()) + " seconds";
 					lst += "\n";
+				}
+			}
 			id++;
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody().find("has_online_user") == 0) {
 		User *user = m_userManager->getUser(getArg(message->getBody()));
 		std::cout << getArg(message->getBody()) << "\n";
 		message->setBody(boost::lexical_cast<std::string>(user != NULL));
+	}
 	else if (message->getBody() == "backends_count") {
 		int backends = m_server->getBackendCount();
 		message->setBody(boost::lexical_cast<std::string>(backends));
+	}
 	else if (message->getBody() == "res_memory") {
 		double shared = 0;
 		double rss = 0;
 #ifndef WIN32
 		process_mem_usage(shared, rss);
 #endif
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			rss += backend->res;
+		}
 		message->setBody(boost::lexical_cast<std::string>(rss));
+	}
 	else if (message->getBody() == "shr_memory") {
 		double shared = 0;
 		double rss = 0;
 #ifndef WIN32
 		process_mem_usage(shared, rss);
 #endif
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			shared += backend->shared;
+		}
 		message->setBody(boost::lexical_cast<std::string>(shared));
+	}
 	else if (message->getBody() == "used_memory") {
 		double shared = 0;
 		double rss = 0;
 #ifndef WIN32
 		process_mem_usage(shared, rss);
 #endif
 		rss -= shared;
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			rss += backend->res - backend->shared;
+		}
 		message->setBody(boost::lexical_cast<std::string>(rss));
+	}
 	else if (message->getBody() == "average_memory_per_user") {
 		if (m_userManager->getUserCount() == 0) {
 			message->setBody(boost::lexical_cast<std::string>(0));
+		}
 		else {
 			unsigned long per_user = 0;
 			const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 			BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 				per_user += (backend->res - backend->init_res);
+			}
 			message->setBody(boost::lexical_cast<std::string>(per_user / m_userManager->getUserCount()));
+		}
+	}
 	else if (message->getBody() == "res_memory_per_backend") {
 		std::string lst;
 		int id = 1;
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			lst += "Backend " + boost::lexical_cast<std::string>(id) + " (ID=" + backend->id + "): " + boost::lexical_cast<std::string>(backend->res) + "\n";
 			id++;
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody() == "shr_memory_per_backend") {
 		std::string lst;
 		int id = 1;
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			lst += "Backend " + boost::lexical_cast<std::string>(id)  + " (ID=" + backend->id + "): " + boost::lexical_cast<std::string>(backend->shared) + "\n";
 			id++;
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody() == "used_memory_per_backend") {
 		std::string lst;
 		int id = 1;
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			lst += "Backend " + boost::lexical_cast<std::string>(id)  + " (ID=" + backend->id + "): " + boost::lexical_cast<std::string>(backend->res - backend->shared) + "\n";
 			id++;
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody() == "average_memory_per_user_per_backend") {
 		std::string lst;
 		int id = 1;
 		const std::list <NetworkPluginServer::Backend *> &backends = m_server->getBackends();
 		BOOST_FOREACH(NetworkPluginServer::Backend * backend, backends) {
 			if (backend->users.size() == 0) {
 				lst += "Backend " + boost::lexical_cast<std::string>(id)  + " (ID=" + backend->id + "): 0\n";
+			}
 			else {
 				lst += "Backend " + boost::lexical_cast<std::string>(id) + " (ID=" + backend->id + "): " + boost::lexical_cast<std::string>((backend->res - backend->init_res) / backend->users.size()) + "\n";
+			}
 			id++;
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody() == "collect_backend") {
 		m_server->collectBackend();
+	}
 	else if (message->getBody() == "crashed_backends") {
 		std::string lst;
 		const std::vector<std::string> &backends = m_server->getCrashedBackends();
 		BOOST_FOREACH(const std::string &backend, backends) {
 			lst += backend + "\n";
+		}
 		message->setBody(lst);
+	}
 	else if (message->getBody() == "messages_from_xmpp") {
 		int msgCount = m_userManager->getMessagesToBackend();
 		message->setBody(boost::lexical_cast<std::string>(msgCount));
+	}
 	else if (message->getBody() == "messages_to_xmpp") {
 		int msgCount = m_userManager->getMessagesToXMPP();
 		message->setBody(boost::lexical_cast<std::string>(msgCount));
+	}
 	else if (message->getBody().find("register ") == 0 && m_userRegistration) {
 		std::string body = message->getBody();
 		std::vector<std::string> args;
 		boost::split(args, body, boost::is_any_of(" "));
 		if (args.size() == 4) {
 			UserInfo res;
 			res.jid = args[1];
 			res.uin = args[2];
 			res.password = args[3];
 			res.language = "en";
 			res.encoding = "utf-8";
 			res.vip = 0;
 			if (m_userRegistration->registerUser(res)) {
 				message->setBody("User registered.");
+			}
 			else {
 				message->setBody("Registration failed: User is already registered");
+			}
+		}
 		else {
 			message->setBody("Bad argument count. See 'help'.");
+		}
+	}
 	else if (message->getBody().find("unregister ") == 0 && m_userRegistration) {
 		std::string body = message->getBody();
 		std::vector<std::string> args;
 		boost::split(args, body, boost::is_any_of(" "));
 		if (args.size() == 2) {
 			if (m_userRegistration->unregisterUser(args[1])) {
 				message->setBody("User unregistered.");
+			}
 			else {
 				message->setBody("Registration failed: User is not registered");
+			}
+		}
 		else {
 			message->setBody("Bad argument count. See 'help'.");
+		}
+	}
 	else if (message->getBody().find("help") == 0) {
 		std::string help;
 		help += "General:\n";
 		help += "    status - shows instance status\n";
 		help += "    reload - Reloads config file\n";
 		help += "Users:\n";
 		help += "    online_users - returns list of all online users\n";
 		help += "    online_users_count - number of online users\n";
 		help += "    online_users_per_backend - shows online users per backends\n";
 		help += "    has_online_user <bare_JID> - returns 1 if user is online\n";
 		if (m_userRegistration) {
 			help += "    register <bare_JID> <legacyName> <password> - registers the new user\n";
 			help += "    unregister <bare_JID> - unregisters existing user\n";
+		}
 		help += "Messages:\n";
 		help += "    messages_from_xmpp - get number of messages received from XMPP users\n";
 		help += "    messages_to_xmpp - get number of messages sent to XMPP users\n";
 		help += "Backends:\n";
 		help += "    backends_count - number of active backends\n";
 		help += "    crashed_backends - returns IDs of crashed backends\n";
 		help += "Memory:\n";
 		help += "    res_memory - Total RESident memory spectrum2 and its backends use in KB\n";
 		help += "    shr_memory - Total SHaRed memory spectrum2 backends share together in KB\n";
 		help += "    used_memory - (res_memory - shr_memory)\n";
 		help += "    average_memory_per_user - (memory_used_without_any_user - res_memory)\n";
 		help += "    res_memory_per_backend - RESident memory used by backends in KB\n";
 		help += "    shr_memory_per_backend - SHaRed memory used by backends in KB\n";
 		help += "    used_memory_per_backend - (res_memory - shr_memory) per backend\n";
 		help += "    average_memory_per_user_per_backend - (memory_used_without_any_user - res_memory) per backend\n";
 		message->setBody(help);
+	}
 	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

 /**
  * libtransport -- C++ library for easy XMPP Transports development
+ *
  * Copyright (C) 2011, Jan Kaluza <hanzz.k@gmail.com>
+ *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
  * (at your option) any later version.
+ *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
+ *
  * You should have received a copy of the GNU General Public License
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111-1301  USA
  */
 #include "transport/networkpluginserver.h"
 #include "transport/user.h"
 #include "transport/transport.h"
 #include "transport/storagebackend.h"
 #include "transport/rostermanager.h"
 #include "transport/usermanager.h"
 #include "transport/conversationmanager.h"
 #include "transport/localbuddy.h"
 #include "transport/config.h"
 #include "transport/conversation.h"
 #include "transport/vcardresponder.h"
 #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"
 #include "Swiften/Elements/StreamError.h"
 #include "Swiften/Network/BoostConnectionServer.h"
 #include "Swiften/Elements/AttentionPayload.h"
 #include "Swiften/Elements/XHTMLIMPayload.h"
 #include "Swiften/Elements/InvisiblePayload.h"
 #include "Swiften/Elements/SpectrumErrorPayload.h"
 #include "transport/protocol.pb.h"
 #include <Swiften/FileTransfer/ReadBytestream.h>
 #include <Swiften/Elements/StreamInitiationFileInfo.h>
 #ifdef _WIN32
 #include "windows.h"
 #include <stdint.h>
 #else
 #include "sys/wait.h"
 #include "sys/signal.h"
 #include "popt.h"
 #endif
 namespace Transport {
 static unsigned long backend_id;
 static unsigned long bytestream_id;
 DEFINE_LOGGER(logger, "NetworkPluginServer");
 class NetworkConversation : public Conversation {
 	public:
 		NetworkConversation(ConversationManager *conversationManager, const std::string &legacyName, bool muc = false) : Conversation(conversationManager, legacyName, muc) {
+		}
 		// Called when there's new message to legacy network from XMPP network
 		void sendMessage(boost::shared_ptr<Swift::Message> &message) {
 			onMessageToSend(this, message);
+		}
 		boost::signal<void (NetworkConversation *, boost::shared_ptr<Swift::Message> &)> onMessageToSend;
 };
 class NetworkFactory : public Factory {
 	public:
 		NetworkFactory(NetworkPluginServer *nps) {
 			m_nps = nps;
+		}
 		// Creates new conversation (NetworkConversation in this case)
 		Conversation *createConversation(ConversationManager *conversationManager, const std::string &legacyName) {
 			NetworkConversation *nc = new NetworkConversation(conversationManager, legacyName);
 			nc->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, m_nps, _1, _2));
 			return nc;
+		}
 		// Creates new LocalBuddy
 		Buddy *createBuddy(RosterManager *rosterManager, const BuddyInfo &buddyInfo) {
 			LocalBuddy *buddy = new LocalBuddy(rosterManager, buddyInfo.id);
 			buddy->setAlias(buddyInfo.alias);
 			buddy->setName(buddyInfo.legacyName);
 			if (buddyInfo.subscription == "both") {
 				buddy->setSubscription(Buddy::Both);
+			}
 			else {
 				buddy->setSubscription(Buddy::Ask);
+			}
 			buddy->setGroups(buddyInfo.groups);
 			buddy->setFlags((BuddyFlag) (buddyInfo.flags));
 			if (buddyInfo.settings.find("icon_hash") != buddyInfo.settings.end())
 				buddy->setIconHash(buddyInfo.settings.find("icon_hash")->second.s);
 			return buddy;
+		}
 	private:
 		NetworkPluginServer *m_nps;
 };
 // Wraps google protobuf payload into WrapperMessage and serialize it to string
 #define WRAP(MESSAGE, TYPE) 	pbnetwork::WrapperMessage wrap; \
 	wrap.set_type(TYPE); \
 	wrap.set_payload(MESSAGE); \
 	wrap.SerializeToString(&MESSAGE);
 // Executes new backend
 static unsigned long exec_(std::string path, const char *host, const char *port, const char *config) {
 	std::string original_path = path;
 	// BACKEND_ID is replaced with unique ID. The ID is increasing for every backend.
 	boost::replace_all(path, "BACKEND_ID", boost::lexical_cast<std::string>(backend_id++));
 	// Add host and port.
 	path += std::string(" --host ") + host + " --port " + port + " " + config;
 	LOG4CXX_INFO(logger, "Starting new backend " << path);
 #ifdef _WIN32
 	STARTUPINFO         si;
 	PROCESS_INFORMATION pi;
 	ZeroMemory (&si, sizeof(si));
 	si.cb=sizeof (si);
 	if (! CreateProcess(
 	NULL,
 	(LPSTR)path.c_str(),         // command line
 ,                    // process attributes
 ,                    // thread attributes
 ,                    // inherit handles
 ,                    // creation flags
 ,                    // environment
 ,                    // cwd
 	&si,
 	&pi
+	)
 	)  {
 		LOG4CXX_ERROR(logger, "Could not start process");
+	}
 	return 0;
 #else
 	// Create array of char * from string using -lpopt library
 	char *p = (char *) malloc(path.size() + 1);
 	strcpy(p, path.c_str());
 	int argc;
 	char **argv;
 	poptParseArgvString(p, &argc, (const char ***) &argv);
 	// fork and exec
 	pid_t pid = fork();
 	if ( pid == 0 ) {
 		setsid();
 		// child process
 		errno = 0;
 		int ret = execv(argv[0], argv);
 		if (ret == -1) {
 			exit(errno);
+		}
 		exit(0);
 	} else if ( pid < 0 ) {
 		LOG4CXX_ERROR(logger, "Fork failed");
+	}
 	free(p);
 	return (unsigned long) pid;
 #endif
+}
 #ifndef _WIN32
 static void SigCatcher(int n) {
 	pid_t result;
 	int status;
 	// Read exit code from all children to not have zombies arround
 	// WARNING: Do not put LOG4CXX_ here, because it can lead to deadlock
 	while ((result = waitpid(-1, &status, WNOHANG)) > 0) {
 		if (result != 0) {
 			if (WIFEXITED(status)) {
 				if (WEXITSTATUS(status) != 0) {
 // 					LOG4CXX_ERROR(logger, "Backend can not be started, exit_code=" << WEXITSTATUS(status));
+				}
+			}
 			else {
 // 				LOG4CXX_ERROR(logger, "Backend can not be started");
+			}
+		}
+	}
+}
 #endif
 static void handleBuddyPayload(LocalBuddy *buddy, const pbnetwork::Buddy &payload) {
 	buddy->setName(payload.buddyname());
 	// Set alias only if it's not empty. Backends are allowed to send empty alias if it has
 	// not changed.
 	if (!payload.alias().empty()) {
 		LOG4CXX_INFO(logger, "Setting alias to " << payload.alias() << " " << buddy->getAlias());
 		buddy->setAlias(payload.alias());
+	}
 	// Change groups if it's not empty. The same as above...
 	std::vector<std::string> groups;
 	for (int i = 0; i < payload.group_size(); i++) {
 		groups.push_back(payload.group(i));
+	}
 	if (!groups.empty()) {
 		buddy->setGroups(groups);
+	}
 	buddy->setStatus(Swift::StatusShow((Swift::StatusShow::Type) payload.status()), payload.statusmessage());
 	buddy->setIconHash(payload.iconhash());
 	buddy->setBlocked(payload.blocked());
+}
 NetworkPluginServer::NetworkPluginServer(Component *component, Config *config, UserManager *userManager, FileTransferManager *ftManager) {
 	m_ftManager = ftManager;
 	m_userManager = userManager;
 	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));
 	m_pingTimer = component->getNetworkFactories()->getTimerFactory()->createTimer(20000);
 	m_pingTimer->onTick.connect(boost::bind(&NetworkPluginServer::pingTimeout, this));
 	m_pingTimer->start();
 	if (CONFIG_INT(m_config, "service.memory_collector_time") != 0) {
 		m_collectTimer = component->getNetworkFactories()->getTimerFactory()->createTimer(CONFIG_INT(m_config, "service.memory_collector_time"));
 		m_collectTimer->onTick.connect(boost::bind(&NetworkPluginServer::collectBackend, this));
 		m_collectTimer->start();
+	}
 	m_vcardResponder = new VCardResponder(component->getIQRouter(), component->getNetworkFactories(), userManager);
 	m_vcardResponder->onVCardRequired.connect(boost::bind(&NetworkPluginServer::handleVCardRequired, this, _1, _2, _3));
 	m_vcardResponder->onVCardUpdated.connect(boost::bind(&NetworkPluginServer::handleVCardUpdated, this, _1, _2));
 	m_vcardResponder->start();
 	m_rosterResponder = new RosterResponder(component->getIQRouter(), userManager);
 	m_rosterResponder->onBuddyAdded.connect(boost::bind(&NetworkPluginServer::handleBuddyAdded, this, _1, _2));
 	m_rosterResponder->onBuddyRemoved.connect(boost::bind(&NetworkPluginServer::handleBuddyRemoved, this, _1));
 	m_rosterResponder->onBuddyUpdated.connect(boost::bind(&NetworkPluginServer::handleBuddyUpdated, this, _1, _2));
 	m_rosterResponder->start();
 	m_blockResponder = new BlockResponder(component->getIQRouter(), userManager);
 	m_blockResponder->onBlockToggled.connect(boost::bind(&NetworkPluginServer::handleBlockToggled, this, _1));
 	m_blockResponder->start();
 	m_server = component->getNetworkFactories()->getConnectionServerFactory()->createConnectionServer(Swift::HostAddress(CONFIG_STRING(m_config, "service.backend_host")), boost::lexical_cast<int>(CONFIG_STRING(m_config, "service.backend_port")));
 	m_server->onNewConnection.connect(boost::bind(&NetworkPluginServer::handleNewClientConnection, this, _1));
 	m_server->start();
 	LOG4CXX_INFO(logger, "Listening on host " << CONFIG_STRING(m_config, "service.backend_host") << " port " << CONFIG_STRING(m_config, "service.backend_port"));
 	unsigned long pid = exec_(CONFIG_STRING(m_config, "service.backend"), CONFIG_STRING(m_config, "service.backend_host").c_str(), CONFIG_STRING(m_config, "service.backend_port").c_str(), m_config->getConfigFile().c_str());
 	LOG4CXX_INFO(logger, "Tried to spawn first backend with pid " << pid);
 	LOG4CXX_INFO(logger, "Backend should now connect to Spectrum2 instance. Spectrum2 won't accept any connection before backend connects");
 #ifndef _WIN32
 	// wait if the backend process will still be alive after 1 second
 	sleep(1);
 	pid_t result;
 	int status;
 	result = waitpid(-1, &status, WNOHANG);
 	if (result != 0) {
 		if (WIFEXITED(status)) {
 			if (WEXITSTATUS(status) != 0) {
 				LOG4CXX_ERROR(logger, "Backend can not be started, exit_code=" << WEXITSTATUS(status) << ", possible error: " << strerror(WEXITSTATUS(status)));
+			}
+		}
 		else {
 			LOG4CXX_ERROR(logger, "Backend can not be started");
+		}
+	}
 	signal(SIGCHLD, SigCatcher);
 #endif
+}
 NetworkPluginServer::~NetworkPluginServer() {
 	for (std::list<Backend *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 		LOG4CXX_INFO(logger, "Stopping backend " << *it);
 		std::string message;
 		pbnetwork::WrapperMessage wrap;
 		wrap.set_type(pbnetwork::WrapperMessage_Type_TYPE_EXIT);
 		wrap.SerializeToString(&message);
 		Backend *c = (Backend *) *it;
 		send(c->connection, message);
+	}
 	m_pingTimer->stop();
 	m_server->stop();
 	m_server.reset();
 	delete m_component->m_factory;
 	delete m_vcardResponder;
 	delete m_rosterResponder;
 	delete m_blockResponder;
+}
 void NetworkPluginServer::handleNewClientConnection(boost::shared_ptr<Swift::Connection> c) {
 	// Create new Backend instance
 	Backend *client = new Backend;
 	client->pongReceived = -1;
 	client->connection = c;
 	client->res = 0;
 	client->init_res = 0;
 	client->shared = 0;
 	// Until we receive first PONG from backend, backend is in willDie state.
 	client->willDie = true;
 	// Backend does not accept new clients automatically if it's long-running
 	client->acceptUsers = !m_isNextLongRun;
 	client->longRun = m_isNextLongRun;
 	LOG4CXX_INFO(logger, "New" + (client->longRun ? std::string(" long-running") : "") +  " backend " << client << " connected. Current backend count=" << (m_clients.size() + 1));
 	m_clients.push_front(client);
 	c->onDisconnected.connect(boost::bind(&NetworkPluginServer::handleSessionFinished, this, client));
 	c->onDataRead.connect(boost::bind(&NetworkPluginServer::handleDataRead, this, client, _1));
 	sendPing(client);
 	// sendPing sets pongReceived to 0, but we want to have it -1 to ignore this backend
 	// in first ::pingTimeout call, because it can be called right after this function
 	// and backend wouldn't have any time to response to ping.
 	client->pongReceived = -1;
+}
 void NetworkPluginServer::handleSessionFinished(Backend *c) {
 	LOG4CXX_INFO(logger, "Backend " << c << " (ID=" << c->id << ") disconnected. Current backend count=" << (m_clients.size() - 1));
 	// This backend will do, so we can't reconnect users to it in User::handleDisconnected call
 	c->willDie = true;
 	// If there are users associated with this backend, it must have crashed, so print error output
 	// and disconnect users
 	if (!c->users.empty()) {
 		m_crashedBackends.push_back(c->id);
+	}
 	for (std::list<User *>::const_iterator it = c->users.begin(); it != c->users.end(); it++) {
 		LOG4CXX_ERROR(logger, "Backend " << c << " (ID=" << c->id << ") disconnected (probably crashed) with active user " << (*it)->getJID().toString());
 		(*it)->setData(NULL);
 		(*it)->handleDisconnected("Internal Server Error, please reconnect.");
+	}
 	std::string message;
 	pbnetwork::WrapperMessage wrap;
 	wrap.set_type(pbnetwork::WrapperMessage_Type_TYPE_EXIT);
 	wrap.SerializeToString(&message);
 	send(c->connection, message);
 	c->connection->onDisconnected.disconnect_all_slots();
 	c->connection->onDataRead.disconnect_all_slots();
 	c->connection->disconnect();
 	c->connection.reset();
 	m_clients.remove(c);
 	delete c;
+}
 void NetworkPluginServer::handleConnectedPayload(const std::string &data) {
 	pbnetwork::Connected payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	User *user = m_userManager->getUser(payload.user());
 	if (!user) {
 		LOG4CXX_ERROR(logger, "Connected payload received for unknown user " << payload.user());
 		return;
+	}
 	user->setConnected(true);
 	m_component->m_userRegistry->onPasswordValid(payload.user());
+}
 void NetworkPluginServer::handleDisconnectedPayload(const std::string &data) {
 	pbnetwork::Disconnected payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	m_component->m_userRegistry->onPasswordInvalid(payload.user(), payload.message());
 	User *user = m_userManager->getUser(payload.user());
 	if (!user) {
 		return;
+	}
 	user->handleDisconnected(payload.message(), (Swift::SpectrumErrorPayload::Error) payload.error());
+}
 void NetworkPluginServer::handleVCardPayload(const std::string &data) {
 	pbnetwork::VCard payload;
 	if (payload.ParseFromString(data) == false) {
 		std::cout << "PARSING ERROR\n";
 		// TODO: ERROR
 		return;
+	}
 	boost::shared_ptr<Swift::VCard> vcard(new Swift::VCard());
 	vcard->setFullName(payload.fullname());
 	vcard->setPhoto(Swift::createByteArray(payload.photo()));
 	vcard->setNickname(payload.nickname());
 	m_vcardResponder->sendVCard(payload.id(), vcard);
+}
 	// Add xhtml-im payload.
 	if (!payload.xhtml().empty()) {
 		msg->addPayload(boost::make_shared<Swift::XHTMLIMPayload>(payload.xhtml()));
+	}
 	// Create new Conversation if it does not exist
 	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.buddyname());
 	if (!conv) {
 		conv = new NetworkConversation(user->getConversationManager(), payload.buddyname());
 		conv->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, this, _1, _2));
+	}
 	// Forward it
 	conv->handleMessage(msg, payload.nickname());
 	m_userManager->messageToXMPPSent();
+}
 void NetworkPluginServer::handleAttentionPayload(const std::string &data) {
 	pbnetwork::ConversationMessage payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	User *user = m_userManager->getUser(payload.username());
 	if (!user)
 		return;
 	boost::shared_ptr<Swift::Message> msg(new Swift::Message());
 	msg->setBody(payload.message());
 	msg->addPayload(boost::make_shared<Swift::AttentionPayload>());
 	// Attentions trigger new Conversation creation
 	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.buddyname());
 	if (!conv) {
 		conv = new NetworkConversation(user->getConversationManager(), payload.buddyname());
 		conv->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, this, _1, _2));
+	}
 	conv->handleMessage(msg);
+}
 void NetworkPluginServer::handleStatsPayload(Backend *c, const std::string &data) {
 	pbnetwork::Stats payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	c->res = payload.res();
 	c->init_res = payload.init_res();
 	c->shared = payload.shared();
 	c->id = payload.id();
+}
 void NetworkPluginServer::handleFTStartPayload(const std::string &data) {
 	pbnetwork::File payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	User *user = m_userManager->getUser(payload.username());
 	if (!user)
 		return;
 	LOG4CXX_INFO(logger, "handleFTStartPayload " << payload.filename() << " " << payload.buddyname());
 	LocalBuddy *buddy = (LocalBuddy *) user->getRosterManager()->getBuddy(payload.buddyname());
 	if (!buddy) {
 		// TODO: escape? reject?
 		return;
+	}
 	Swift::StreamInitiationFileInfo fileInfo;
 	fileInfo.setSize(payload.size());
 	fileInfo.setName(payload.filename());
 	Backend *c = (Backend *) user->getData();
 	boost::shared_ptr<MemoryReadBytestream> bytestream(new MemoryReadBytestream(payload.size()));
 	bytestream->onDataNeeded.connect(boost::bind(&NetworkPluginServer::handleFTDataNeeded, this, c, bytestream_id + 1));
 	LOG4CXX_INFO(logger, "jid=" << buddy->getJID());
 	FileTransferManager::Transfer transfer = m_ftManager->sendFile(user, buddy, bytestream, fileInfo);
 	if (!transfer.ft) {
 		handleFTRejected(user, payload.buddyname(), payload.filename(), payload.size());
 		return;
+	}
 	m_filetransfers[++bytestream_id] = transfer;
 	transfer.ft->onStateChange.connect(boost::bind(&NetworkPluginServer::handleFTStateChanged, this, _1, payload.username(), payload.buddyname(), payload.filename(), payload.size(), bytestream_id));
 	transfer.ft->start();
+}
 void NetworkPluginServer::handleFTFinishPayload(const std::string &data) {
 	pbnetwork::File payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 	if (payload.has_ftid()) {
 		if (m_filetransfers.find(payload.ftid()) != m_filetransfers.end()) {
 			FileTransferManager::Transfer &transfer = m_filetransfers[payload.ftid()];
 			transfer.ft->cancel();
+		}
 		else {
 			LOG4CXX_ERROR(logger, "FTFinishPayload for unknown ftid=" << payload.ftid());
+		}
+	}
+}
 void NetworkPluginServer::handleFTDataPayload(Backend *b, const std::string &data) {
 	pbnetwork::FileTransferData payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
 		return;
+	}
 // 	User *user = m_userManager->getUser(payload.username());
 // 	if (!user)
 // 		return;
 	FileTransferManager::Transfer &transfer = m_filetransfers[payload.ftid()];
 	MemoryReadBytestream *bytestream = (MemoryReadBytestream *) transfer.readByteStream.get();
 	if (bytestream->appendData(payload.data()) > 5000000) {
 		pbnetwork::FileTransferData f;
 		f.set_ftid(payload.ftid());
 		f.set_data("");
 		std::string message;
 		f.SerializeToString(&message);
 		WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_FT_PAUSE);
 		send(b->connection, message);
+	}
+}
 void NetworkPluginServer::handleFTDataNeeded(Backend *b, unsigned long ftid) {
 	pbnetwork::FileTransferData f;
 	f.set_ftid(ftid);
 	f.set_data("");
 	std::string message;
 	f.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_FT_CONTINUE);
 	send(b->connection, message);
+}
 void NetworkPluginServer::handlePongReceived(Backend *c) {
 	// This could be first PONG from the backend
 	if (c->pongReceived == -1) {
 		// Backend is fully ready to handle requests
 		c->willDie = false;
 		if (m_clients.size() == 1) {
 			// first backend connected, start the server, we're ready.
 			m_component->start();
+		}
 		// some users are in queue waiting for this backend
 		while(!m_waitingUsers.empty()) {
 			// There's no new backend, so stop associating users and wait for new backend,
 			// which has been already spawned in getFreeClient() call.
 			if (getFreeClient() == NULL)
 				break;
 			User *u = m_waitingUsers.front();
 			m_waitingUsers.pop_front();
 			LOG4CXX_INFO(logger, "Associating " << u->getJID().toString() << " with this backend");
 			// associate backend with user
 			handleUserCreated(u);
 			// connect user if it's ready
 			if (u->isReadyToConnect()) {
 				handleUserReadyToConnect(u);
+			}
+		}
+	}
 	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());
 	// Parse data while there are some
 	while (c->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 (c->data.size() >= 4) {
 			expected_size = *((unsigned int*) &c->data[0]);
 			expected_size = ntohl(expected_size);
 			// If we don't have whole wrapper message, wait for next
 			// handleDataRead call.
 			if (c->data.size() - 4 < expected_size)
 				return;
+		}
 		else {
 			return;
+		}
 		// Parse wrapper message and erase it from buffer.
 		pbnetwork::WrapperMessage wrapper;
 		if (wrapper.ParseFromArray(&c->data[4], expected_size) == false) {
 			std::cout << "PARSING ERROR " << expected_size << "\n";
 			c->data.erase(c->data.begin(), c->data.begin() + 4 + expected_size);
 			continue;
+		}
 		c->data.erase(c->data.begin(), c->data.begin() + 4 + expected_size);
 		// Handle payload in wrapper message
 		switch(wrapper.type()) {
 			case pbnetwork::WrapperMessage_Type_TYPE_CONNECTED:
 				handleConnectedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_DISCONNECTED:
 				handleDisconnectedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_CHANGED:
 				handleBuddyChangedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_CONV_MESSAGE:
 				handleConvMessagePayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_ROOM_SUBJECT_CHANGED:
 				handleConvMessagePayload(wrapper.payload(), true);
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_PONG:
 				handlePongReceived(c);
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_PARTICIPANT_CHANGED:
 				handleParticipantChangedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_ROOM_NICKNAME_CHANGED:
 				handleRoomChangedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_VCARD:
 				handleVCardPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPING:
 				handleChatStatePayload(wrapper.payload(), Swift::ChatState::Composing);
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPED:
 				handleChatStatePayload(wrapper.payload(), Swift::ChatState::Paused);
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_STOPPED_TYPING:
 				handleChatStatePayload(wrapper.payload(), Swift::ChatState::Active);
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_AUTH_REQUEST:
 				handleAuthorizationPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_ATTENTION:
 				handleAttentionPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_STATS:
 				handleStatsPayload(c, wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_FT_START:
 				handleFTStartPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_FT_FINISH:
 				handleFTFinishPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_FT_DATA:
 				handleFTDataPayload(c, wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_REMOVED:
 				handleBuddyRemovedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_QUERY:
 				handleQueryPayload(c, wrapper.payload());
 				break;
 			default:
 				return;
+		}
+	}
+}
 void NetworkPluginServer::send(boost::shared_ptr<Swift::Connection> &c, const std::string &data) {
 	// generate header - size of wrapper message
 	uint32_t size = htonl(data.size());
 	char *header = (char *) &size;
 	// send header together with wrapper message
 	c->write(Swift::createSafeByteArray(std::string(header, 4) + data));
+}
 void NetworkPluginServer::pingTimeout() {
 	// TODO: move to separate timer, those 2 loops could be expensive
 	// Some users are connected for weeks and they are blocking backend to be destroyed and its memory
 	// to be freed. We are finding users who are inactive for more than "idle_reconnect_time" seconds and
 	// reconnect them to long-running backend, where they can idle hapilly till the end of ages.
 	time_t now = time(NULL);
 	std::vector<User *> usersToMove;
 	unsigned long diff = CONFIG_INT(m_config, "service.idle_reconnect_time");
 	if (diff != 0) {
 		for (std::list<Backend *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 			// Users from long-running backends can't be moved
 			if ((*it)->longRun) {
 				continue;
+			}
 			// Find users which are inactive for more than 'diff'
 			BOOST_FOREACH(User *u, (*it)->users) {
 				if (now - u->getLastActivity() > diff) {
 					usersToMove.push_back(u);
+				}
+			}
+		}
 		// Move inactive users to long-running backend.
 		BOOST_FOREACH(User *u, usersToMove) {
 			LOG4CXX_INFO(logger, "Moving user " << u->getJID().toString() << " to long-running backend");
 			if (!moveToLongRunBackend(u))
 				break;
+		}
+	}
 	// check ping responses
 	std::vector<Backend *> toRemove;
 	for (std::list<Backend *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 		// pong has been received OR backend just connected and did not have time to answer the ping
 		// request.
 		if ((*it)->pongReceived || (*it)->pongReceived == -1) {
 			// Don't send another ping if pongReceived == -1, because we've already sent one
 			// when registering backend.
 			if ((*it)->pongReceived) {
 				sendPing((*it));
+			}
+		}
 		else {
 			LOG4CXX_INFO(logger, "Disconnecting backend " << (*it) << " (ID=" << (*it)->id << "). PING response not received.");
 			toRemove.push_back(*it);
+		}
 		if ((*it)->users.size() == 0) {
 			LOG4CXX_INFO(logger, "Disconnecting backend " << (*it) << " (ID=" << (*it)->id << "). There are no users.");
 			toRemove.push_back(*it);
+		}
+	}
 	BOOST_FOREACH(Backend *b, toRemove) {
 		handleSessionFinished(b);
+	}
 	m_pingTimer->start();
+}
 void NetworkPluginServer::collectBackend() {
 	// Stop accepting new users to backend with the biggest memory usage. This prevents backends
 	// which are leaking to eat whole memory by connectin new users to legacy network.
 	LOG4CXX_INFO(logger, "Collect backend called, finding backend which will be set to die");
 	unsigned long max = 0;
 	Backend *backend = NULL;
 	for (std::list<Backend *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 		if ((*it)->res > max) {
 			max = (*it)->res;
 			backend = (*it);
+		}
+	}
 	if (backend) {
 		if (m_collectTimer) {
 			m_collectTimer->start();
+		}
 		LOG4CXX_INFO(logger, "Backend " << backend << " (ID=" << backend->id << ") is set to die");
 		backend->acceptUsers = false;
+	}
+}
 bool NetworkPluginServer::moveToLongRunBackend(User *user) {
 	// Check if user has already some backend
 	Backend *old = (Backend *) user->getData();
 	if (!old) {
 		LOG4CXX_INFO(logger, "User " << user->getJID().toString() << " does not have old backend. Not moving.");
 		return true;
+	}
 	// if he's already on long run, do nothing
 	if (old->longRun) {
 		LOG4CXX_INFO(logger, "User " << user->getJID().toString() << " is already on long-running backend. Not moving.");
 		return true;
+	}
 	// Get free longrun backend, if there's no longrun backend, create one and wait
 	// for its connection
 	Backend *backend = getFreeClient(false, true);
 	if (!backend) {
 		LOG4CXX_INFO(logger, "No free long-running backend for user " << user->getJID().toString() << ". Will try later");
 		return false;
+	}
 	// old backend will trigger disconnection which has to be ignored to keep user online
 	user->setIgnoreDisconnect(true);
 	// remove user from the old backend
 	// If backend is empty, it will be collected by pingTimeout
 	old->users.remove(user);
 	// switch to new backend and connect
 	user->setData(backend);
 	backend->users.push_back(user);
 	// connect him
 	handleUserReadyToConnect(user);
 	return true;
+}
 void NetworkPluginServer::handleUserCreated(User *user) {
 	// Get free backend to handle this user or spawn new one if there's no free one.
 	Backend *c = getFreeClient();
 	// Add user to queue if there's no free backend to handle him so far.
 	if (!c) {
 		LOG4CXX_INFO(logger, "There is no backend to handle user " << user->getJID().toString() << ". Adding him to queue.");
 		m_waitingUsers.push_back(user);
 		return;
+	}
 	// Associate users with backend
 	user->setData(c);
 	c->users.push_back(user);
 	// Don't forget to disconnect these in handleUserDestroyed!!!
 	user->onReadyToConnect.connect(boost::bind(&NetworkPluginServer::handleUserReadyToConnect, this, user));
 	user->onPresenceChanged.connect(boost::bind(&NetworkPluginServer::handleUserPresenceChanged, this, user, _1));
 	user->onRoomJoined.connect(boost::bind(&NetworkPluginServer::handleRoomJoined, this, user, _1, _2, _3, _4));
 	user->onRoomLeft.connect(boost::bind(&NetworkPluginServer::handleRoomLeft, this, user, _1));
+}
 void NetworkPluginServer::handleUserReadyToConnect(User *user) {
 	UserInfo userInfo = user->getUserInfo();
 	pbnetwork::Login login;
 	login.set_user(user->getJID().toBare());
 	login.set_legacyname(userInfo.uin);
 	login.set_password(userInfo.password);
 	std::string message;
 	login.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LOGIN);
 	Backend *c = (Backend *) user->getData();
 	if (!c) {
 		return;
+	}
 	send(c->connection, message);
+}
 void NetworkPluginServer::handleUserPresenceChanged(User *user, Swift::Presence::ref presence) {
 	if (presence->getShow() == Swift::StatusShow::None)
 		return;
 	UserInfo userInfo = user->getUserInfo();
 	pbnetwork::Status status;
 	status.set_username(user->getJID().toBare());
 	bool isInvisible = presence->getPayload<Swift::InvisiblePayload>() != NULL;
 	if (isInvisible) {
 		LOG4CXX_INFO(logger, "This presence is invisible");
 		status.set_status((pbnetwork::STATUS_INVISIBLE));

0 comments (0 inline, 0 general)