Changeset - 899d6fe53fdb
[Not reviewed]
0 2 0
Jan Kaluza - 10 years ago 2016-02-11 14:42:17
jkaluza@redhat.com
Libtransport: Debug lost PING to backend
2 files changed with 8 insertions and 0 deletions:
0 comments (0 inline, 0 general)
libtransport/NetworkPluginServer.cpp
Show inline comments
 
@@ -1116,429 +1116,433 @@ void NetworkPluginServer::handleElement(boost::shared_ptr<Swift::Element> elemen
 
		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() {
 
	LOG4CXX_INFO(logger, "Sending PING to backends");
 
	// 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, "Tried to send PING to backend without pongReceived= " << (*it)->pongReceived << ": (ID=" << (*it)->id << ")");
 
			}
 
		}
 
		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());
 
			buddy->set_iconhash(b->getIconHash());
 
			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);
 
		send(c->connection, msg);
 
	}
 
}
 

	
 
void NetworkPluginServer::handleUserPresenceChanged(User *user, Swift::Presence::ref presence) {
 
	if (presence->getShow() == Swift::StatusShow::None)
 
		return;
 

	
 
	handleRawPresenceReceived(presence);
 

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

	
 
@@ -1777,327 +1781,330 @@ void NetworkPluginServer::handleBuddyUpdated(Buddy *b, const Swift::RosterItemPa
 

	
 
	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::handleUserBuddyAdded(User *user, Buddy *b) {
 
	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::handleUserBuddyRemoved(User *user, Buddy *b) {
 
	handleBuddyRemoved(b);
 
}
 

	
 
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) {
 
	if (!v) {
 
		LOG4CXX_INFO(logger, user->getJID().toString() << ": Received empty VCard");
 
		return;
 
	}
 

	
 
	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 !HAVE_SWIFTEN_3
 
	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);
 
	}
 
#endif
 
}
 

	
 
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;
 
	}
 
	else {
 
		LOG4CXX_WARN(logger, "Tried to send PING to backend without connection: " << c << " (ID=" << c->id << ")");
 
	}
 
// 	LOG4CXX_INFO(logger, "PING to " << c);
 
}
 

	
 
void NetworkPluginServer::sendAPIVersion(Backend *c) {
 

	
 
	pbnetwork::APIVersion apiver;
 
	apiver.set_version(NETWORK_PLUGIN_API_VERSION);
 

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

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_API_VERSION);
 

	
 
	if (c->connection) {
 
		LOG4CXX_INFO(logger, "API Version to " << c << " (ID=" << c->id << ")");
 
		send(c->connection, message);
 
	}
 
}
 

	
 
void NetworkPluginServer::handlePIDTerminated(unsigned long pid) {
 
	std::vector<unsigned long>::iterator log_id_it;
 
	log_id_it = std::find(m_pids.begin(), m_pids.end(), pid);
 
	if (log_id_it != m_pids.end()) {
 
		*log_id_it = 0;
 
	}
 
}
 

	
 
#ifndef _WIN32
 

	
 
static int sig_block_count = 0;
 
static sigset_t block_mask;
 

	
 
static void __block_signals ( void )
 
{
 
  static int init_done = 0;
 

	
 
  if ( (sig_block_count++) != 1 ) return;
 

	
 
  if ( init_done == 0 ) {
 
    sigemptyset ( &block_mask );
 
    sigaddset ( &block_mask, SIGPIPE );
 
    sigaddset ( &block_mask, SIGHUP );
 
    sigaddset ( &block_mask, SIGINT );
 
    sigaddset ( &block_mask, SIGQUIT );
 
    sigaddset ( &block_mask, SIGTERM );
 
    sigaddset ( &block_mask, SIGABRT );
 
    sigaddset ( &block_mask, SIGCHLD );
 
    init_done = 1;
 
  }
 

	
 
  sigprocmask ( SIG_BLOCK, &block_mask, NULL );
 
  return;
 
}
 

	
 
static void __unblock_signals ( void )
 
