libtransport.git Changeset - 98687145b02a

Changeset - 98687145b02a

Parent rev.

Child rev.

[Not reviewed]

0 5 0

Jan Kaluza - 14 years ago 2011-05-26 08:53:05
hanzz.k@gmail.com

first more or less working version of multiple-backend support

5 files changed with 92 insertions and 47 deletions:

include/transport/networkplugin.h

include/transport/networkpluginserver.h

include/transport/user.h

src/networkplugin.cpp

src/networkpluginserver.cpp

0 comments (0 inline, 0 general)

include/transport/networkplugin.h

➞

Show inline comments

@@ @@ -71,18 +71,20 @@ class NetworkPlugin { @@
 		void handleDataRead(const Swift::ByteArray&);
 		void handleConnected(bool error);
 		void handleDisconnected();
 		void send(const std::string &data);
 		void sendPong();
 		void pingTimeout();
 		std::string m_data;
 		std::string m_host;
 		int m_port;
 		Swift::BoostNetworkFactories *m_factories;
 		Swift::BoostIOServiceThread m_boostIOServiceThread;
 		boost::shared_ptr<Swift::Connection> m_conn;
 		Swift::Timer::ref m_reconnectTimer;
 		Swift::Timer::ref m_pingTimer;
 		bool m_pingReceived;
 };
+}

include/transport/networkpluginserver.h

➞

Show inline comments

@@ @@ -37,12 +37,18 @@ class Buddy; @@
 class LocalBuddy;
 class Config;
 class NetworkConversation;
 class NetworkPluginServer {
 	public:
 		struct Client {
 			bool pongReceived;
 			std::list<User *> users;
 			std::string data;
 		};
 		NetworkPluginServer(Component *component, Config *config, UserManager *userManager);
 		virtual ~NetworkPluginServer();
 		void handleMessageReceived(NetworkConversation *conv, boost::shared_ptr<Swift::Message> &message);
@@ @@ -64,19 +70,17 @@ class NetworkPluginServer { @@
 		void handleUserReadyToConnect(User *user);
 		void handleUserDestroyed(User *user);
 		void send(boost::shared_ptr<Swift::Connection> &, const std::string &data);
 		void pingTimeout();
 		void sendPing();
 		void sendPing(boost::shared_ptr<Swift::Connection> c);
 		boost::shared_ptr<Swift::Connection> getFreeClient();
 		std::string m_command;
 		std::string m_data;
 		UserManager *m_userManager;
 		Config *m_config;
 		boost::shared_ptr<Swift::ConnectionServer> m_server;
 		boost::shared_ptr<Swift::Connection> m_client;
 		bool m_pongReceived;
 		std::map<boost::shared_ptr<Swift::Connection>, Client >  m_clients;
 		Swift::Timer::ref m_pingTimer;
 };
+}

include/transport/user.h

➞

Show inline comments

@@ @@ -93,9 +93,11 @@ class User { @@
 		Swift::PresenceOracle *m_presenceOracle;
 		UserInfo m_userInfo;
 		void *m_data;
 		bool m_connected;
 		bool m_readyForConnect;
 		Swift::Timer::ref m_reconnectTimer;
 		boost::shared_ptr<Swift::Connection> connection;
 		friend class NetworkPluginServer;
 };
+}

src/networkplugin.cpp

➞

Show inline comments

@@ @@ -38,19 +38,20 @@ namespace Transport { @@
 	wrap.SerializeToString(&MESSAGE);
 NetworkPlugin::NetworkPlugin(Swift::EventLoop *loop, const std::string &host, int port) {
 	m_factories = new Swift::BoostNetworkFactories(loop);
 	m_host = host;
 	m_port = port;
 	m_pingReceived = false;
 	m_conn = m_factories->getConnectionFactory()->createConnection();
 	m_conn->onDataRead.connect(boost::bind(&NetworkPlugin::handleDataRead, this, _1));
 	m_conn->onConnectFinished.connect(boost::bind(&NetworkPlugin::handleConnected, this, _1));
 	m_conn->onDisconnected.connect(boost::bind(&NetworkPlugin::handleDisconnected, this));
 	m_reconnectTimer = m_factories->getTimerFactory()->createTimer(1000);
 	m_reconnectTimer->onTick.connect(boost::bind(&NetworkPlugin::connect, this));
 	m_pingTimer = m_factories->getTimerFactory()->createTimer(30000);
 	m_pingTimer->onTick.connect(boost::bind(&NetworkPlugin::pingTimeout, this));
 	connect();
+}
 NetworkPlugin::~NetworkPlugin() {
 	delete m_factories;
+}
 	send(message);
+}
 void NetworkPlugin::handleConnected(bool error) {
 	if (error) {
 		std::cout << "Connecting error\n";
 // 		m_reconnectTimer->start();
 		std::cerr << "Connecting error\n";
 		m_pingTimer->stop();
 		exit(1);
+	}
 	else {
 		std::cout << "Connected\n";
-		m_reconnectTimer->stop();
+		m_pingTimer->start();
+	}
+}
 void NetworkPlugin::handleDisconnected() {
 	std::cout << "Disconnected\n";
 // 	m_reconnectTimer->start();
 	std::cerr << "Disconnected\n";
 	m_pingTimer->stop();
 	exit(1);
+}
 void NetworkPlugin::connect() {
 	std::cout << "Trying to connect the server\n";
 	m_conn->connect(Swift::HostAddressPort(Swift::HostAddress(m_host), m_port));
 	m_reconnectTimer->stop();
+}
 void NetworkPlugin::handleLoginPayload(const std::string &data) {
 	pbnetwork::Login payload;
 	if (payload.ParseFromString(data) == false) {
 		// TODO: ERROR
@@ @@ -285,16 +285,24 @@ void NetworkPlugin::send(const std::string &data) { @@
+	}
 	m_conn->write(Swift::ByteArray(header + data));
+}
 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);
 // 	std::cout << "SENDING PONG\n";
