Changeset - 2ae5cba932b1
[Not reviewed]
0 3 0
HanzZ - 13 years ago 2012-12-27 15:02:21
hanzz.k@gmail.com
service.login_delay
3 files changed with 44 insertions and 14 deletions:
0 comments (0 inline, 0 general)
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 DiscoItemsResponder;
 

	
 
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, DiscoItemsResponder *discoItemsResponder);
 

	
 
		virtual ~NetworkPluginServer();
 

	
 
		void start();
 

	
 
		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);
 

	
 
	public:
 
		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 handleConvMessageAckPayload(const std::string &payload);
 
		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 handleBackendConfigPayload(const std::string &payload);
 
		void handleRoomListPayload(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);
 

	
 
	private:
 
		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);
 
		Backend *getFreeClient(bool acceptUsers = true, bool longRun = false, bool check = false);
 
		void connectWaitingUsers();
 
		void loginDelayFinished();
 

	
 
		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;
 
		Swift::Timer::ref m_loginTimer;
 
		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;
 
		bool m_startingBackend;
 
		DiscoItemsResponder *m_discoItemsResponder;
 
		time_t m_lastLogin;
 
};
 

	
 
}
src/config.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/config.h"
 
#include <fstream>
 
#ifdef _WIN32
 
#include <direct.h>
 
#define getcwd _getcwd
 
#include <windows.h>
 
#ifdef _MSC_VER
 
#define PATH_MAX MAX_PATH
 
#endif
 
#endif
 

	
 
#include "iostream"
 
#include "boost/version.hpp"
 
#include "boost/algorithm/string.hpp"
 

	
 
#define BOOST_MAJOR_VERSION BOOST_VERSION / 100000
 
#define BOOST_MINOR_VERSION BOOST_VERSION / 100 % 1000
 

	
 
using namespace boost::program_options;
 

	
 