{
 
  sigset_t sigset;
 

	
 
  if ( (sig_block_count--) != 0 ) return;
 
  sigprocmask ( SIG_UNBLOCK, &block_mask, NULL );
 

	
 
  if ( sigpending ( &sigset ) == 0 ) {
 
    if ( sigismember ( &sigset, SIGCHLD ) ) {
 
      raise ( SIGCHLD );
 
    }
 
  }
 
}
 

	
 
#endif
 

	
 
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->stop();
 
		m_loginTimer = m_component->getNetworkFactories()->getTimerFactory()->createTimer((diff - (now - m_lastLogin)) * 1000);
 
		m_loginTimer->onTick.connect(boost::bind(&NetworkPluginServer::loginDelayFinished, this));
 
		m_loginTimer->start();
 
		LOG4CXX_INFO(logger, "Postponing login because of service.login_delay setting");
 
		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 (!check && 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;
 

	
 
#ifndef _WIN32
 
		__block_signals();
 
#endif
 
		std::vector<unsigned long>::iterator log_id_it;
 
		log_id_it = std::find(m_pids.begin(), m_pids.end(), 0);
 
		std::string log_id = "";
 
		if (log_id_it == m_pids.end()) {
 
			log_id = boost::lexical_cast<std::string>(m_pids.size() + 1);
 
		}
 
		else {
 
			log_id = boost::lexical_cast<std::string>(log_id_it - m_pids.begin() + 1);
 
		}
 
		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(), log_id.c_str(), m_config->getCommandLineArgs().c_str());
 
		if (log_id_it == m_pids.end()) {
 
			m_pids.push_back(pid);
 
		}
 
		else {
 
			*log_id_it = pid;
 
		}
 
#ifndef _WIN32
 
		__unblock_signals();
 
#endif
 
	}
 

	
 
	return c;
 
}
 

	
 
}
plugin/cpp/networkplugin.cpp
Show inline comments
 
@@ -487,235 +487,236 @@ void NetworkPlugin::handleFTContinuePayload(const std::string &data) {
 

	
 
void NetworkPlugin::handleJoinRoomPayload(const std::string &data) {
 
	pbnetwork::Room payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	handleJoinRoomRequest(payload.username(), payload.room(), payload.nickname(), payload.password());
 
}
 

	
 
void NetworkPlugin::handleLeaveRoomPayload(const std::string &data) {
 
	pbnetwork::Room payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	handleLeaveRoomRequest(payload.username(), payload.room());
 
}
 

	
 
void NetworkPlugin::handleVCardPayload(const std::string &data) {
 
	pbnetwork::VCard payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 

	
 
	if (payload.has_photo()) {
 
		handleVCardUpdatedRequest(payload.username(), payload.photo(), payload.nickname());
 
	}
 
	else if (!payload.buddyname().empty()) {
 
		handleVCardRequest(payload.username(), payload.buddyname(), payload.id());
 
	}
 
}
 

	
 
void NetworkPlugin::handleBuddyChangedPayload(const std::string &data) {
 
	pbnetwork::Buddy payload;
 
	if (payload.ParseFromString(data) == false) {
 
		// TODO: ERROR
 
		return;
 
	}
 
	if (payload.has_blocked()) {
 
		handleBuddyBlockToggled(payload.username(), payload.buddyname(), payload.blocked());
 
	}
 
	else {
 
		std::vector<std::string> groups;
 
		for (int i = 0; i < payload.group_size(); i++) {
 
			groups.push_back(payload.group(i));
 
		}
 
		handleBuddyUpdatedRequest(payload.username(), payload.buddyname(), payload.alias(), groups);
 
	}
 
}
 

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

	
 
	std::vector<std::string> groups;
 
	for (int i = 0; i < payload.group_size(); i++) {
 
		groups.push_back(payload.group(i));
 
	}
 

	
 
	handleBuddyRemovedRequest(payload.username(), payload.buddyname(), groups);
 
}
 

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

	
 
	switch(type) {
 
		case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPING:
 
			handleTypingRequest(payload.username(), payload.buddyname());
 
			break;
 
		case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPED:
 
			handleTypedRequest(payload.username(), payload.buddyname());
 
			break;
 
		case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_STOPPED_TYPING:
 
			handleStoppedTypingRequest(payload.username(), payload.buddyname());
 
			break;
 
		default:
 
			break;
 
	}
 
}
 

	
 
