Changeset - 9ff51c193cc3
[Not reviewed]
0 3 0
Jan Kaluza - 10 years ago 2016-01-05 15:44:18
jkaluza@redhat.com
Twitter: Fix joining room by disabling sending raw xml to backends which do not need it
3 files changed with 10 insertions and 1 deletions:
0 comments (0 inline, 0 general)
backends/twitter/TwitterPlugin.cpp
Show inline comments
 
#include "TwitterPlugin.h"
 
#include "Requests/StatusUpdateRequest.h"
 
#include "Requests/DirectMessageRequest.h"
 
#include "Requests/TimelineRequest.h"
 
#include "Requests/FetchFriends.h"
 
#include "Requests/HelpMessageRequest.h"
 
#include "Requests/PINExchangeProcess.h"
 
#include "Requests/OAuthFlow.h"
 
#include "Requests/CreateFriendRequest.h"
 
#include "Requests/DestroyFriendRequest.h"
 
#include "Requests/RetweetRequest.h"
 
#include "Requests/ProfileImageRequest.h"
 
#include "Swiften/StringCodecs/Hexify.h"
 

	
 
DEFINE_LOGGER(logger, "Twitter Backend");
 

	
 
TwitterPlugin *np = NULL;
 
Swift::SimpleEventLoop *loop_; // Event Loop
 

	
 
const std::string OLD_APP_KEY = "PCWAdQpyyR12ezp2fVwEhw";
 
const std::string OLD_APP_SECRET = "EveLmCXJIg2R7BTCpm6OWV8YyX49nI0pxnYXh7JMvDg";
 

	
 
#define abs(x) ((x)<0?-(x):(x))
 
#define SHA(x) (Swift::Hexify::hexify(Swift::SHA1::getHash(Swift::createByteArray((x)))))
 

	
 
//Compares two +ve intergers 'a' and 'b' represented as strings 
 
static int cmp(std::string a, std::string b)
 
{
 
	int diff = abs((int)a.size() - (int)b.size());
 
	if(a.size() < b.size()) a = std::string(diff,'0') + a;
 
	else b = std::string(diff,'0') + b;
 
	
 
	if(a == b) return 0;
 
	if(a < b) return -1;
 
	return 1;
 
}
 

	
 

	
 
TwitterPlugin::TwitterPlugin(Config *config, Swift::SimpleEventLoop *loop, StorageBackend *storagebackend, const std::string &host, int port) : NetworkPlugin() 
 
{
 
	this->config = config;
 
	this->storagebackend = storagebackend;
 
	this->m_firstPing = true;
 

	
 
	if (CONFIG_HAS_KEY(config, "twitter.consumer_key") == false) {
 
		consumerKey = "5mFePMiJi0KpeURONkelg";
 
	}
 
	else {
 
		consumerKey = CONFIG_STRING(config, "twitter.consumer_key");
 
	}
 
	if (CONFIG_HAS_KEY(config, "twitter.consumer_secret") == false) {
 
		consumerSecret = "YFZCDJwRhbkccXEnaYr1waCQejTJcOY8F7l5Wim3FA";
 
	}
 
	else {
 
		consumerSecret = CONFIG_STRING(config, "twitter.consumer_secret");
 
	}
 

	
 
	if (consumerSecret.empty() || consumerKey.empty()) {
 
		LOG4CXX_ERROR(logger, "Consumer key and Consumer secret can't be empty.");
 
		exit(1);
 
	}
 

	
 
	adminLegacyName = "twitter.com"; 
 
	adminChatRoom = "#twitter"; 
 
	adminNickName = "twitter"; 
 
	adminAlias = "twitter";
 

	
 
	OAUTH_KEY = "twitter_oauth_token";
 
	OAUTH_SECRET = "twitter_oauth_secret";
 
	MODE = "mode";
 

	
 
	m_factories = new Swift::BoostNetworkFactories(loop);
 
	m_conn = m_factories->getConnectionFactory()->createConnection();
 
	m_conn->onDataRead.connect(boost::bind(&TwitterPlugin::_handleDataRead, this, _1));
 
	m_conn->connect(Swift::HostAddressPort(Swift::HostAddress(host), port));
 

	
 
	tp = new ThreadPool(loop_, 10);
 

	
 
	LOG4CXX_INFO(logger, "Fetch timeout is set to " << CONFIG_INT_DEFAULTED(config, "twitter.fetch_timeout", 90000));
 
	tweet_timer = m_factories->getTimerFactory()->createTimer(CONFIG_INT_DEFAULTED(config, "twitter.fetch_timeout", 90000));
 
	message_timer = m_factories->getTimerFactory()->createTimer(CONFIG_INT_DEFAULTED(config, "twitter.fetch_timeout", 90000));
 

	
 
	tweet_timer->onTick.connect(boost::bind(&TwitterPlugin::pollForTweets, this));
 
	message_timer->onTick.connect(boost::bind(&TwitterPlugin::pollForDirectMessages, this));
 

	
 
	tweet_timer->start();
 
	message_timer->start();
 

	
 
#if HAVE_SWIFTEN_3
 
		cryptoProvider = boost::shared_ptr<Swift::CryptoProvider>(Swift::PlatformCryptoProvider::create());
 
#endif
 
	
 
	
 
	LOG4CXX_INFO(logger, "Starting the plugin.");
 
}
 

	
 
TwitterPlugin::~TwitterPlugin() 
 
