stv0g/spectrum2
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge branch 'master' of github.com:hanzz/libtransport

Browse source
This commit is contained in:
HanzZ 2011-05-28 15:34:47 +02:00
commit 9f934c7a1f
7 changed files with 118 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
backends/libpurple/main.cpp
View file

@ -74,6 +74,7 @@ class SpectrumNetworkPlugin : public NetworkPlugin {
purple_accounts_add(account);
}
m_sessions[user] = account;
purple_account_set_password(account, password.c_str());
purple_account_set_enabled(account, "spectrum", TRUE);
@ -88,6 +89,7 @@ class SpectrumNetworkPlugin : public NetworkPlugin {
const char *protocol = CONFIG_STRING(config, "service.protocol").c_str();
PurpleAccount *account = purple_accounts_find(legacyName.c_str(), protocol);
if (account) {
m_sessions[user] = NULL;
purple_account_set_enabled(account, "spectrum", FALSE);
// Remove conversations.
@ -109,8 +111,7 @@ class SpectrumNetworkPlugin : public NetworkPlugin {
void handleMessageSendRequest(const std::string &user, const std::string &legacyName, const std::string &message) {
const char *protocol = CONFIG_STRING(config, "service.protocol").c_str();
PurpleAccount *account = purple_accounts_find(user.c_str(), protocol);
std::cout << user << "\n";
PurpleAccount *account = m_sessions[user];
if (account) {
PurpleConversation *conv = purple_find_conversation_with_account(PURPLE_CONV_TYPE_IM, legacyName.c_str(), account);
if (!conv) {
@ -122,6 +123,7 @@ class SpectrumNetworkPlugin : public NetworkPlugin {
}
}
std::map<std::string, PurpleAccount *> m_sessions;
std::map<PurpleAccount *, std::string> m_accounts;
private:
Config *config;

4
include/transport/networkplugin.h
View file

@ -74,6 +74,7 @@ class NetworkPlugin {
void send(const std::string &data);
void sendPong();
void pingTimeout();
std::string m_data;
std::string m_host;
@ -81,7 +82,8 @@ class NetworkPlugin {
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;
};

19
include/transport/networkpluginserver.h
View file

@ -40,6 +40,13 @@ class NetworkConversation;
class NetworkPluginServer {
public:
struct Client {
bool pongReceived;
std::list<User *> users;
std::string data;
boost::shared_ptr<Swift::Connection> connection;
};
NetworkPluginServer(Component *component, Config *config, UserManager *userManager);
virtual ~NetworkPluginServer();
@ -48,8 +55,8 @@ class NetworkPluginServer {
private:
void handleNewClientConnection(boost::shared_ptr<Swift::Connection> c);
void handleSessionFinished(boost::shared_ptr<Swift::Connection>);
void handleDataRead(boost::shared_ptr<Swift::Connection>, const Swift::ByteArray&);
void handleSessionFinished(Client *c);
void handleDataRead(Client *c, const Swift::ByteArray&);
void handleConnectedPayload(const std::string &payload);
void handleDisconnectedPayload(const std::string &payload);
@ -67,15 +74,13 @@ class NetworkPluginServer {
void send(boost::shared_ptr<Swift::Connection> &, const std::string &data);
void pingTimeout();
void sendPing();
void sendPing(Client *c);
Client *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::list<Client *> m_clients;
Swift::Timer::ref m_pingTimer;
};

1
include/transport/user.h
View file

@ -96,6 +96,7 @@ class User {
bool m_connected;
bool m_readyForConnect;
Swift::Timer::ref m_reconnectTimer;
boost::shared_ptr<Swift::Connection> connection;
};
}

4
spectrum/src/sample.cfg
View file

@ -4,8 +4,8 @@ password = secret
server = 127.0.0.1
port = 5222
server_mode = 1
#backend=../../backends/libpurple/libpurple_backend
backend=../../backends/libircclient-qt/libircclient-qt_backend
backend=../../backends/libpurple/libpurple_backend
#backend=../../backends/libircclient-qt/libircclient-qt_backend
protocol=prpl-jabber
[database]

26
src/networkplugin.cpp
View file

@ -41,13 +41,14 @@ NetworkPlugin::NetworkPlugin(Swift::EventLoop *loop, const std::string &host, in
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();
}
@ -156,26 +157,25 @@ void NetworkPlugin::handleRoomChanged(const std::string &user, const std::string
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) {
@ -288,6 +288,7 @@ void NetworkPlugin::send(const std::string &data) {
}
void NetworkPlugin::sendPong() {
m_pingReceived = true;
std::string message;
pbnetwork::WrapperMessage wrap;
wrap.set_type(pbnetwork::WrapperMessage_Type_TYPE_PONG);
@ -297,4 +298,13 @@ void NetworkPlugin::sendPong() {
// std::cout << "SENDING PONG\n";
}
void NetworkPlugin::pingTimeout() {
std::cout << "PINGTIMEOUT " << m_pingReceived << " " << this << "\n";
if (m_pingReceived == false) {
exit(1);
}
m_pingReceived = false;
m_pingTimer->start();
}
}

115
src/networkpluginserver.cpp
View file

@ -106,7 +106,6 @@ NetworkPluginServer::NetworkPluginServer(Component *component, Config *config, U
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));
@ -127,24 +126,33 @@ NetworkPluginServer::~NetworkPluginServer() {
}
void NetworkPluginServer::handleNewClientConnection(boost::shared_ptr<Swift::Connection> c) {
if (m_client) {
c->disconnect();
}
m_client = c;
m_pongReceived = false;
Client *client = new Client;
client->pongReceived = true;
client->connection = c;
c->onDisconnected.connect(boost::bind(&NetworkPluginServer::handleSessionFinished, this, c));
c->onDataRead.connect(boost::bind(&NetworkPluginServer::handleDataRead, this, c, _1));
sendPing();
m_clients.push_back(client);
c->onDisconnected.connect(boost::bind(&NetworkPluginServer::handleSessionFinished, this, client));
c->onDataRead.connect(boost::bind(&NetworkPluginServer::handleDataRead, this, client, _1));
sendPing(client);
m_pingTimer->start();
}
void NetworkPluginServer::handleSessionFinished(boost::shared_ptr<Swift::Connection> c) {
if (c == m_client) {
m_client.reset();
void NetworkPluginServer::handleSessionFinished(Client *c) {
for (std::list<User *>::const_iterator it = c->users.begin(); it != c->users.end(); it++) {
(*it)->setData(NULL);
(*it)->handleDisconnected("Internal Server Error, please reconnect.");
}
m_clients.remove(c);
delete c;
// Execute new session only if there's no free one after this crash/disconnection
for (std::list<Client *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
if ((*it)->users.size() < 1) {
return;
}
}
m_pingTimer->stop();
exec_(CONFIG_STRING(m_config, "service.backend").c_str(), "localhost", "10000", m_config->getConfigFile().c_str());
}
@ -272,25 +280,25 @@ void NetworkPluginServer::handleConvMessagePayload(const std::string &data, bool
conv->handleMessage(msg, payload.nickname());
}
void NetworkPluginServer::handleDataRead(boost::shared_ptr<Swift::Connection> c, const Swift::ByteArray &data) {
void NetworkPluginServer::handleDataRead(Client *c, const Swift::ByteArray &data) {
long expected_size = 0;
m_data += data.toString();
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 (c->data.size() != 0) {
if (c->data.size() >= 4) {
unsigned char * head = (unsigned char*) 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 (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 = c->data.substr(4, expected_size);
c->data.erase(0, 4 + expected_size);
pbnetwork::WrapperMessage wrapper;
if (wrapper.ParseFromString(msg) == false) {
@ -315,7 +323,7 @@ void NetworkPluginServer::handleDataRead(boost::shared_ptr<Swift::Connection> c,
handleConvMessagePayload(wrapper.payload(), true);
break;
case pbnetwork::WrapperMessage_Type_TYPE_PONG:
m_pongReceived = true;
c->pongReceived = true;
break;
case pbnetwork::WrapperMessage_Type_TYPE_PARTICIPANT_CHANGED:
handleParticipantChangedPayload(wrapper.payload());
@ -339,16 +347,22 @@ void NetworkPluginServer::send(boost::shared_ptr<Swift::Connection> &c, const st
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::list<Client *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
if ((*it)->pongReceived) {
sendPing((*it));
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) {
Client *c = getFreeClient();
user->setData(c);
c->users.push_back(user);
// UserInfo userInfo = user->getUserInfo();
user->onReadyToConnect.connect(boost::bind(&NetworkPluginServer::handleUserReadyToConnect, this, user));
user->onRoomJoined.connect(boost::bind(&NetworkPluginServer::handleRoomJoined, this, user, _1, _2, _3));
@ -368,7 +382,8 @@ void NetworkPluginServer::handleUserReadyToConnect(User *user) {
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LOGIN);
send(m_client, message);
Client *c = (Client *) user->getData();
send(c->connection, message);
}
void NetworkPluginServer::handleRoomJoined(User *user, const std::string &r, const std::string &nickname, const std::string &password) {
@ -385,7 +400,8 @@ void NetworkPluginServer::handleRoomJoined(User *user, const std::string &r, con
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_JOIN_ROOM);
send(m_client, message);
Client *c = (Client *) user->getData();
send(c->connection, message);
NetworkConversation *conv = new NetworkConversation(user->getConversationManager(), r, true);
conv->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, this, _1, _2));
@ -406,7 +422,8 @@ void NetworkPluginServer::handleRoomLeft(User *user, const std::string &r) {
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LEAVE_ROOM);
send(m_client, message);
Client *c = (Client *) user->getData();
send(c->connection, message);
NetworkConversation *conv = (NetworkConversation *) user->getConversationManager()->getConversation(r);
if (!conv) {
@ -428,7 +445,18 @@ void NetworkPluginServer::handleUserDestroyed(User *user) {
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_LOGOUT);
send(m_client, message);
Client *c = (Client *) user->getData();
if (!c) {
return;
}
send(c->connection, message);
c->users.remove(user);
if (c->users.size() == 0) {
std::cout << "DISCONNECTING\n";
c->connection->disconnect();
c->connection.reset();
// m_clients.erase(user->connection);
}
}
void NetworkPluginServer::handleMessageReceived(NetworkConversation *conv, boost::shared_ptr<Swift::Message> &msg) {
@ -442,19 +470,32 @@ void NetworkPluginServer::handleMessageReceived(NetworkConversation *conv, boost
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_CONV_MESSAGE);
send(m_client, message);
Client *c = (Client *) conv->getConversationManager()->getUser()->getData();
send(c->connection, message);
}
void NetworkPluginServer::sendPing() {
void NetworkPluginServer::sendPing(Client *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->connection, message);
c->pongReceived = false;
std::cout << "SENDING PING\n";
}
NetworkPluginServer::Client *NetworkPluginServer::getFreeClient() {
for (std::list<Client *>::const_iterator it = m_clients.begin(); it != m_clients.end(); it++) {
if ((*it)->users.size() < 1) {
if ((*it)->users.size() + 1 == 1) {
exec_(CONFIG_STRING(m_config, "service.backend").c_str(), "localhost", "10000", m_config->getConfigFile().c_str());
}
return (*it);
}
}
return NULL;
}
}