Changeset - 2b768b45acff
[Not reviewed]
0 1 0
Jan Kaluza - 10 years ago 2016-02-05 08:42:41
jkaluza@redhat.com
Libtransport: Use \40 instead of % in room list
1 file changed with 1 insertions and 5 deletions:
0 comments (0 inline, 0 general)
libtransport/NetworkPluginServer.cpp
Show inline comments
 
@@ -615,773 +615,769 @@ void NetworkPluginServer::handleBuddyRemovedPayload(const std::string &data) {
 

	
 
	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(), (Conversation::ParticipantFlag) 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) {
 
		LOG4CXX_ERROR(logger, "handleConvMessagePayload: unknown username " << payload.username());
 
		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()) {
 
		LOG4CXX_WARN(logger, "handleConvMessagePayload: No conversation with name " << payload.buddyname());
 
		return;
 
	}
 

	
 
	if (conv && payload.pm()) {
 
		conv = (NetworkConversation *) user->getConversationManager()->getConversation(payload.buddyname() + "/" + payload.nickname());
 
		if (!conv) {
 
			conv = new NetworkConversation(user->getConversationManager(), payload.nickname());
 
			std::string name = payload.buddyname();
 
			conv->setRoom(name);
 
			conv->setNickname(payload.buddyname() + "/" + payload.nickname());
 

	
 
			user->getConversationManager()->addConversation(conv);
 
			conv->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, this, _1, _2));
 
		}
 
	}
 

	
 
	// 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_firstPong) {
 
			// first backend connected, start the server, we're ready.
 
			m_component->start();
 
			m_firstPong = false;
 
		}
 

	
 
		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++) {
 
		std::string legacyName = payload.room(i);
 
		if (legacyName.find_last_of("@") != std::string::npos) {
 
			legacyName.replace(legacyName.find_last_of("@"), 1, "%"); // OK
 
		}
 
		m_component->getFrontend()->addRoomToRoomList(Swift::JID::getEscapedNode(legacyName) + "@" + m_component->getJID().toString(), payload.name(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");
0 comments (0 inline, 0 general)