Changeset - cc954f60d118
[Not reviewed]
0 1 0
Jan Kaluza - 14 years ago 2011-06-21 08:18:40
hanzz.k@gmail.com
Check if backend can be executed and write error message if not
1 file changed with 21 insertions and 10 deletions:
0 comments (0 inline, 0 general)
src/networkpluginserver.cpp
Show inline comments
 
@@ -68,135 +68,146 @@ class NetworkFactory : public Factory {
 
			nc->onMessageToSend.connect(boost::bind(&NetworkPluginServer::handleMessageReceived, m_nps, _1, _2));
 
			return nc;
 
		}
 

	
 
		Buddy *createBuddy(RosterManager *rosterManager, const BuddyInfo &buddyInfo) {
 
			LocalBuddy *buddy = new LocalBuddy(rosterManager, buddyInfo.id);
 
			buddy->setAlias(buddyInfo.alias);
 
			buddy->setName(buddyInfo.legacyName);
 
			buddy->setSubscription(buddyInfo.subscription);
 
			buddy->setGroups(buddyInfo.groups);
 
			buddy->setFlags((BuddyFlag) buddyInfo.flags);
 
			if (buddyInfo.settings.find("icon_hash") != buddyInfo.settings.end())
 
				buddy->setIconHash(buddyInfo.settings.find("icon_hash")->second.s);
 
			return buddy;
 
		}
 
	private:
 
		NetworkPluginServer *m_nps;
 
};
 

	
 
#define WRAP(MESSAGE, TYPE) 	pbnetwork::WrapperMessage wrap; \
 
	wrap.set_type(TYPE); \
 
	wrap.set_payload(MESSAGE); \
 
	wrap.SerializeToString(&MESSAGE);
 
	
 