+}
 void NetworkPlugin::pingTimeout() {
 	if (m_pingReceived == false) {
 		exit(1);
+	}
 	m_pingReceived = false;
+}
+}

src/networkpluginserver.cpp

➞

Show inline comments

+}
 NetworkPluginServer::NetworkPluginServer(Component *component, Config *config, UserManager *userManager) {
 	m_userManager = userManager;
 	m_config = config;
 	component->m_factory = new NetworkFactory(this);
 	m_pongReceived = false;
 	m_userManager->onUserCreated.connect(boost::bind(&NetworkPluginServer::handleUserCreated, this, _1));
 	m_userManager->onUserDestroyed.connect(boost::bind(&NetworkPluginServer::handleUserDestroyed, this, _1));
 	m_pingTimer = component->getFactories()->getTimerFactory()->createTimer(10000);
 	m_pingTimer->onTick.connect(boost::bind(&NetworkPluginServer::pingTimeout, this));
 NetworkPluginServer::~NetworkPluginServer() {
 	m_pingTimer->stop();
+}
 void NetworkPluginServer::handleNewClientConnection(boost::shared_ptr<Swift::Connection> c) {
 	if (m_client) {
 		c->disconnect();
+	}
 	m_client = c;
 	m_pongReceived = false;
 	Client client;
 	client.pongReceived = true;
 	m_clients[c] = client;
 	c->onDisconnected.connect(boost::bind(&NetworkPluginServer::handleSessionFinished, this, c));
 	c->onDataRead.connect(boost::bind(&NetworkPluginServer::handleDataRead, this, c, _1));
 	sendPing();
+	sendPing(c);
 	m_pingTimer->start();
+}
 void NetworkPluginServer::handleSessionFinished(boost::shared_ptr<Swift::Connection> c) {
 	if (c == m_client) {
 		m_client.reset();
 	for (std::list<User *>::const_iterator it = m_clients[c].users.begin(); it != m_clients[c].users.end(); it++) {
 		(*it)->handleDisconnected("Internal Server Error, please reconnect.");
+	}
 	m_clients.erase(c);
 	// Execute new session only if there's no free one after this crash/disconnection
 	for (std::map<boost::shared_ptr<Swift::Connection>, Client>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 		if ((*it).second.users.size() < 1) {
 			return;
+		}
+	}
 	m_pingTimer->stop();
 	exec_(CONFIG_STRING(m_config, "service.backend").c_str(), "localhost", "10000", m_config->getConfigFile().c_str());
+}
 void NetworkPluginServer::handleConnectedPayload(const std::string &data) {
 	pbnetwork::Connected payload;
 	if (payload.ParseFromString(data) == false) {
 	conv->handleMessage(msg, payload.nickname());
+}
 void NetworkPluginServer::handleDataRead(boost::shared_ptr<Swift::Connection> c, const Swift::ByteArray &data) {
 	long expected_size = 0;
-	m_data += data.toString();
+	m_clients[c].data += data.toString();
 // 	std::cout << "received data; size = " << m_data.size() << "\n";
 	while (m_data.size() != 0) {
 		if (m_data.size() >= 4) {
 			unsigned char * head = (unsigned char*) m_data.c_str();
 	while (m_clients[c].data.size() != 0) {
 		if (m_clients[c].data.size() >= 4) {
 			unsigned char * head = (unsigned char*) m_clients[c].data.c_str();
 			expected_size = (((((*head << 8) | *(head + 1)) << 8) | *(head + 2)) << 8) | *(head + 3);
 			//expected_size = m_data[0];
 // 			std::cout << "expected_size=" << expected_size << "\n";
-			if (m_data.size() - 4 < expected_size)
+			if (m_clients[c].data.size() - 4 < expected_size)
 				return;
+		}
 		else {
 			return;
+		}
 		std::string msg = m_data.substr(4, expected_size);
 		m_data.erase(0, 4 + expected_size);
 		std::string msg = m_clients[c].data.substr(4, expected_size);
 		m_clients[c].data.erase(0, 4 + expected_size);
 		pbnetwork::WrapperMessage wrapper;
 		if (wrapper.ParseFromString(msg) == false) {
 			// TODO: ERROR
 			return;
+		}
 				handleConvMessagePayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_ROOM_SUBJECT_CHANGED:
 				handleConvMessagePayload(wrapper.payload(), true);
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_PONG:
-				m_pongReceived = true;
+				m_clients[c].pongReceived = true;
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_PARTICIPANT_CHANGED:
 				handleParticipantChangedPayload(wrapper.payload());
 				break;
 			case pbnetwork::WrapperMessage_Type_TYPE_ROOM_NICKNAME_CHANGED:
 				handleRoomChangedPayload(wrapper.payload());
 	c->write(Swift::ByteArray(header + data));
+}
 void NetworkPluginServer::pingTimeout() {
 	std::cout << "pingtimeout\n";
 	if (m_pongReceived) {
 		sendPing();
 		m_pingTimer->start();
+	}
 	else {
 		exec_(CONFIG_STRING(m_config, "service.backend").c_str(), "localhost", "10000", m_config->getConfigFile().c_str());
 	for (std::map<boost::shared_ptr<Swift::Connection>, Client>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 		if ((*it).second.pongReceived) {
 			sendPing((*it).first);
 			m_pingTimer->start();
+		}
 		else {
 			exec_(CONFIG_STRING(m_config, "service.backend").c_str(), "localhost", "10000", m_config->getConfigFile().c_str());
+		}
+	}
+}
 void NetworkPluginServer::handleUserCreated(User *user) {
 // 	UserInfo userInfo = user->getUserInfo();
 	user->onReadyToConnect.connect(boost::bind(&NetworkPluginServer::handleUserReadyToConnect, this, user));
 	login.set_password(userInfo.password);
 	std::string message;
 	login.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LOGIN);
 	user->connection = getFreeClient();
 	m_clients[user->connection].users.push_back(user);
-	send(m_client, message);
+	send(user->connection, message);
+}
 void NetworkPluginServer::handleRoomJoined(User *user, const std::string &r, const std::string &nickname, const std::string &password) {
 	UserInfo userInfo = user->getUserInfo();
 	pbnetwork::Room room;
 	std::string message;
 	room.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_JOIN_ROOM);
-	send(m_client, message);
+	send(user->connection, message);
 	NetworkConversation *conv = new NetworkConversation(user->getConversationManager(), r, true);
 	conv->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, this, _1, _2));
 	conv->setNickname(nickname);
+}
 	std::string message;
 	room.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LEAVE_ROOM);
-	send(m_client, message);
+	send(user->connection, message);
 	NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(r);
 	if (!conv) {
 		return;
+	}
 	std::string message;
 	logout.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LOGOUT);
 	send(m_client, message);
 	send(user->connection, message);
 	m_clients[user->connection].users.remove(user);
 	if (m_clients[user->connection].users.size() == 0) {
 		std::cout << "DISCONNECTING\n";
 		user->connection->disconnect();
 		user->connection.reset();
 		m_clients.erase(user->connection);
+	}
+}
 void NetworkPluginServer::handleMessageReceived(NetworkConversation *conv, boost::shared_ptr<Swift::Message> &msg) {
 	pbnetwork::ConversationMessage m;
 	m.set_username(conv->getConversationManager()->getUser()->getJID().toBare());
 	m.set_buddyname(conv->getLegacyName());
 	std::string message;
 	m.SerializeToString(&message);
 	WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_CONV_MESSAGE);
-	send(m_client, message);
+	send(conv->getConversationManager()->getUser()->connection, message);
+}
 void NetworkPluginServer::sendPing() {
+void NetworkPluginServer::sendPing(boost::shared_ptr<Swift::Connection> c) {
 	std::string message;
 	pbnetwork::WrapperMessage wrap;
 	wrap.set_type(pbnetwork::WrapperMessage_Type_TYPE_PING);
 	wrap.SerializeToString(&message);
 	send(m_client, message);
 	m_pongReceived = false;
 	send(c, message);
 	m_clients[c].pongReceived = false;
 	std::cout << "SENDING PING\n";
+}
 boost::shared_ptr<Swift::Connection> NetworkPluginServer::getFreeClient() {
 	for (std::map<boost::shared_ptr<Swift::Connection>, Client>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
 		if ((*it).second.users.size() < 1) {
 			if ((*it).second.users.size() + 1 == 1) {
 				exec_(CONFIG_STRING(m_config, "service.backend").c_str(), "localhost", "10000", m_config->getConfigFile().c_str());
+			}
 			return (*it).first;
+		}
+	}
 	return boost::shared_ptr<Swift::Connection>();
+}
+}

0 comments (0 inline, 0 general)