namespace Transport {
 
static int getRandomPort(const std::string &s) {
 
	unsigned long r = 0;
 
	BOOST_FOREACH(char c, s) {
 
		r += (int) c;
 
	}
 
	srand(time(NULL) + r);
 
	return 30000 + rand() % 10000;
 
}
 

	
 
bool Config::load(const std::string &configfile, boost::program_options::options_description &opts, const std::string &jid) {
 
	std::ifstream ifs(configfile.c_str());
 
	if (!ifs.is_open())
 
		return false;
 

	
 
	m_file = configfile;
 
	m_jid = jid;
 
	bool ret = load(ifs, opts, jid);
 
	ifs.close();
 
#ifndef WIN32
 
	char path[PATH_MAX] = "";
 
	if (m_file.find_first_of("/") != 0) {
 
		getcwd(path, PATH_MAX);
 
		m_file = std::string(path) + "/" + m_file;
 
	}
 
#endif
 

	
 
	return ret;
 
}
 

	
 
bool Config::load(std::istream &ifs, boost::program_options::options_description &opts, const std::string &_jid) {
 
	m_unregistered.clear();
 
	opts.add_options()
 
		("service.jid", value<std::string>()->default_value(""), "Transport Jabber ID")
 
		("service.server", value<std::string>()->default_value(""), "Server to connect to")
 
		("service.password", value<std::string>()->default_value(""), "Password used to auth the server")
 
		("service.port", value<int>()->default_value(0), "Port the server is listening on")
 
		("service.user", value<std::string>()->default_value(""), "The name of user Spectrum runs as.")
 
		("service.group", value<std::string>()->default_value(""), "The name of group Spectrum runs as.")
 
		("service.backend", value<std::string>()->default_value("libpurple_backend"), "Backend")
 
		("service.protocol", value<std::string>()->default_value(""), "Protocol")
 
		("service.pidfile", value<std::string>()->default_value("/var/run/spectrum2/$jid.pid"), "Full path to pid file")
 
		("service.portfile", value<std::string>()->default_value("/var/run/spectrum2/$jid.port"), "File to store backend_port to. It's used by spectrum2_manager.")
 
		("service.working_dir", value<std::string>()->default_value("/var/lib/spectrum2/$jid"), "Working dir")
 
		("service.allowed_servers", value<std::vector<std::string> >()->multitoken(), "Only users from these servers can connect")
 
		("service.server_mode", value<bool>()->default_value(false), "True if Spectrum should behave as server")
 
		("service.users_per_backend", value<int>()->default_value(100), "Number of users per one legacy network backend")
 
		("service.backend_host", value<std::string>()->default_value("localhost"), "Host to bind backend server to")
 
		("service.backend_port", value<std::string>()->default_value("0"), "Port to bind backend server to")
 
		("service.cert", value<std::string>()->default_value(""), "PKCS#12 Certificate.")
 
		("service.cert_password", value<std::string>()->default_value(""), "PKCS#12 Certificate password.")
 
		("service.admin_jid", value<std::vector<std::string> >()->multitoken(), "Administrator jid.")
 
		("service.admin_password", value<std::string>()->default_value(""), "Administrator password.")
 
		("service.reuse_old_backends", value<bool>()->default_value(true), "True if Spectrum should use old backends which were full in the past.")
 
		("service.idle_reconnect_time", value<int>()->default_value(0), "Time in seconds after which idle users are reconnected to let their backend die.")
 
		("service.memory_collector_time", value<int>()->default_value(0), "Time in seconds after which backend with most memory is set to die.")
 
		("service.more_resources", value<bool>()->default_value(false), "Allow more resources to be connected in server mode at the same time.")
 
		("service.enable_privacy_lists", value<bool>()->default_value(true), "")
 
		("service.enable_xhtml", value<bool>()->default_value(true), "")
 
		("service.max_room_list_size", value<int>()->default_value(100), "")
 
		("service.login_delay", value<int>()->default_value(0), "")
 
		("service.jid_escaping", value<bool>()->default_value(true), "")
 
		("service.vip_only", value<bool>()->default_value(false), "")
 
		("service.vip_message", value<std::string>()->default_value(""), "")
 
		("vhosts.vhost", value<std::vector<std::string> >()->multitoken(), "")
 
		("identity.name", value<std::string>()->default_value("Spectrum 2 Transport"), "Name showed in service discovery.")
 
		("identity.category", value<std::string>()->default_value("gateway"), "Disco#info identity category. 'gateway' by default.")
 
		("identity.type", value<std::string>()->default_value(""), "Type of transport ('icq','msn','gg','irc', ...)")
 
		("registration.enable_public_registration", value<bool>()->default_value(true), "True if users should be able to register.")
 
		("registration.language", value<std::string>()->default_value("en"), "Default language for registration form")
 
		("registration.instructions", value<std::string>()->default_value("Enter your legacy network username and password."), "Instructions showed to user in registration form")
 
		("registration.username_label", value<std::string>()->default_value("Legacy network username:"), "Label for username field")
 
		("registration.username_mask", value<std::string>()->default_value(""), "Username mask")
 
		("registration.allowed_usernames", value<std::string>()->default_value(""), "Allowed usernames")
 
		("registration.auto_register", value<bool>()->default_value(false), "Register new user automatically when the presence arrives.")
 
		("registration.encoding", value<std::string>()->default_value("utf8"), "Default encoding in registration form")
 
		("registration.require_local_account", value<bool>()->default_value(false), "True if users have to have a local account to register to this transport from remote servers.")
 
		("registration.notify_jid", value<std::vector<std::string> >()->multitoken(), "Send message to this JID if user registers/unregisters")
 
		("registration.local_username_label", value<std::string>()->default_value("Local username:"), "Label for local usernme field")
 
		("registration.local_account_server", value<std::string>()->default_value("localhost"), "The server on which the local accounts will be checked for validity")
 
		("registration.local_account_server_timeout", value<int>()->default_value(10000), "Timeout when checking local user on local_account_server (msecs)")
 
		("gateway_responder.prompt", value<std::string>()->default_value("Contact ID"), "Value of <prompt> </promt> field")
 
		("gateway_responder.label", value<std::string>()->default_value("Enter legacy network contact ID."), "Label for add contact ID field")
 
		("database.type", value<std::string>()->default_value("none"), "Database type.")
 
		("database.database", value<std::string>()->default_value("/var/lib/spectrum2/$jid/database.sql"), "Database used to store data")
 
		("database.server", value<std::string>()->default_value("localhost"), "Database server.")
 
		("database.user", value<std::string>()->default_value(""), "Database user.")
 
		("database.password", value<std::string>()->default_value(""), "Database Password.")
 
		("database.port", value<int>()->default_value(0), "Database port.")
 
		("database.prefix", value<std::string>()->default_value(""), "Prefix of tables in database")
 
		("database.encryption_key", value<std::string>()->default_value(""), "Encryption key.")
 
		("database.vip_statement", value<std::string>()->default_value(""), "Encryption key.")
 
		("logging.config", value<std::string>()->default_value(""), "Path to log4cxx config file which is used for Spectrum 2 instance")
 
		("logging.backend_config", value<std::string>()->default_value(""), "Path to log4cxx config file which is used for backends")
 
		("backend.default_avatar", value<std::string>()->default_value(""), "Full path to default avatar")
 
		("backend.avatars_directory", value<std::string>()->default_value(""), "Path to directory with avatars")
 
		("backend.no_vcard_fetch", value<bool>()->default_value(false), "True if VCards for buddies should not be fetched. Only avatars will be forwarded.")
 
		("proxy.server", value<std::string>()->default_value("localhost"), "Proxy IP.")
 
		("proxy.user", value<std::string>()->default_value(""), "Proxy user.")
 
		("proxy.password", value<std::string>()->default_value(""), "Proxy Password.")
 
		("proxy.port", value<int>()->default_value(0), "Proxy port.")
 

	
 
	;
 

	
 
	parsed_options parsed = parse_config_file(ifs, opts, true);
 

	
 
	bool found_working = false;
 
	bool found_pidfile = false;
 
	bool found_portfile = false;
 
	bool found_backend_port = false;
 
	bool found_database = false;
 
	std::string jid = "";
 
	BOOST_FOREACH(option &opt, parsed.options) {
 
		if (opt.string_key == "service.jid") {
 
			if (_jid.empty()) {
 
				jid = opt.value[0];
 
			}
 
			else {
 
				opt.value[0] = _jid;
 
				jid = _jid;
 
			}
 
		}
 
		else if (opt.string_key == "service.backend_port") {
 
			found_backend_port = true;
 
			if (opt.value[0] == "0") {
 
				opt.value[0] = boost::lexical_cast<std::string>(getRandomPort(_jid.empty() ? jid : _jid));
 
			}
 
		}
 
		else if (opt.string_key == "service.working_dir") {
 
			found_working = true;
 
		}
 
		else if (opt.string_key == "service.pidfile") {
 
			found_pidfile = true;
 
		}
 
		else if (opt.string_key == "service.portfile") {
 
			found_portfile = true;
 
		}
 
		else if (opt.string_key == "database.database") {
 
			found_database = true;
 
		}
 
	}
 

	
 
	if (!found_working) {
 
		std::vector<std::string> value;
 
		value.push_back("/var/lib/spectrum2/$jid");
 
		parsed.options.push_back(boost::program_options::basic_option<char>("service.working_dir", value));
 
	}
 
	if (!found_pidfile) {
 
		std::vector<std::string> value;
 
		value.push_back("/var/run/spectrum2/$jid.pid");
 
		parsed.options.push_back(boost::program_options::basic_option<char>("service.pidfile", value));
 
	}
 
	if (!found_portfile) {
 
		std::vector<std::string> value;
 
		value.push_back("/var/run/spectrum2/$jid.port");
 
		parsed.options.push_back(boost::program_options::basic_option<char>("service.portfile", value));
 
	}
 
	if (!found_backend_port) {
 
		std::vector<std::string> value;
 
		std::string p = boost::lexical_cast<std::string>(getRandomPort(_jid.empty() ? jid : _jid));
 
		value.push_back(p);
 
		parsed.options.push_back(boost::program_options::basic_option<char>("service.backend_port", value));
 
	}
 
	if (!found_database) {
 
		std::vector<std::string> value;
 
		value.push_back("/var/lib/spectrum2/$jid/database.sql");
 
		parsed.options.push_back(boost::program_options::basic_option<char>("database.database", value));
 
	}
 

	
 
	std::list<std::string> has_key;
 
	BOOST_FOREACH(option &opt, parsed.options) {
 
		if (opt.unregistered) {
 
			if (std::find(has_key.begin(), has_key.end(), opt.string_key) == has_key.end()) {
 
				has_key.push_back(opt.string_key);
 
				m_unregistered[opt.string_key] = variable_value(opt.value[0], false);
 
			}
 
			else {
 
				std::list<std::string> list;
 
				try {
 
					list = m_unregistered[opt.string_key].as<std::list<std::string> >();
 
				}
 
				catch(...) {
 
					list.push_back(m_unregistered[opt.string_key].as<std::string>());
 
				}
 
				
 
				list.push_back(opt.value[0]);
 
				m_unregistered[opt.string_key] = variable_value(list, false);
 
			}
 
		}
 
		else if (opt.value[0].find("$jid") != std::string::npos) {
 
			boost::replace_all(opt.value[0], "$jid", jid);
 
		}
 
	}
 

	
 
	// Load configs passed by command line
 
	if (m_argc != 0 && m_argv) {
 
		basic_command_line_parser<char> parser = command_line_parser(m_argc, m_argv).options(opts).allow_unregistered();
 
		parsed_options parsed = parser.run();
 
		BOOST_FOREACH(option &opt, parsed.options) {
 
			if (opt.unregistered && !opt.value.empty()) {
 
				m_unregistered[opt.string_key] = variable_value(opt.value[0], false);
 
			}
 
		}
 
		store(parsed, m_variables);
 
	}
 

	
 
	store(parsed, m_variables);
 
	notify(m_variables);
 

	
 
	onConfigReloaded();
 

	
 
	return true;
 
}
 

	
 
bool Config::load(std::istream &ifs) {
 
	options_description opts("Transport options");
 
	return load(ifs, opts);
 
}
 

	
 
bool Config::load(const std::string &configfile, const std::string &jid) {
 
	try {
 
		options_description opts("Transport options");
 
		return load(configfile, opts, jid);
 
	} catch ( const boost::program_options::multiple_occurrences& e ) {
 
#if (BOOST_MAJOR_VERSION >= 1 && BOOST_MINOR_VERSION >= 42)
 
		std::cerr << configfile << " parsing error: " << e.what() << " from option: " << e.get_option_name() << std::endl;
 
#else
 
		std::cerr << configfile << " parsing error: " << e.what() << std::endl;
 
#endif
 
		return false;
 
	}
 
}
 

	
 
bool Config::reload() {
 
	if (m_file.empty()) {
 
		return false;
 
	}
 

	
 
	return load(m_file, m_jid);
 
}
 

	
 
Config::SectionValuesCont Config::getSectionValues(const std::string& sectionName) {
 
	SectionValuesCont sectionValues;
 

	
 
	std::string sectionSearchString = sectionName + ".";
 
	BOOST_FOREACH (const Variables::value_type & varItem, m_variables) {
 
		if (boost::istarts_with(varItem.first, sectionSearchString))
 
			sectionValues[varItem.first] = varItem.second;
 
	}
 

	
 
	BOOST_FOREACH (const UnregisteredCont::value_type & varItem, m_unregistered) {
 
		if (boost::istarts_with(varItem.first, sectionSearchString))
 
			sectionValues[varItem.first] = varItem.second;
 
	}
 

	
 
	return sectionValues;
 
}
 

	
 
std::string Config::getCommandLineArgs() const {
 
	std::ostringstream commandLineArgs;
 

	
 
	// Return the command-line arguments that were passed to us originally (but remove the initial .exe part)
 
	for (int i = 1; i < m_argc; ++i) 	{
 
		commandLineArgs << "\"" << m_argv[i] << "\" ";
 
	}
 

	
 
	return commandLineArgs.str();
 
}
 

	
 
void Config::updateBackendConfig(const std::string &backendConfig) {
 
	options_description opts("Backend options");
 
	opts.add_options()
 
		("registration.needPassword", value<bool>()->default_value(true), "")
 
		("registration.needRegistration", value<bool>()->default_value(false), "")
 
		("registration.extraField", value<std::vector<std::string> >()->multitoken(), "")
 
		("features.receipts", value<bool>()->default_value(false), "")
 
	;
 

	
 
	std::stringstream ifs(backendConfig);
 
	parsed_options parsed = parse_config_file(ifs, opts, true);
 

	
 
	store(parsed, m_backendConfig);
 
	notify(m_backendConfig);
 

	
 
	onBackendConfigUpdated();
 
}
 

	
 
Config *Config::createFromArgs(int argc, char **argv, std::string &error, std::string &host, int &port) {
 
	std::string jid;
 
	std::ostringstream os;
 
	std::string configFile;
 
	boost::program_options::variables_map vm;
 
	boost::program_options::options_description desc("Usage: spectrum <config_file.cfg>\nAllowed options");
 
	desc.add_options()
 
		("help", "help")
 
		("host,h", boost::program_options::value<std::string>(&host)->default_value(""), "Host to connect to")
 
		("port,p", boost::program_options::value<int>(&port)->default_value(10000), "Port to connect to")
 
		("no-daemonize,n", "Do not run spectrum as daemon")
 
		("no-debug,d", "Create coredumps on crash")
 
		("jid,j", boost::program_options::value<std::string>(&jid)->default_value(""), "Specify JID of transport manually")
 
		("config", boost::program_options::value<std::string>(&configFile)->default_value(""), "Config file")
 
		;
 

	
 
	os << desc;
 
	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("help"))
 