{
 
	delete storagebackend;
 
	std::set<std::string>::iterator it;
 
	for(it = onlineUsers.begin() ; it != onlineUsers.end() ; it++) delete userdb[*it].sessions;
 
	delete tp;
 
}
 

	
 
// Send data to NetworkPlugin server
 
void TwitterPlugin::sendData(const std::string &string) 
 
{
 
	m_conn->write(Swift::createSafeByteArray(string));
 
}
 

	
 
// Receive date from the NetworkPlugin server and invoke the appropirate payload handler (implement in the NetworkPlugin class)
 
void TwitterPlugin::_handleDataRead(boost::shared_ptr<Swift::SafeByteArray> data) 
 
{
 
	if (m_firstPing) {
 
		m_firstPing = false;
 
		// Users can join the network without registering if we allow
 
		// one user to connect multiple IRC networks.
 
		NetworkPlugin::PluginConfig cfg;
 
		cfg.setNeedPassword(false);
 
		sendConfig(cfg);
 
	}
 

	
 
	std::string d(data->begin(), data->end());
 
	handleDataRead(d);
 
}
 

	
 
// User trying to login into his twitter account
 
void TwitterPlugin::handleLoginRequest(const std::string &user, const std::string &legacyName, const std::string &password) 
 
{
 
	if(userdb.count(user) && (userdb[user].connectionState == NEW || 
 
										userdb[user].connectionState == CONNECTED || 
 
										userdb[user].connectionState == WAITING_FOR_PIN)) {
 
		LOG4CXX_INFO(logger, std::string("A session corresponding to ") + user + std::string(" is already active"))
 
		return;
 
	}
 
	
 
	LOG4CXX_INFO(logger, std::string("Received login request for ") + user)	
 
	initUserSession(user, legacyName, password);
 
	handleConnected(user);
 
	
 
	LOG4CXX_INFO(logger, "SPECTRUM 1 USER? - " << (userdb[user].spectrum1User? "true" : "false")) 
 
	
 
	LOG4CXX_INFO(logger, user << ": Adding Buddy " << adminLegacyName << " " << adminAlias)
 
	handleBuddyChanged(user, adminLegacyName, adminAlias, std::vector<std::string>(), pbnetwork::STATUS_ONLINE);
 
	userdb[user].nickName = "";
 
	
 
	LOG4CXX_INFO(logger, "Querying database for usersettings of " << user)
 
	std::string key, secret;
 
	getUserOAuthKeyAndSecret(user, key, secret);
 

	
 
	if(key == "" || secret == "") {			
 
		LOG4CXX_INFO(logger, "Intiating OAuth Flow for user " << user)
 
		setTwitterMode(user, 0);
 
		tp->runAsThread(new OAuthFlow(np, userdb[user].sessions, user, userdb[user].sessions->getTwitterUsername()));
 
	} else {
 
		LOG4CXX_INFO(logger, user << " is already registerd. Using the stored oauth key and secret")
 
		LOG4CXX_INFO(logger, key << " " << secret)	
 
		pinExchangeComplete(user, key, secret);
 
	}
 
}
 

	
 
// User logging out
 
void TwitterPlugin::handleLogoutRequest(const std::string &user, const std::string &legacyName) 
 
{
 
	if (userdb.count(user)) {
 
		delete userdb[user].sessions;
 
		userdb[user].sessions = NULL;
 
		userdb[user].connectionState = DISCONNECTED;
 
	}
 

	
 
	if(onlineUsers.count(user)) {
 
		onlineUsers.erase(user);
 
	}
 
}
 

	
 
// User joining a Chatroom
 
void TwitterPlugin::handleJoinRoomRequest(const std::string &user, const std::string &room, const std::string &nickname, const std::string &password)
 
{
 
	if(room == adminChatRoom) {	
 
		LOG4CXX_INFO(logger, "Received Join Twitter room request for " << user)
 
		LOG4CXX_INFO(logger, "Received Join Twitter room request for " << user << " '" << nickname << "'")
 

	
 
		setTwitterMode(user, 2);
 
		handleParticipantChanged(user, nickname, room, 0, pbnetwork::STATUS_ONLINE);
 
		handleParticipantChanged(user, adminNickName, room, 0, pbnetwork::STATUS_ONLINE);
 
		userdb[user].nickName = nickname;
 
		handleMessage(user, adminChatRoom, "Connected to Twitter room! Populating your followers list", adminNickName);
 
		tp->runAsThread(new FetchFriends(userdb[user].sessions, user,
 
										 boost::bind(&TwitterPlugin::populateRoster, this, _1, _2, _3, _4)));
 
	} else {
 
		setTwitterMode(user, 0);
 
		LOG4CXX_ERROR(logger, "Couldn't connect to chatroom - " << room <<"! Try twitter-chatroom as the chatroom to access Twitter account")
 
		handleMessage(user, adminLegacyName, "Couldn't connect to chatroom! Try twitter-chatroom as the chatroom to access Twitter account");
 
	}	
 
}
 

	
 
// User leaving a Chatroom
 
void TwitterPlugin::handleLeaveRoomRequest(const std::string &user, const std::string &room)
 
{
 
	if(room == adminChatRoom && onlineUsers.count(user)) {
 
		LOG4CXX_INFO(logger, "Leaving chatroom! Switching back to default mode 0")
 
		setTwitterMode(user, 0);
 
		handleBuddyChanged(user, adminLegacyName, adminAlias, std::vector<std::string>(), pbnetwork::STATUS_ONLINE);
 
	}
 
}
 

	
 