static int exec_(const char *path, const char *host, const char *port, const char *config) {
 
static pid_t exec_(const char *path, const char *host, const char *port, const char *config) {
 
	LOG4CXX_INFO(logger, "Starting new backend " << path);
 
// 	char *argv[] = {(char*)script_name, '\0'}; 
 
	int status = 0;
 
	pid_t pid = fork();
 
	if ( pid == 0 ) {
 
		// child process
 
		execlp(path, path, "--host", host, "--port", port, config, NULL);
 
		abort();
 
		exit(execlp(path, path, "--host", host, "--port", port, config, NULL));
 
	} else if ( pid < 0 ) {
 
		// fork failed
 
		status = -1;
 
	}
 
	return status;
 

	
 
	return pid;
 
}
 

	
 
static void SigCatcher(int n) {
 
	wait3(NULL,WNOHANG,NULL);
 
	pid_t result;
 
	int status;
 
	while ((result = waitpid(0, &status, WNOHANG)) > 0) {
 
		if (result != 0) {
 
			if (WIFEXITED(status)) {
 
				if (WEXITSTATUS(status) != 0) {
 
					LOG4CXX_ERROR(logger, "Backend can not be started, exit_code=" << WEXITSTATUS(status));
 
				}
 
			}
 
			else {
 
				LOG4CXX_ERROR(logger, "Backend can not be started");
 
			}
 
		}
 
	}
 
}
 

	
 
static void handleBuddyPayload(LocalBuddy *buddy, const pbnetwork::Buddy &payload) {
 
	buddy->setName(payload.buddyname());
 
	buddy->setAlias(payload.alias());
 
	std::vector<std::string> groups;
 
	groups.push_back(payload.groups());
 
	buddy->setGroups(groups);
 
	buddy->setStatus(Swift::StatusShow((Swift::StatusShow::Type) payload.status()), payload.statusmessage());
 
	buddy->setIconHash(payload.iconhash());
 
}
 

	
 
NetworkPluginServer::NetworkPluginServer(Component *component, Config *config, UserManager *userManager) {
 
	m_userManager = userManager;
 
	m_config = config;
 
	m_component = component;
 
	m_component->m_factory = new NetworkFactory(this);
 
	m_userManager->onUserCreated.connect(boost::bind(&NetworkPluginServer::handleUserCreated, this, _1));
 
	m_userManager->onUserDestroyed.connect(boost::bind(&NetworkPluginServer::handleUserDestroyed, this, _1));
 

	
 
	m_pingTimer = component->getNetworkFactories()->getTimerFactory()->createTimer(10000);
 
	m_pingTimer->onTick.connect(boost::bind(&NetworkPluginServer::pingTimeout, this)); 
 
	m_pingTimer->onTick.connect(boost::bind(&NetworkPluginServer::pingTimeout, this));
 
	m_pingTimer->start();
 

	
 
	m_vcardResponder = new VCardResponder(component->getIQRouter(), userManager);
 
	m_vcardResponder->onVCardRequired.connect(boost::bind(&NetworkPluginServer::handleVCardRequired, this, _1, _2, _3));
 
	m_vcardResponder->onVCardUpdated.connect(boost::bind(&NetworkPluginServer::handleVCardUpdated, this, _1, _2));
 
	m_vcardResponder->start();
 

	
 
	m_rosterResponder = new RosterResponder(component->getIQRouter(), userManager);
 
	m_rosterResponder->onBuddyAdded.connect(boost::bind(&NetworkPluginServer::handleBuddyAdded, this, _1, _2));
 
	m_rosterResponder->onBuddyRemoved.connect(boost::bind(&NetworkPluginServer::handleBuddyRemoved, this, _1));
 
	m_rosterResponder->onBuddyUpdated.connect(boost::bind(&NetworkPluginServer::handleBuddyUpdated, this, _1, _2));
 
	m_rosterResponder->start();
 

	
 
	m_server = Swift::BoostConnectionServer::create(Swift::HostAddress(CONFIG_STRING(m_config, "service.backend_host")), boost::lexical_cast<int>(CONFIG_STRING(m_config, "service.backend_port")), component->getNetworkFactories()->getIOServiceThread()->getIOService(), component->m_loop);
 
	m_server->onNewConnection.connect(boost::bind(&NetworkPluginServer::handleNewClientConnection, this, _1));
 
	m_server->start();
 

	
 
	LOG4CXX_INFO(logger, "Listening on host " << CONFIG_STRING(m_config, "service.backend_host") << " port " << CONFIG_STRING(m_config, "service.backend_port"));
 

	
 
	signal(SIGCHLD, SigCatcher);
 

	
 
	exec_(CONFIG_STRING(m_config, "service.backend").c_str(), CONFIG_STRING(m_config, "service.backend_host").c_str(), CONFIG_STRING(m_config, "service.backend_port").c_str(), m_config->getConfigFile().c_str());
 
}
 

	
 
NetworkPluginServer::~NetworkPluginServer() {
 
	m_pingTimer->stop();
 
	delete m_vcardResponder;
 
	delete m_rosterResponder;
 
}
 

	
 
void NetworkPluginServer::handleNewClientConnection(boost::shared_ptr<Swift::Connection> c) {
 
	Client *client = new Client;
 
	client->pongReceived = true;
 
	client->connection = c;
 

	
 
	LOG4CXX_INFO(logger, "New backend " << client << " connected. Current backend count=" << (m_clients.size() + 1));
 

	
 
	if (m_clients.size() == 0) {
 
		// first backend connected, start the server, we're ready.
 
		m_component->start();
 
	}
 

	
 
	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(Client *c) {
 
	LOG4CXX_INFO(logger, "Backend " << c << " disconnected. Current backend count=" << (m_clients.size() - 1));
 
	for (std::list<User *>::const_iterator it = c->users.begin(); it != c->users.end(); it++) {
 
		LOG4CXX_ERROR(logger, "Backend " << c << " disconnected (probably crashed) with active user " << (*it)->getJID().toString());
 
		(*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() < CONFIG_INT(m_config, "service.users_per_backend")) {
 
			return;
 
		}
 
	}
 
	exec_(CONFIG_STRING(m_config, "service.backend").c_str(), CONFIG_STRING(m_config, "service.backend_host").c_str(), CONFIG_STRING(m_config, "service.backend_port").c_str(), m_config->getConfigFile().c_str());
 
}
 

	
 
void NetworkPluginServer::handleConnectedPayload(const std::string &data) {
 
	pbnetwork::Connected payload;
 
@@ -443,52 +454,52 @@ void NetworkPluginServer::handleDataRead(Client *c, const Swift::SafeByteArray &
 
				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;
 
			default:
 
				return;
 
		}
 
	}
 
}
 

	
 
void NetworkPluginServer::send(boost::shared_ptr<Swift::Connection> &c, const std::string &data) {
 
	char header[4];
 
	*((int*)(header)) = htonl(data.size());
 
	c->write(Swift::createSafeByteArray(std::string(header, 4) + data));
 
}
 

	
 
void NetworkPluginServer::pingTimeout() {
 
	// check ping responses
 
	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(), CONFIG_STRING(m_config, "service.backend_host").c_str(), CONFIG_STRING(m_config, "service.backend_port").c_str(), m_config->getConfigFile().c_str());
 
		}
 
	}
 
}
 

	
 
void NetworkPluginServer::handleUserCreated(User *user) {
 
	Client *c = getFreeClient();
 
	if (!c) {
 
		LOG4CXX_ERROR(logger, "There is no backend to handle user " << user->getJID().toString());
 
		user->handleDisconnected("Internal Server Error, please reconnect.");
 
		return;
 
	}
 
	user->setData(c);
 
	c->users.push_back(user);
 

	
 
// 	UserInfo userInfo = user->getUserInfo();
 
	user->onReadyToConnect.connect(boost::bind(&NetworkPluginServer::handleUserReadyToConnect, this, user));
 
	user->onPresenceChanged.connect(boost::bind(&NetworkPluginServer::handleUserPresenceChanged, this, user, _1));
 
	user->onRoomJoined.connect(boost::bind(&NetworkPluginServer::handleRoomJoined, this, user, _1, _2, _3));
 
	user->onRoomLeft.connect(boost::bind(&NetworkPluginServer::handleRoomLeft, this, user, _1));
 
}
 

	
0 comments (0 inline, 0 general)