		{
 
			error = os.str();
 
			return NULL;
 
		}
 

	
 
		if(vm.count("config") == 0) {
 
			error = os.str();
 
			return NULL;
 
		}
 
	}
 
	catch (std::runtime_error& e)
 
	{
 
		error = os.str();
 
		return NULL;
 
	}
 
	catch (...)
 
	{
 
		error = os.str();
 
		return NULL;
 
	}
 

	
 
	Config *config = new Config(argc, argv);
 
	if (!config->load(configFile)) {
 
		error = "Can't open " + configFile + " configuration file.\n";
 
		delete config;
 
		return NULL;
 
	}
 
	return config;
 
}
 

	
 
}
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 "transport/util.h"
 
#include "transport/discoitemsresponder.h"
 

	
 
#include "utf8.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 <sys/types.h>
 
#include <signal.h>
 
#include "popt.h"
 
#endif
 

	
 
using namespace Transport::Util;
 

	
 
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;
 
		}
 

	
 
		virtual ~NetworkFactory() {}
 

	
 
		// Creates new conversation (NetworkConversation in this case)
 
		Conversation *createConversation(ConversationManager *conversationManager, const std::string &legacyName, bool isMuc) {
 
			NetworkConversation *nc = new NetworkConversation(conversationManager, legacyName, isMuc);
 
			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, buddyInfo.legacyName, buddyInfo.alias, buddyInfo.groups, (BuddyFlag) buddyInfo.flags);
 
			if (!buddy->isValid()) {
 
				delete buddy;
 
				return NULL;
 
			}
 
			if (buddyInfo.subscription == "both") {
 
				buddy->setSubscription(Buddy::Both);
 
			}
 
			else {
 
				buddy->setSubscription(Buddy::Ask);
 
			}
 
			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_(const std::string& exePath, const char *host, const char *port, const char *cmdlineArgs) {
 
	// BACKEND_ID is replaced with unique ID. The ID is increasing for every backend.
 
	std::string finalExePath = boost::replace_all_copy(exePath, "BACKEND_ID", boost::lexical_cast<std::string>(backend_id++));	
 

	
 
#ifdef _WIN32
 
	// Add host and port.
 
	std::ostringstream fullCmdLine;
 
	fullCmdLine << "\"" << finalExePath << "\" --host " << host << " --port " << port;
 

	
 
	if (cmdlineArgs)
 
		fullCmdLine << " " << cmdlineArgs;
 

	
 
	LOG4CXX_INFO(logger, "Starting new backend " << fullCmdLine.str());
 

	
 
	// We must provide a non-const buffer to CreateProcess below
 
	std::vector<wchar_t> rawCommandLineArgs( fullCmdLine.str().size() + 1 );
 
	wcscpy_s(&rawCommandLineArgs[0], rawCommandLineArgs.size(), utf8ToUtf16(fullCmdLine.str()).c_str());
 

	
 
	STARTUPINFO         si;
 
	PROCESS_INFORMATION pi;
 

	
 
	ZeroMemory (&si, sizeof(si));
 
	si.cb=sizeof (si);
 

	
 
	if (! CreateProcess(
 
		utf8ToUtf16(finalExePath).c_str(),
 
		&rawCommandLineArgs[0],
 
		0,                    // process attributes
 
		0,                    // thread attributes
 
		0,                    // inherit handles
 
		0,                    // creation flags
 
		0,                    // environment
 
		0,                    // cwd
 
		&si,
 
		&pi
 
		)
 
	)  {
 
		LOG4CXX_ERROR(logger, "Could not start process");
 
	}
 

	
 
	return 0;
 
#else
 
	// Add host and port.
 
	finalExePath += std::string(" --host ") + host + " --port " + port + " " + cmdlineArgs;
 
	LOG4CXX_INFO(logger, "Starting new backend " << finalExePath);
 

	
 
	// Create array of char * from string using -lpopt library
 
	char *p = (char *) malloc(finalExePath.size() + 1);
 
	strcpy(p, finalExePath.c_str());
 
	int argc;
 
	char **argv;
 
	poptParseArgvString(p, &argc, (const char ***) &argv);
 

	
 
	// fork and exec
 
	pid_t pid = fork();
 
	if ( pid == 0 ) {
 
		setsid();
 
		// close all files
 
		int maxfd=sysconf(_SC_OPEN_MAX);
 
		for(int fd=3; fd<maxfd; fd++) {
 
			close(fd);
 
		}
 
		// 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) {
 
	// Set alias only if it's not empty. Backends are allowed to send empty alias if it has
 
	// not changed.
 
	if (!payload.alias().empty()) {
 
		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++) {
 
		std::string group = payload.group(i);
 
		utf8::replace_invalid(payload.group(i).begin(), payload.group(i).end(), group.begin(), '_');
 
		groups.push_back(group);
 
	}
 
	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, DiscoItemsResponder *discoItemsResponder) {
 
	m_ftManager = ftManager;
 
	m_userManager = userManager;
 
	m_config = config;
 
	m_component = component;
 
	m_isNextLongRun = false;
 
	m_adminInterface = NULL;
 
	m_startingBackend = false;
 
	m_lastLogin = 0;
 
	m_discoItemsResponder = discoItemsResponder;
 
	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();
 

	
 
	m_loginTimer = component->getNetworkFactories()->getTimerFactory()->createTimer(CONFIG_INT(config, "service.login_delay") * 1000);
 
	m_loginTimer->onTick.connect(boost::bind(&NetworkPluginServer::loginDelayFinished, this));
 
	m_loginTimer->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));
 
}
 

	
 
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::start() {
 
	m_server->start();
 

	
 
	LOG4CXX_INFO(logger, "Listening on host " << CONFIG_STRING(m_config, "service.backend_host") << " port " << CONFIG_STRING(m_config, "service.backend_port"));
 

	
 
	while (true) {
 
		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->getCommandLineArgs().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) {
 
					if (status == 254) {
 
						LOG4CXX_ERROR(logger, "Backend can not be started, because it needs database to store data, but the database backend is not configured.");
 
					}
 
					else {
 
						LOG4CXX_ERROR(logger, "Backend can not be started, exit_code=" << WEXITSTATUS(status) << ", possible error: " << strerror(WEXITSTATUS(status)));
 
					}
 
					LOG4CXX_ERROR(logger, "Check backend log for more details");
 
					continue;
 
				}
 
			}
 
			else {
 
				LOG4CXX_ERROR(logger, "Backend can not be started");
 
				continue;
 
			}
 
		}
 

	
 
		signal(SIGCHLD, SigCatcher);
 