// Messages to be sent to Twitter 
 
void TwitterPlugin::handleMessageSendRequest(const std::string &user, const std::string &legacyName, const std::string &message, const std::string &xhtml, const std::string &/*id*/) 
 
{
 

	
 
	LOG4CXX_INFO(logger, "Received " << user << " --> " << legacyName << " - " << message)
 
	
 
	if(legacyName == adminLegacyName || legacyName == adminChatRoom)  {
 
		std::string cmd = "", data = "";
 
	 
 
		/** Parsing the message - Assuming message format to be <cmd>[ ]*<data>**/	
 
		int i;
 
		for(i=0 ; i<message.size() && message[i] != ' '; i++) cmd += message[i];
 
		while(i<message.size() && message[i] == ' ') i++;
 
		data = message.substr(i);
 
		/***********************************************************************/
 
		
 
		if(cmd == "#pin") 
 
			tp->runAsThread(new PINExchangeProcess(np, userdb[user].sessions, user, data));
 
		else if(cmd == "#help") 
 
			tp->runAsThread(new HelpMessageRequest(user, CONFIG_STRING(config, "service.jid"), boost::bind(&TwitterPlugin::helpMessageResponse, this, _1, _2)));
 
		else if(cmd[0] == '@') {
 
			std::string username = cmd.substr(1); 
 
			tp->runAsThread(new DirectMessageRequest(userdb[user].sessions, user, username, data,
 
												     boost::bind(&TwitterPlugin::directMessageResponse, this, _1, _2, _3, _4)));
 
		}
 
		else if(cmd == "#status") 
 
			tp->runAsThread(new StatusUpdateRequest(userdb[user].sessions, user, data,
 
														boost::bind(&TwitterPlugin::statusUpdateResponse, this, _1, _2)));
 
		else if(cmd == "#timeline") 
 
			tp->runAsThread(new TimelineRequest(userdb[user].sessions, user, data, "",
 
														boost::bind(&TwitterPlugin::displayTweets, this, _1, _2, _3, _4)));
 
		else if(cmd == "#friends") 
 
			tp->runAsThread(new FetchFriends(userdb[user].sessions, user,
 
													   boost::bind(&TwitterPlugin::displayFriendlist, this, _1, _2, _3, _4)));
 
		else if(cmd == "#follow") 
 
			tp->runAsThread(new CreateFriendRequest(userdb[user].sessions, user, data.substr(0,data.find('@')),
 
													   boost::bind(&TwitterPlugin::createFriendResponse, this, _1, _2, _3, _4)));
 
		else if(cmd == "#unfollow") 
 
			tp->runAsThread(new DestroyFriendRequest(userdb[user].sessions, user, data.substr(0,data.find('@')),
 
													   boost::bind(&TwitterPlugin::deleteFriendResponse, this, _1, _2, _3)));
 
		else if(cmd == "#retweet") 
 
			tp->runAsThread(new RetweetRequest(userdb[user].sessions, user, data,
 
													   boost::bind(&TwitterPlugin::RetweetResponse, this, _1, _2)));
 
		else if(cmd == "#mode") {
 
			int m = 0;
 
			m = atoi(data.c_str());
 
			mode prevm = userdb[user].twitterMode;
 

	
 
			if((mode)m == userdb[user].twitterMode) return; //If same as current mode return
 
			if(m < 0 || m > 1) { // Invalid modes
 
				handleMessage(user, adminLegacyName, std::string("Error! Unknown mode ") + data + ". Allowed values 0 or 1." );
 
				return;
 
			}
 

	
 
			setTwitterMode(user, m);
 
			if((userdb[user].twitterMode == SINGLECONTACT || userdb[user].twitterMode == CHATROOM) && prevm == MULTIPLECONTACT) clearRoster(user);
 
			else if(userdb[user].twitterMode == MULTIPLECONTACT) 
 
				tp->runAsThread(new FetchFriends(userdb[user].sessions, user, boost::bind(&TwitterPlugin::populateRoster, this, _1, _2, _3, _4)));
 

	
 
			handleMessage(user, userdb[user].twitterMode == CHATROOM ? adminChatRoom : adminLegacyName,
 
								std::string("Changed mode to ") + data, userdb[user].twitterMode == CHATROOM ? adminNickName : "");
 

	
 
			LOG4CXX_INFO(logger, user << ": Changed mode to " << data  << " <" << (userdb[user].twitterMode == CHATROOM ? adminNickName : "") << ">" )
 
		}
 

	
 
		else if(userdb[user].twitterMode == CHATROOM) {
 
			std::string buddy = message.substr(0, message.find(":"));
 
			if(userdb[user].buddies.count(buddy) == 0) {
 
				tp->runAsThread(new StatusUpdateRequest(userdb[user].sessions, user, message,
 
														boost::bind(&TwitterPlugin::statusUpdateResponse, this, _1, _2)));
 
			} else {
 
				data = message.substr(message.find(":")+1); //Can parse better??:P
 
				tp->runAsThread(new DirectMessageRequest(userdb[user].sessions, user, buddy, data,
 
												 		 boost::bind(&TwitterPlugin::directMessageResponse, this, _1, _2, _3, _4)));
 
			}
 
		}
 
		else handleMessage(user, userdb[user].twitterMode == CHATROOM ? adminChatRoom : adminLegacyName,
 
				                 "Unknown command! Type #help for a list of available commands.", userdb[user].twitterMode == CHATROOM ? adminNickName : "");
 
	} 
 

	
 
	else {	
 
		std::string buddy = legacyName;
 
		if(userdb[user].twitterMode == CHATROOM) buddy = legacyName.substr(legacyName.find("/") + 1);
 
		if(legacyName != "twitter") {
 
			tp->runAsThread(new DirectMessageRequest(userdb[user].sessions, user, buddy, message,
 
												 boost::bind(&TwitterPlugin::directMessageResponse, this, _1, _2, _3, _4)));
 
		}
 
	}
 
}
 

	
 