void NetworkPlugin::handleDataRead(std::string &data) {
 
	m_data.insert(m_data.end(), data.begin(), data.end());
 

	
 
	while (m_data.size() != 0) {
 
		unsigned int expected_size;
 

	
 
		if (m_data.size() >= 4) {
 
			expected_size = *((unsigned int*) &m_data[0]);
 
			expected_size = ntohl(expected_size);
 
			if (m_data.size() - 4 < expected_size)
 
				return;
 
		}
 
		else {
 
			return;
 
		}
 

	
 
		pbnetwork::WrapperMessage wrapper;
 
		if (wrapper.ParseFromArray(&m_data[4], expected_size) == false) {
 
			m_data.erase(m_data.begin(), m_data.begin() + 4 + expected_size);
 
			return;
 
		}
 
		m_data.erase(m_data.begin(), m_data.begin() + 4 + expected_size);
 

	
 
		switch(wrapper.type()) {
 
			case pbnetwork::WrapperMessage_Type_TYPE_LOGIN:
 
				handleLoginPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_LOGOUT:
 
				handleLogoutPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_PING:
 
				sendPong();
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_CONV_MESSAGE:
 
				handleConvMessagePayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_ROOM_SUBJECT_CHANGED:
 
				handleRoomSubjectChangedPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_JOIN_ROOM:
 
				handleJoinRoomPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_LEAVE_ROOM:
 
				handleLeaveRoomPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_VCARD:
 
				handleVCardPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_CHANGED:
 
				handleBuddyChangedPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_REMOVED:
 
				handleBuddyRemovedPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_STATUS_CHANGED:
 
				handleStatusChangedPayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPING:
 
				handleChatStatePayload(wrapper.payload(), pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPING);
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPED:
 
				handleChatStatePayload(wrapper.payload(), pbnetwork::WrapperMessage_Type_TYPE_BUDDY_TYPED);
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_STOPPED_TYPING:
 
				handleChatStatePayload(wrapper.payload(), pbnetwork::WrapperMessage_Type_TYPE_BUDDY_STOPPED_TYPING);
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_ATTENTION:
 
				handleAttentionPayload(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_PAUSE:
 
				handleFTPausePayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_FT_CONTINUE:
 
				handleFTContinuePayload(wrapper.payload());
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_EXIT:
 
				handleExitRequest();
 
				break;
 
			case pbnetwork::WrapperMessage_Type_TYPE_RAW_XML:
 
				handleRawXML(wrapper.payload());
 
				break;
 
			default:
 
				return;
 
		}
 
	}
 
}
 

	
 
void NetworkPlugin::send(const std::string &data) {
 
	uint32_t size = htonl(data.size());
 
	char *header = (char *) &size;
 
	sendData(std::string(header, 4) + data);
 
}
 

	
 
void NetworkPlugin::checkPing() {
 
	if (m_pingReceived == false) {
 
		LOG4CXX_ERROR(logger, "PING request not received - exiting...");
 
		handleExitRequest();
 
	}
 
	m_pingReceived = false;
 
}
 

	
 
void NetworkPlugin::sendPong() {
 
	m_pingReceived = true;
 
	std::string message;
 
	pbnetwork::WrapperMessage wrap;
 
	wrap.set_type(pbnetwork::WrapperMessage_Type_TYPE_PONG);
 
	wrap.SerializeToString(&message);
 

	
 
	send(message);
 
	sendMemoryUsage();
 
}
 

	
 
void NetworkPlugin::sendMemoryUsage() {
 
	pbnetwork::Stats stats;
 

	
 
	stats.set_init_res(m_init_res);
 
	double res = 0;
 
	double shared = 0;
 
#ifndef WIN32
 
	process_mem_usage(shared, res);
 
#endif
 

	
 
	double e_res;
 
	double e_shared;
 
	handleMemoryUsage(e_res, e_shared);
 

	
 
	stats.set_res(res + e_res);
 
	stats.set_shared(shared + e_shared);
 
	stats.set_id(stringOf(getpid()));
 

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

	
 
	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_STATS);
 

	
 
	send(message);
 
}
 

	
 
}
0 comments (0 inline, 0 general)