#endif
 
		// quit the while loop
 
		break;
 
	}
 
}
 

	
 
void NetworkPluginServer::loginDelayFinished() {
 
	m_loginTimer->stop();
 
	connectWaitingUsers();
 
}
 

	
 
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;
 

	
 
	m_startingBackend = false;
 

	
 
	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;
 
	}
 
	std::string field = payload.fullname();
 

	
 
	boost::shared_ptr<Swift::VCard> vcard(new Swift::VCard());
 

	
 
	utf8::replace_invalid(payload.fullname().begin(), payload.fullname().end(), field.begin(), '_');
 
	vcard->setFullName(field);
 

	
 
	field = payload.nickname();
 

	
 
	utf8::replace_invalid(payload.nickname().begin(), payload.nickname().end(), field.begin(), '_');
 
	vcard->setNickname(field);
 

	
 
	vcard->setPhoto(Swift::createByteArray(payload.photo()));
 

	
 
	m_vcardResponder->sendVCard(payload.id(), vcard);
 
}
 

	
 
void NetworkPluginServer::handleAuthorizationPayload(const std::string &data) {
 
	pbnetwork::Buddy payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	// Create subscribe presence and forward it to XMPP side
 
	Swift::Presence::ref response = Swift::Presence::create();
 
	response->setTo(user->getJID());
 
	std::string name = payload.buddyname();
 

	
 
	if (CONFIG_BOOL_DEFAULTED(m_config, "service.jid_escaping", true)) {
 
		name = Swift::JID::getEscapedNode(name);
 
	}
 
	else {
 
		if (name.find_last_of("@") != std::string::npos) {
 
			name.replace(name.find_last_of("@"), 1, "%");
 
		}
 
	}
 

	
 
	response->setFrom(Swift::JID(name, m_component->getJID().toString()));
 
	response->setType(Swift::Presence::Subscribe);
 
	m_component->getStanzaChannel()->sendPresence(response);
 
}
 

	
 