void TwitterPlugin::handleBuddyUpdatedRequest(const std::string &user, const std::string &buddyName, const std::string &alias, const std::vector<std::string> &groups) 
 
{
 
	if(userdb[user].connectionState != CONNECTED) {
 
		LOG4CXX_ERROR(logger, user << " is not connected to twitter!")
 
		return;
 
	}
 

	
 
	LOG4CXX_INFO(logger, user << " - Adding Twitter contact " << buddyName)
 
	tp->runAsThread(new CreateFriendRequest(userdb[user].sessions, user, buddyName, 
 
											boost::bind(&TwitterPlugin::createFriendResponse, this, _1, _2, _3, _4)));
 
}
 

	
 
void TwitterPlugin::handleBuddyRemovedRequest(const std::string &user, const std::string &buddyName, const std::vector<std::string> &groups) 
 
{
 
	if(userdb[user].connectionState != CONNECTED) {
 
		LOG4CXX_ERROR(logger, user << " is not connected to twitter!")
 
		return;
 
	}
 

	
 
	if (getTwitterMode(user) == MULTIPLECONTACT) {
 
		LOG4CXX_ERROR(logger, user << " not removing Twitter contact " << buddyName << ", because the mode is not MULTIPLECONTACT")
 
		return;
 
	}
 
	
 
	LOG4CXX_INFO(logger, user << " - Removing Twitter contact " << buddyName)
 
	tp->runAsThread(new DestroyFriendRequest(userdb[user].sessions, user, buddyName, 
 
											 boost::bind(&TwitterPlugin::deleteFriendResponse, this, _1, _2, _3)));
 
}
 

	
 
void TwitterPlugin::handleVCardRequest(const std::string &user, const std::string &legacyName, unsigned int id)
 
{
 
	if(userdb[user].connectionState != CONNECTED) {
 
		LOG4CXX_ERROR(logger, user << " is not connected to twitter!")
 
		return;
 
	}
 
	
 
	LOG4CXX_INFO(logger, user << " - VCardRequest for " << legacyName << ", " << userdb[user].buddiesInfo[legacyName].getProfileImgURL())
 

	
 
	if(getTwitterMode(user) != SINGLECONTACT && userdb[user].buddies.count(legacyName) 
 
		&& userdb[user].buddiesInfo[legacyName].getProfileImgURL().length()) {
 
		if(userdb[user].buddiesImgs.count(legacyName) == 0) {
 
			tp->runAsThread(new ProfileImageRequest(config, user, legacyName, userdb[user].buddiesInfo[legacyName].getProfileImgURL(), id,
 
													boost::bind(&TwitterPlugin::profileImageResponse, this, _1, _2, _3, _4, _5)));
 
		}
 
		handleVCard(user, id, legacyName, legacyName, "", userdb[user].buddiesImgs[legacyName]);
 
	}
 
}
 

	
 
void TwitterPlugin::pollForTweets()
 
{
 
	boost::mutex::scoped_lock lock(userlock);
 
	std::set<std::string>::iterator it = onlineUsers.begin();
 
	while(it != onlineUsers.end()) {
 
		std::string user = *it;
 
		tp->runAsThread(new TimelineRequest(userdb[user].sessions, user, "", getMostRecentTweetIDUnsafe(user),
 
											boost::bind(&TwitterPlugin::displayTweets, this, _1, _2, _3, _4)));
 
		it++;
 
	}
 
	tweet_timer->start();
 
}
 

	
 
void TwitterPlugin::pollForDirectMessages()
 
{
 
	boost::mutex::scoped_lock lock(userlock);
 
	std::set<std::string>::iterator it = onlineUsers.begin();
 
	while(it != onlineUsers.end()) {
 
		std::string user = *it;
 
		tp->runAsThread(new DirectMessageRequest(userdb[user].sessions, user, "", getMostRecentDMIDUnsafe(user),
 
											boost::bind(&TwitterPlugin::directMessageResponse, this, _1, _2, _3, _4)));
 
		it++;
 
	}
 
	message_timer->start();
 
}
 

	
 

	
 
bool TwitterPlugin::getUserOAuthKeyAndSecret(const std::string user, std::string &key, std::string &secret) 
 
{
 
	boost::mutex::scoped_lock lock(dblock);
 
	
 
	UserInfo info;
 
	if(storagebackend->getUser(user, info) == false) {
 
		LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
		return false;
 
	}
 

	
 
	key="", secret=""; int type = TYPE_STRING;;
 
	storagebackend->getUserSetting((long)info.id, OAUTH_KEY, type, key);
 
	storagebackend->getUserSetting((long)info.id, OAUTH_SECRET, type, secret);
 
	return true;
 
}
 

	
 