void NetworkPluginServer::handleChatStatePayload(const std::string &data, Swift::ChatState::ChatStateType type) {
 
	pbnetwork::Buddy payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	// We're not creating new Conversation just because of chatstates.
 
	// Some networks/clients spams with chatstates a lot and it leads to bigger memory usage.
 
	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.buddyname());
 
	if (!conv) {
 
		return;
 
	}
 

	
 
	// Forward chatstate
 
	boost::shared_ptr<Swift::Message> msg(new Swift::Message());
 
	msg->addPayload(boost::make_shared<Swift::ChatState>(type));
 

	
 
	conv->handleMessage(msg);
 
}
 

	
 
void NetworkPluginServer::handleBuddyChangedPayload(const std::string &data) {
 
	pbnetwork::Buddy payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	LocalBuddy *buddy = (LocalBuddy *) user->getRosterManager()->getBuddy(payload.buddyname());
 
	if (buddy) {
 
		handleBuddyPayload(buddy, payload);
 
		buddy->handleBuddyChanged();
 
	}
 
	else {
 
		if (payload.buddyname() == user->getUserInfo().uin) {
 
			return;
 
		}
 

	
 
		std::vector<std::string> groups;
 
		for (int i = 0; i < payload.group_size(); i++) {
 
			groups.push_back(payload.group(i));
 
		}
 
		if (CONFIG_BOOL_DEFAULTED(m_config, "service.jid_escaping", true)) {
 
			buddy = new LocalBuddy(user->getRosterManager(), -1, payload.buddyname(), payload.alias(), groups, BUDDY_JID_ESCAPING);
 
		}
 
		else {
 
			buddy = new LocalBuddy(user->getRosterManager(), -1, payload.buddyname(), payload.alias(), groups, BUDDY_NO_FLAG);
 
		}
 
		if (!buddy->isValid()) {
 
			delete buddy;
 
			return;
 
		}
 

	
 
		buddy->setStatus(Swift::StatusShow((Swift::StatusShow::Type) payload.status()), payload.statusmessage());
 
		buddy->setIconHash(payload.iconhash());
 
		buddy->setBlocked(payload.blocked());
 
		user->getRosterManager()->setBuddy(buddy);
 
	}
 
}
 

	
 
void NetworkPluginServer::handleBuddyRemovedPayload(const std::string &data) {
 
	pbnetwork::Buddy payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	user->getRosterManager()->removeBuddy(payload.buddyname());
 
}
 

	
 
void NetworkPluginServer::handleParticipantChangedPayload(const std::string &data) {
 
	pbnetwork::Participant payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.room());
 
	if (!conv) {
 
		return;
 
	}
 

	
 
	conv->handleParticipantChanged(payload.nickname(), payload.flag(), payload.status(), payload.statusmessage(), payload.newname());
 
}
 

	
 
void NetworkPluginServer::handleRoomChangedPayload(const std::string &data) {
 
	pbnetwork::Room payload;
 
	if (payload.ParseFromString(data) == false) {
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.room());
 
	if (!conv) {
 
		return;
 
	}
 

	
 
	conv->setNickname(payload.nickname());
 
}
 

	
 
void NetworkPluginServer::handleConvMessagePayload(const std::string &data, bool subject) {
 
	pbnetwork::ConversationMessage payload;
 

	
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(payload.username());
 
	if (!user)
 
		return;
 

	
 
	// Message from legacy network triggers network acticity
 
	user->updateLastActivity();
 

	
 
	// Set proper body.
 
	boost::shared_ptr<Swift::Message> msg(new Swift::Message());
 
	if (subject) {
 
		msg->setSubject(payload.message());
 
	}
 
	else {
 
		msg->setBody(payload.message());
 
	}
 

	
 
	if (payload.headline()) {
 
		msg->setType(Swift::Message::Headline);
 
	}
 

	
 
	// Add xhtml-im payload.
 
	if (CONFIG_BOOL(m_config, "service.enable_xhtml") && !payload.xhtml().empty()) {
 
		msg->addPayload(boost::make_shared<Swift::XHTMLIMPayload>(payload.xhtml()));
 
	}
 

	
 
	if (!payload.timestamp().empty()) {
 
		boost::posix_time::ptime timestamp = boost::posix_time::from_iso_string(payload.timestamp());
 
		boost::shared_ptr<Swift::Delay> delay(boost::make_shared<Swift::Delay>());
 
		delay->setStamp(timestamp);
 
		msg->addPayload(delay);
 
	}
 

	
 
	
 
	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.buddyname());
 

	
 
	// We can't create Conversation for payload with nickname, because this means the message is from room,
 
	// but this user is not in any room, so it's OK to just reject this message
 
	if (!conv && !payload.nickname().empty()) {
 
		return;
 
	}
 

	
 
	// Create new Conversation if it does not exist
 
	if (!conv) {
 
		conv = new NetworkConversation(user->getConversationManager(), payload.buddyname());
 
		user->getConversationManager()->addConversation(conv);
 
		conv->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, this, _1, _2));
 
	}
 

	
 
	// Forward it
 
	conv->handleMessage(msg, payload.nickname());
 
	m_userManager->messageToXMPPSent();
 
}
 

	
 
void NetworkPluginServer::handleConvMessageAckPayload(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->addPayload(boost::make_shared<Swift::DeliveryReceipt>(payload.id()));
 

	
 
	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.buddyname());
 

	
 
	// Receipts don't create conversation
 
	if (!conv) {
 
		return;
 
	}
 

	
 
	// Forward it
 
	conv->handleMessage(msg);
 
}
 

	
 
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());
 
		user->getConversationManager()->addConversation(conv);
 
		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;
 

	
 
	if (m_filetransfers.find(payload.ftid()) == m_filetransfers.end()) {
 
		LOG4CXX_ERROR(logger, "Uknown filetransfer with id " << payload.ftid());
 
		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();
 
		}
 

	
 
void NetworkPluginServer::connectWaitingUsers() {
 
	// 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)
 
		if (getFreeClient(true, false, true) == 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);
 
		}
 
	}
 
}
 

	
 
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();
 
		}
 

	
 
		connectWaitingUsers();
 
	}
 

	
 
	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 response;
 
	response.set_config(msg->getBody());
 

	
 
	std::string message;
 
	response.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_QUERY);
 

	
 
	send(b->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleBackendConfigPayload(const std::string &data) {
 
	pbnetwork::BackendConfig payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	m_config->updateBackendConfig(payload.config());
 
}
 

	
 
void NetworkPluginServer::handleRoomListPayload(const std::string &data) {
 
	pbnetwork::RoomList payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	m_discoItemsResponder->clearRooms();
 
	for (int i = 0; i < payload.room_size() && i < payload.name_size(); i++) {
 
		m_discoItemsResponder->addRoom(Swift::JID::getEscapedNode(payload.room(i)) + "@" + m_component->getJID().toString(), payload.name(i));
 
	}
 
}
 

	
 
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);
 

	
 
		// If backend is slow and it is sending us lot of message, there is possibility
 
		// that we don't receive PONG response before timeout. However, if we received
 
		// at least some data, it means backend is not dead and we can treat it as
 
		// PONG received event.
 
		if (c->pongReceived == false) {
 
			c->pongReceived = true;
 
		}
 

	
 
		// 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;
 
			case pbnetwork::WrapperMessage_Type_TYPE_BACKEND_CONFIG:
 
				handleBackendConfigPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_ROOM_LIST:
 
				handleRoomListPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_CONV_MESSAGE_ACK:
 
				handleConvMessageAckPayload(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;
 
		}
 
	}
 

	
 
	// We have to remove startingBackend flag otherwise 1 broken backend start could
 
	// block the backend.
 
	m_startingBackend = false;
 

	
 
	// 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);
 

	
 
#ifndef WIN32
 
			// generate coredump for this backend to find out why it wasn't able to respond to PING
 
			std::string pid = (*it)->id;
 
			if (!pid.empty()) {
 
				try {
 
					kill(boost::lexical_cast<int>(pid), SIGABRT);
 
				}
 
				catch (...) { }
 
			}
 
#endif
 
		}
 

	
 
		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));
 
	}
 
	else {
 
		status.set_status((pbnetwork::StatusType) presence->getShow());
 
	}
 

	
 
	status.set_statusmessage(presence->getStatus());
 

	
 
	std::string message;
 
	status.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_STATUS_CHANGED);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleRoomJoined(User *user, const Swift::JID &who, const std::string &r, const std::string &nickname, const std::string &password) {
 
	UserInfo userInfo = user->getUserInfo();
 

	
 
	pbnetwork::Room room;
 
	room.set_username(user->getJID().toBare());
 
	room.set_nickname(nickname);
 
	room.set_room(r);
 
	room.set_password(password);
 

	
 
	std::string message;
 
	room.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_JOIN_ROOM);
 
 
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleRoomLeft(User *user, const std::string &r) {
 
	UserInfo userInfo = user->getUserInfo();
 

	
 
	pbnetwork::Room room;
 
	room.set_username(user->getJID().toBare());
 
	room.set_nickname("");
 
	room.set_room(r);
 
	room.set_password("");
 

	
 
	std::string message;
 
	room.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LEAVE_ROOM);
 
 
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleUserDestroyed(User *user) {
 
	m_waitingUsers.remove(user);
 
	UserInfo userInfo = user->getUserInfo();
 

	
 
	user->onReadyToConnect.disconnect(boost::bind(&NetworkPluginServer::handleUserReadyToConnect, this, user));
 
	user->onPresenceChanged.disconnect(boost::bind(&NetworkPluginServer::handleUserPresenceChanged, this, user, _1));
 
	user->onRoomJoined.disconnect(boost::bind(&NetworkPluginServer::handleRoomJoined, this, user, _1, _2, _3, _4));
 
	user->onRoomLeft.disconnect(boost::bind(&NetworkPluginServer::handleRoomLeft, this, user, _1));
 

	
 
	pbnetwork::Logout logout;
 
	logout.set_user(user->getJID().toBare());
 
	logout.set_legacyname(userInfo.uin);
 

	
 
	std::string message;
 
	logout.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LOGOUT);
 
 
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
	c->users.remove(user);
 

	
 
	// If backend should handle only one user, it must not accept another one before 
 
	// we kill it, so set up willDie to true
 
	if (c->users.size() == 0 && CONFIG_INT(m_config, "service.users_per_backend") == 1) {
 
		LOG4CXX_INFO(logger, "Backend " << c->id << " will die, because the last user disconnected");
 
		c->willDie = true;
 
	}
 
}
 

	
 