bool TwitterPlugin::checkSpectrum1User(const std::string user) 
 
{
 
	boost::mutex::scoped_lock lock(dblock);
 
	
 
	UserInfo info;
 
	if(storagebackend->getUser(user, info) == false) {
 
		LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
		return false;
 
	}
 

	
 
	std::string first_synchronization_done = "";
 
	int type = TYPE_STRING;
 
	storagebackend->getUserSetting((long)info.id, "first_synchronization_done", type, first_synchronization_done);
 

	
 
	LOG4CXX_INFO(logger, "first_synchronization_done: " << first_synchronization_done)
 

	
 
	if(first_synchronization_done.length()) return true;
 
	return false;
 
}
 

	
 
int TwitterPlugin::getTwitterMode(const std::string user) 
 
{
 
	boost::mutex::scoped_lock lock(dblock);
 
	
 
	UserInfo info;
 
	if(storagebackend->getUser(user, info) == false) {
 
		LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
		return -1;
 
	}
 

	
 
	int type = TYPE_STRING; int m;
 
	std::string s_m;
 
	storagebackend->getUserSetting((long)info.id, MODE, type, s_m);
 
	if(s_m == "") {
 
		s_m  = "0";
 
		storagebackend->updateUserSetting((long)info.id, MODE, s_m);
 
	}
 
	m = atoi(s_m.c_str());
 
	return m;
 
}
 

	
 
bool TwitterPlugin::setTwitterMode(const std::string user, int m) 
 
{
 
	boost::mutex::scoped_lock lock(dblock);
 
	
 
	UserInfo info;
 
	if(storagebackend->getUser(user, info) == false) {
 
		LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
		return false;
 
	}
 

	
 
	if(m < 0 || m > 2) {
 
		LOG4CXX_ERROR(logger, "Unknown mode " << m <<". Using default mode 0")
 
		m = 0;
 
	}
 

	
 
	userdb[user].twitterMode = (mode)m;
 

	
 
	//int type;
 
	std::string s_m = std::string(1,m+'0');
 
	LOG4CXX_INFO(logger, "Storing mode " << m <<" for user " << user)
 
	storagebackend->updateUserSetting((long)info.id, MODE, s_m);
 
	return true;
 
}
 

	
 
bool TwitterPlugin::storeUserOAuthKeyAndSecret(const std::string user, const std::string OAuthKey, const std::string OAuthSecret) 
 
{
 

	
 
	boost::mutex::scoped_lock lock(dblock);
 

	
 
	UserInfo info;
 
	if(storagebackend->getUser(user, info) == false) {
 
		LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
		return false;
 
	}
 

	
 
	storagebackend->updateUserSetting((long)info.id, OAUTH_KEY, OAuthKey);	
 
	storagebackend->updateUserSetting((long)info.id, OAUTH_SECRET, OAuthSecret);
 
	return true;
 
}
 

	
 
void TwitterPlugin::initUserSession(const std::string user, const std::string legacyName, const std::string password)
 
{
 
	boost::mutex::scoped_lock lock(userlock);
 

	
 
	std::string username = legacyName;
 
	std::string passwd = password;
 
	LOG4CXX_INFO(logger, username + "  " + passwd)
 

	
 
	userdb[user].sessions = new twitCurl();	
 
	if(CONFIG_HAS_KEY(config,"proxy.server")) {			
 
		std::string ip = CONFIG_STRING(config,"proxy.server");
 

	
 
		std::ostringstream out; 
 
		out << CONFIG_INT(config,"proxy.port");
 
		std::string port = out.str();
 

	
 
		std::string puser = CONFIG_STRING(config,"proxy.user");
 
		std::string ppasswd = CONFIG_STRING(config,"proxy.password");
 

	
 
		LOG4CXX_INFO(logger, ip << " " << port << " " << puser << " " << ppasswd)
 
		
 
		if(ip != "localhost" && port != "0") {
 
			userdb[user].sessions->setProxyServerIp(ip);
 
			userdb[user].sessions->setProxyServerPort(port);
 
			userdb[user].sessions->setProxyUserName(puser);
 
			userdb[user].sessions->setProxyPassword(ppasswd);
 
		}
 
	}
 

	
 
	//Check if the user is spectrum1 user
 
	userdb[user].spectrum1User = checkSpectrum1User(user);
 

	
 
	userdb[user].connectionState = NEW;
 
	userdb[user].legacyName = username;	
 
	userdb[user].sessions->setTwitterUsername(username);
 
	userdb[user].sessions->setTwitterPassword(passwd);
 

	
 
	if(!userdb[user].spectrum1User) {
 
		userdb[user].sessions->getOAuth().setConsumerKey(consumerKey);
 
		userdb[user].sessions->getOAuth().setConsumerSecret(consumerSecret);
 
	} else {
 
		userdb[user].sessions->getOAuth().setConsumerKey(OLD_APP_KEY);
 
		userdb[user].sessions->getOAuth().setConsumerSecret(OLD_APP_SECRET);
 
	}
 
}
 

	
 
void TwitterPlugin::OAuthFlowComplete(const std::string user, twitCurl *obj) 
 
{
 
	boost::mutex::scoped_lock lock(userlock);	
 

	
 
	delete userdb[user].sessions;
 
	userdb[user].sessions = obj->clone();	
 
	userdb[user].connectionState = WAITING_FOR_PIN;
 
}	
 

	
 