void NetworkPluginServer::handleMessageReceived(NetworkConversation *conv, boost::shared_ptr<Swift::Message> &msg) {
 
	conv->getConversationManager()->getUser()->updateLastActivity();
 
	boost::shared_ptr<Swift::ChatState> statePayload = msg->getPayload<Swift::ChatState>();
 
	if (statePayload) {
 
		pbnetwork::WrapperMessage_Type type = pbnetwork::WrapperMessage_Type_TYPE_BUDDY_CHANGED;
 
		switch (statePayload->getChatState()) {
 
			case Swift::ChatState::Active:
 
				type = pbnetwork::WrapperMessage_Type_TYPE_BUDDY_STOPPED_TYPING;
 
				break;
 
			case Swift::ChatState::Composing:
 
				type = pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPING;
 
				break;
 
			case Swift::ChatState::Paused:
 
				type = pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPED;
 
				break;
 
			default:
 
				break;
 
		}
 
		if (type != pbnetwork::WrapperMessage_Type_TYPE_BUDDY_CHANGED) {
 
			pbnetwork::Buddy buddy;
 
			buddy.set_username(conv->getConversationManager()->getUser()->getJID().toBare());
 
			buddy.set_buddyname(conv->getLegacyName());
 

	
 
			std::string message;
 
			buddy.SerializeToString(&message);
 

	
 
			WRAP(message, type);
 

	
 
			Backend *c = (Backend *) conv->getConversationManager()->getUser()->getData();
 
			if (!c) {
 
				return;
 
			}
 
			send(c->connection, message);
 
		}
 
	}
 

	
 
	boost::shared_ptr<Swift::AttentionPayload> attentionPayload = msg->getPayload<Swift::AttentionPayload>();
 
	if (attentionPayload) {
 
		pbnetwork::ConversationMessage m;
 
		m.set_username(conv->getConversationManager()->getUser()->getJID().toBare());
 
		m.set_buddyname(conv->getLegacyName());
 
		m.set_message(msg->getBody());
 

	
 
		std::string message;
 
		m.SerializeToString(&message);
 

	
 
		WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_ATTENTION);
 

	
 
		Backend *c = (Backend *) conv->getConversationManager()->getUser()->getData();
 
		send(c->connection, message);
 
		return;
 
	}
 

	
 
	if (!msg->getSubject().empty()) {
 
		pbnetwork::ConversationMessage m;
 
		m.set_username(conv->getConversationManager()->getUser()->getJID().toBare());
 
		m.set_buddyname(conv->getLegacyName());
 
		m.set_message(msg->getSubject());
 

	
 
		std::string message;
 
		m.SerializeToString(&message);
 

	
 
		WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_ROOM_SUBJECT_CHANGED);
 

	
 
		Backend *c = (Backend *) conv->getConversationManager()->getUser()->getData();
 
		send(c->connection, message);
 
		return;
 
	}
 
	
 

	
 
	std::string xhtml;
 
	boost::shared_ptr<Swift::XHTMLIMPayload> xhtmlPayload = msg->getPayload<Swift::XHTMLIMPayload>();
 
	if (xhtmlPayload) {
 
		xhtml = xhtmlPayload->getBody();
 
	}
 

	
 
	// Send normal message
 
	if (!msg->getBody().empty() || !xhtml.empty()) {
 
		pbnetwork::ConversationMessage m;
 
		m.set_username(conv->getConversationManager()->getUser()->getJID().toBare());
 
		m.set_buddyname(conv->getLegacyName());
 
		m.set_message(msg->getBody());
 
		m.set_xhtml(xhtml);
 
		boost::shared_ptr<Swift::DeliveryReceiptRequest> receiptPayload = msg->getPayload<Swift::DeliveryReceiptRequest>();
 
		if (receiptPayload && !msg->getID().empty()) {
 
			m.set_id(msg->getID());
 
		}
 

	
 
		std::string message;
 
		m.SerializeToString(&message);
 

	
 
		WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_CONV_MESSAGE);
 

	
 
		Backend *c = (Backend *) conv->getConversationManager()->getUser()->getData();
 
		if (!c) {
 
			return;
 
		}
 
		send(c->connection, message);
 
	}
 
}
 

	
 
void NetworkPluginServer::handleBuddyRemoved(Buddy *b) {
 
	User *user = b->getRosterManager()->getUser();
 

	
 
	pbnetwork::Buddy buddy;
 
	buddy.set_username(user->getJID().toBare());
 
	buddy.set_buddyname(b->getName());
 
	buddy.set_alias(b->getAlias());
 
	BOOST_FOREACH(const std::string &g, b->getGroups()) {
 
		buddy.add_group(g);
 
	}
 
	buddy.set_status(pbnetwork::STATUS_NONE);
 

	
 
	std::string message;
 
	buddy.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_BUDDY_REMOVED);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleBuddyUpdated(Buddy *b, const Swift::RosterItemPayload &item) {
 
	User *user = b->getRosterManager()->getUser();
 

	
 
	dynamic_cast<LocalBuddy *>(b)->setAlias(item.getName());
 
	dynamic_cast<LocalBuddy *>(b)->setGroups(item.getGroups());
 

	
 
	pbnetwork::Buddy buddy;
 
	buddy.set_username(user->getJID().toBare());
 
	buddy.set_buddyname(b->getName());
 
	buddy.set_alias(b->getAlias());
 
	BOOST_FOREACH(const std::string &g, b->getGroups()) {
 
		buddy.add_group(g);
 
	}
 
	buddy.set_status(pbnetwork::STATUS_NONE);
 

	
 
	std::string message;
 
	buddy.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_BUDDY_CHANGED);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleBuddyAdded(Buddy *buddy, const Swift::RosterItemPayload &item) {
 
	handleBuddyUpdated(buddy, item);
 
}
 

	
 
void NetworkPluginServer::handleBlockToggled(Buddy *b) {
 
	User *user = b->getRosterManager()->getUser();
 

	
 
	pbnetwork::Buddy buddy;
 
	buddy.set_username(user->getJID().toBare());
 
	buddy.set_buddyname(b->getName());
 
	buddy.set_alias(b->getAlias());
 
	BOOST_FOREACH(const std::string &g, b->getGroups()) {
 
		buddy.add_group(g);
 
	}
 
	buddy.set_status(pbnetwork::STATUS_NONE);
 
	buddy.set_blocked(!b->isBlocked());
 

	
 
	std::string message;
 
	buddy.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_BUDDY_CHANGED);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 

	
 
void NetworkPluginServer::handleVCardUpdated(User *user, boost::shared_ptr<Swift::VCard> v) {
 
	pbnetwork::VCard vcard;
 
	vcard.set_username(user->getJID().toBare());
 
	vcard.set_buddyname("");
 
	vcard.set_id(0);
 
	vcard.set_photo(&v->getPhoto()[0], v->getPhoto().size());
 
	vcard.set_nickname(v->getNickname());
 

	
 
	std::string message;
 
	vcard.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_VCARD);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleVCardRequired(User *user, const std::string &name, unsigned int id) {
 
	pbnetwork::VCard vcard;
 
	vcard.set_username(user->getJID().toBare());
 
	vcard.set_buddyname(name);
 
	vcard.set_id(id);
 

	
 
	std::string message;
 
	vcard.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_VCARD);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleFTAccepted(User *user, const std::string &buddyName, const std::string &fileName, unsigned long size, unsigned long ftID) {
 
	pbnetwork::File f;
 
	f.set_username(user->getJID().toBare());
 
	f.set_buddyname(buddyName);
 
	f.set_filename(fileName);
 
	f.set_size(size);
 
	f.set_ftid(ftID);
 

	
 
	std::string message;
 
	f.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_FT_START);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleFTRejected(User *user, const std::string &buddyName, const std::string &fileName, unsigned long size) {
 
	pbnetwork::File f;
 
	f.set_username(user->getJID().toBare());
 
	f.set_buddyname(buddyName);
 
	f.set_filename(fileName);
 
	f.set_size(size);
 
	f.set_ftid(0);
 

	
 
	std::string message;
 
	f.SerializeToString(&message);
 

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_FT_FINISH);
 

	
 
	Backend *c = (Backend *) user->getData();
 
	if (!c) {
 
		return;
 
	}
 
	send(c->connection, message);
 
}
 

	
 
void NetworkPluginServer::handleFTStateChanged(Swift::FileTransfer::State state, const std::string &userName, const std::string &buddyName, const std::string &fileName, unsigned long size, unsigned long id) {
 
	User *user = m_userManager->getUser(userName);
 
	if (!user) {
 
		// TODO: FIXME We have to remove filetransfer when use disconnects
 
		return;
 
	}
 
	if (state.state == Swift::FileTransfer::State::Transferring) {
 
		handleFTAccepted(user, buddyName, fileName, size, id);
 
	}
 
	else if (state.state == Swift::FileTransfer::State::Canceled) {
 
		handleFTRejected(user, buddyName, fileName, size);
 
	}
 
}
 

	
 
void NetworkPluginServer::sendPing(Backend *c) {
 

	
 
	std::string message;
 
	pbnetwork::WrapperMessage wrap;
 
	wrap.set_type(pbnetwork::WrapperMessage_Type_TYPE_PING);
 
	wrap.SerializeToString(&message);
 

	
 
	if (c->connection) {
 
		LOG4CXX_INFO(logger, "PING to " << c << " (ID=" << c->id << ")");
 
		send(c->connection, message);
 
		c->pongReceived = false;
 
	}
 
// 	LOG4CXX_INFO(logger, "PING to " << c);
 
}
 

	
 
NetworkPluginServer::Backend *NetworkPluginServer::getFreeClient(bool acceptUsers, bool longRun) {
 
NetworkPluginServer::Backend *NetworkPluginServer::getFreeClient(bool acceptUsers, bool longRun, bool check) {
 
	NetworkPluginServer::Backend *c = NULL;
 

	
 
	unsigned long diff = CONFIG_INT(m_config, "service.login_delay");
 
	time_t now = time(NULL);
 
	if (diff && (now - m_lastLogin < diff)) {
 
		m_loginTimer->start();
 
		return NULL;
 
	}
 

	
 
	if (!check) {
 
		m_lastLogin = time(NULL);
 
	}
 

	
 
	// Check all backends and find free one
 
	for (std::list<Backend *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 
		if ((*it)->willDie == false && (*it)->acceptUsers == acceptUsers && (*it)->users.size() < CONFIG_INT(m_config, "service.users_per_backend") && (*it)->connection && (*it)->longRun == longRun) {
 
			c = *it;
 
			// if we're not reusing all backends and backend is full, stop accepting new users on this backend
 
			if (!CONFIG_BOOL(m_config, "service.reuse_old_backends")) {
 
				if (c->users.size() + 1 >= CONFIG_INT(m_config, "service.users_per_backend")) {
 
					c->acceptUsers = false;
 
				}
 
			}
 
			break;
 
		}
 
	}
 

	
 
	// there's no free backend, so spawn one.
 
	if (c == NULL && !m_startingBackend) {
 
		m_isNextLongRun = longRun;
 
		m_startingBackend = true;
 
		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->getCommandLineArgs().c_str());
 
	}
 

	
 
	return c;
 
}
 

	
 
}
0 comments (0 inline, 0 general)