void TwitterPlugin::pinExchangeComplete(const std::string user, const std::string OAuthAccessTokenKey, const std::string OAuthAccessTokenSecret) 
 
{
 
	boost::mutex::scoped_lock lock(userlock);	
 
		
 
	userdb[user].sessions->getOAuth().setOAuthTokenKey( OAuthAccessTokenKey );
 
	userdb[user].sessions->getOAuth().setOAuthTokenSecret( OAuthAccessTokenSecret );
 
	userdb[user].connectionState = CONNECTED;
 
	userdb[user].twitterMode = (mode)getTwitterMode(user);
 
	
 
	if(userdb[user].twitterMode == MULTIPLECONTACT) {
 
		tp->runAsThread(new FetchFriends(userdb[user].sessions, user, boost::bind(&TwitterPlugin::populateRoster, this, _1, _2, _3, _4)));
 
	}
 

	
 
	onlineUsers.insert(user);
 
	userdb[user].mostRecentTweetID = "";
 
	userdb[user].mostRecentDirectMessageID = "";
 
}	
 

	
 
void TwitterPlugin::updateLastTweetID(const std::string user, const std::string ID)
 
{
 
	boost::mutex::scoped_lock lock(userlock);	
 
	userdb[user].mostRecentTweetID = ID;
 

	
 
	UserInfo info;
 
	if(storagebackend->getUser(user, info) == false) {
 
		LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
		return;
 
	}
 

	
 
	storagebackend->updateUserSetting((long)info.id, "twitter_last_tweet", ID);
 
}
 

	
 
std::string TwitterPlugin::getMostRecentTweetIDUnsafe(const std::string user)
 
{	
 
	std::string ID = "";
 
	if(onlineUsers.count(user)) {
 
		ID = userdb[user].mostRecentTweetID;
 
		if (ID.empty()) {
 
			int type = TYPE_STRING;
 
			UserInfo info;
 
			if(storagebackend->getUser(user, info) == false) {
 
				LOG4CXX_ERROR(logger, "Didn't find entry for " << user << " in the database!")
 
			}
 
			else {
 
				storagebackend->getUserSetting(info.id, "twitter_last_tweet", type, ID);
libtransport/NetworkPluginServer.cpp
Show inline comments
 
@@ -722,768 +722,772 @@ void NetworkPluginServer::handleConvMessagePayload(const std::string &data, bool
 

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

	
 
	if (payload.id().empty()) {
 
		LOG4CXX_WARN(logger, "Received message ack with empty ID, not forwarding to XMPP.");
 
		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;
 
#if !HAVE_SWIFTEN_3
 
	transfer.ft->onStateChange.connect(boost::bind(&NetworkPluginServer::handleFTStateChanged, this, _1, payload.username(), payload.buddyname(), payload.filename(), payload.size(), bytestream_id));
 
	transfer.ft->start();
 
#endif
 
}
 

	
 
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::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(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_component->getFrontend()->clearRoomList();
 
	for (int i = 0; i < payload.room_size() && i < payload.name_size(); i++) {
 
		m_component->getFrontend()->addRoomToRoomList(Swift::JID::getEscapedNode(payload.room(i)) + "@" + m_component->getJID().toString(), payload.name(i));
 
	}
 
}
 
#if HAVE_SWIFTEN_3
 
void NetworkPluginServer::handleElement(boost::shared_ptr<Swift::ToplevelElement> element) {
 
#else
 
void NetworkPluginServer::handleElement(boost::shared_ptr<Swift::Element> element) {
 
#endif
 
	boost::shared_ptr<Swift::Stanza> stanza = boost::dynamic_pointer_cast<Swift::Stanza>(element);
 
	if (!stanza) {
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(stanza->getTo().toBare());
 
	if (!user)
 
		return;
 

	
 
	Swift::JID originalJID = stanza->getFrom();
 
	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(originalJID.toBare());
 

	
 
	LocalBuddy *buddy = (LocalBuddy *) user->getRosterManager()->getBuddy(stanza->getFrom().toBare());
 
	if (buddy) {
 
		const Swift::JID &jid = buddy->getJID();
 
		if (stanza->getFrom().getResource().empty()) {
 
			stanza->setFrom(Swift::JID(jid.getNode(), jid.getDomain()));
 
		}
 
		else {
 
			stanza->setFrom(Swift::JID(jid.getNode(), jid.getDomain(), stanza->getFrom().getResource()));
 
		}
 
	}
 
	else {
 
		std::string name = stanza->getFrom().toBare();
 
		if (conv && conv->isMUC()) {
 
			if (name.find_last_of("@") != std::string::npos) {
 
				name.replace(name.find_last_of("@"), 1, "%");
 
			}
 
		}
 
		else {
 
			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, "%");
 
				}
 
			}
 
		}
 
		if (stanza->getFrom().getResource().empty()) {
 
			stanza->setFrom(Swift::JID(name, m_component->getJID().toString()));
 
		}
 
		else {
 
			stanza->setFrom(Swift::JID(name, m_component->getJID().toString(), stanza->getFrom().getResource()));
 
		}
 
	}
 

	
 
	boost::shared_ptr<Swift::Message> message = boost::dynamic_pointer_cast<Swift::Message>(stanza);
 
	if (message) {
 
		if (conv) {
 
			conv->handleRawMessage(message);
 
			return;
 
		}
 

	
 
		m_component->getFrontend()->sendMessage(message);
 
		return;
 
	}
 

	
 
	boost::shared_ptr<Swift::Presence> presence = boost::dynamic_pointer_cast<Swift::Presence>(stanza);
 
	if (presence) {
 
		if (buddy) {
 
			if (!buddy->isAvailable() && presence->getType() != Swift::Presence::Unavailable) {
 
				buddy->m_status.setType(Swift::StatusShow::Online);
 
			}
 
			buddy->handleRawPresence(presence);
 
		}
 
		else if (conv) {
 
			conv->handleRawPresence(presence);
 
		}
 
		else {
 
			m_component->getFrontend()->sendPresence(presence);
 
		}
 

	
 
		return;
 
	}
 

	
 
	// TODO: FIX TO MAKE RAW XML BACKENDS WORKING AGAIN.
 
// 	boost::shared_ptr<Swift::IQ> iq = boost::dynamic_pointer_cast<Swift::IQ>(stanza);
 
// 	if (iq) {
 
// 		if (m_id2resource.find(stanza->getTo().toBare().toString() + stanza->getID()) != m_id2resource.end()) {
 
// 			iq->setTo(Swift::JID(iq->getTo().getNode(), iq->getTo().getDomain(), m_id2resource[stanza->getTo().toBare().toString() + stanza->getID()]));
 
// 			m_id2resource.erase(stanza->getTo().toBare().toString() + stanza->getID());
 
// 		}
 
// 		else {
 
// 			Swift::Presence::ref highest = m_component->getPresenceOracle()->getHighestPriorityPresence(user->getJID());
 
// 			if (highest) {
 
// 			    iq->setTo(highest->getFrom());
 
// 			} else {
 
// 			    iq->setTo(user->getJID());
 
// 			}
 
// 		}
 
// 		m_component->getFrontend()->sendIQ(iq);
 
// 		return;
 
// 	}
 
}
 

	
 
void NetworkPluginServer::handleRawXML(const std::string &xml) {
 
	m_xmppParser->parse(xml);
 
}
 

	
 
void NetworkPluginServer::handleRawPresenceReceived(boost::shared_ptr<Swift::Presence> presence) {
 
	if (!CONFIG_BOOL_DEFAULTED(m_config, "features.rawxml", false)) {
 
		return;
 
	}
 

	
 
	User *user = m_userManager->getUser(presence->getFrom().toBare());
 
	if (!user)
 
		return;
 

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

	
 
	Swift::JID legacyname = Swift::JID(Buddy::JIDToLegacyName(presence->getTo()));
 
	if (!presence->getTo().getResource().empty()) {
 
		presence->setTo(Swift::JID(legacyname.getNode(), legacyname.getDomain(), presence->getTo().getResource()));
 
	}
 
	else {
 
		presence->setTo(Swift::JID(legacyname.getNode(), legacyname.getDomain()));
 
	}
 

	
 
	std::string xml = safeByteArrayToString(m_serializer->serializeElement(presence));
 
	WRAP(xml, pbnetwork::WrapperMessage_Type_TYPE_RAW_XML);
 
	send(c->connection, xml);
 
}
 

	
 
void NetworkPluginServer::handleRawIQReceived(boost::shared_ptr<Swift::IQ> iq) {
 
	User *user = m_userManager->getUser(iq->getFrom().toBare());
 
	if (!user)
 
		return;
 

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

	
 
	if (iq->getType() == Swift::IQ::Get) {
 
		m_id2resource[iq->getFrom().toBare().toString() + iq->getID()] = iq->getFrom().getResource();
 
	}
 

	
 
	Swift::JID legacyname = Swift::JID(Buddy::JIDToLegacyName(iq->getTo()));
 
	if (!iq->getTo().getResource().empty()) {
 
		iq->setTo(Swift::JID(legacyname.getNode(), legacyname.getDomain(), iq->getTo().getResource()));
 
	}
 
	else {
 
		iq->setTo(Swift::JID(legacyname.getNode(), legacyname.getDomain()));
 
	}
 

	
 
	std::string xml = safeByteArrayToString(m_serializer->serializeElement(iq));
 
	WRAP(xml, pbnetwork::WrapperMessage_Type_TYPE_RAW_XML);
 
	send(c->connection, xml);
 
}
 

	
 
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;
 
			case pbnetwork::WrapperMessage_Type_TYPE_RAW_XML:
 
				handleRawXML(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->onRawPresenceReceived.connect(boost::bind(&NetworkPluginServer::handleRawPresenceReceived, this, _1));
 
	user->onRoomJoined.connect(boost::bind(&NetworkPluginServer::handleRoomJoined, this, user, _1, _2, _3, _4));
 
	user->onRoomLeft.connect(boost::bind(&NetworkPluginServer::handleRoomLeft, this, user, _1));
 

	
 
	user->getRosterManager()->onBuddyAdded.connect(boost::bind(&NetworkPluginServer::handleUserBuddyAdded, this, user, _1));
 
	user->getRosterManager()->onBuddyRemoved.connect(boost::bind(&NetworkPluginServer::handleUserBuddyRemoved, 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);
 

	
 
	// Send buddies
 
	if (CONFIG_BOOL_DEFAULTED(m_config, "features.send_buddies_on_login", false)) {
 
		pbnetwork::Buddies buddies;
 

	
 
		const RosterManager::BuddiesMap &roster = user->getRosterManager()->getBuddies();
 
		for(RosterManager::BuddiesMap::const_iterator bt = roster.begin(); bt != roster.end(); bt++) {
 
			Buddy *b = (*bt).second;
 
			if (!b) {
 
				continue;
 
			}
 

	
 
			pbnetwork::Buddy *buddy = buddies.add_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 msg;
 
		buddies.SerializeToString(&msg);
 

	
 
		WRAP(msg, pbnetwork::WrapperMessage_Type_TYPE_BUDDIES);
tests/twitter/twitter_test.py
Show inline comments
 
import optparse
 
import sys
 
import time
 
import subprocess
 
import os
 

	
 
import sleekxmpp
 
from sleekxmpp.xmlstream.matcher import StanzaPath, MatchXPath
 
from sleekxmpp.xmlstream.handler import Callback
 

	
 

	
 
class Responder(sleekxmpp.ClientXMPP):
 
	def __init__(self, jid, password, room, room_password, nick):
 
		sleekxmpp.ClientXMPP.__init__(self, jid, password)
 
		self.room = room
 
		self.room_password = room_password
 
		self.nick = nick
 
		self.finished = False
 

	
 
		self.tests = {}
 

	
 

	
 
class Client(sleekxmpp.ClientXMPP):
 
	def __init__(self, jid, password, room, nick):
 
		sleekxmpp.ClientXMPP.__init__(self, jid, password)
 
		self.room = room
 
		self.nick = nick
 
		self.add_event_handler("session_start", self.start)
 
		self.add_event_handler("message", self.message)
 
		self.add_event_handler('roster_update', self.roster_update)
 
		self.finished = False
 

	
 
		self.tests = {}
 
		self.tests["timeline_poll"] = ["Timeline received automatically", False]
 
		self.tests["help"] = ["#help command", False]
 
		self.tests["status1"] = ["#status command response", False]
 
		self.tests["status2"] = ["#status command pushes status", False]
 
		self.tests["follow"] = ["#follow command", False]
 
		self.tests["friends"] = ["#friends command", False]
 
		self.tests["unfollow"] = ["#unfollow command", False]
 
		self.tests["friends2"] = ["#friends after unfollow command", False]
 
		self.tests["mode1"] = ["#mode 1 response", False]
 
		self.tests["mode1buddy"] = ["#mode 1 buddy added", False]
 
		self.tests["mode2"] = ["#mode 2 join room", False]
 

	
 
		self.status = "timeline"
 
		self.timestamp = int(time.time())
 

	
 
	def message(self, msg):
 
		if self.status == "timeline" and msg['body'].find("spectrum2tests") != -1 and msg['body'].find("MsgId") != -1:
 
			self.tests["timeline_poll"][1] = True
 
			self.send_message(mto=msg['from'], mbody="#help")
 
			self.status = "help"
 
		elif self.status == "help" and msg['body'].find("You will receive tweets of people you follow") != -1:
 
			self.tests["help"][1] = True
 
			self.send_message(mto=msg['from'], mbody="#status My testing status " + str(self.timestamp))
 
			self.status = "status"
 
		elif self.status == "status" and msg['body'].find("Status Update successful") != -1:
 
			self.tests["status1"][1] = True
 
		elif self.status == "status" and msg['body'].find("spectrum2tests") != -1 and msg['body'].find("MsgId") != -1:
 
			self.tests["status2"][1] = True
 
			self.status = "follow"
 
			self.send_message(mto=msg['from'], mbody="#follow colinpwheeler")
 
		elif self.status == "follow" and msg['body'] == "You are now following colinpwheeler":
 
			self.status = "friends"
 
			self.tests["follow"][1] = True
 
			self.send_message(mto=msg['from'], mbody="#friends")
 
		elif self.status == "friends" and msg['body'].find("USER LIST") != -1 and msg['body'].find("colinpwheeler") != -1:
 
			self.status = "mode1"
 
			self.tests["friends"][1] = True
 
			self.send_message(mto=msg['from'], mbody="#mode 1")
 
		elif self.status == "mode1" and msg['body'] == "Changed mode to 1":
 
			self.tests["mode1"][1] = True
 
		elif self.status == "unfollow" and msg['body'] == "You are not following colinpwheeler anymore":
 
			self.status = "friends2"
 
			self.tests["unfollow"][1] = True
 
			self.send_message(mto=msg['from'], mbody="#friends")
 
		elif self.status == "friends2" and msg['body'].find("USER LIST") != -1 and msg['body'].find("colinpwheeler") == -1:
 
			self.status = "mode1"
 
			self.tests["friends2"][1] = True
 
			self.finished = True
 

	
 
	def roster_update(self, iq):
 
		roster = iq['roster']['items']
 

	
 
		if self.status == "mode1" and roster.has_key("colinpwheeler@localhost"):
 
			self.plugin['xep_0045'].joinMUC("#twitter@localhost", "client", wait=True)
 
			self.tests["mode2"][1] = True
 
			self.plugin['xep_0045'].leaveMUC("#twitter@localhost", "client")
 
			self.status = "unfollow"
 
			self.tests["mode1buddy"][1] = True
 
			self.send_message(mto="twitter.com@localhost", mbody="#unfollow colinpwheeler")
 

	
 
	def start(self, event):
 
		self.getRoster()
 
		self.sendPresence()
 

	
0 comments (0 inline, 0 general)