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 2012-07-18 18:08:16 +02:00
commit 09880c668b
17 changed files with 435 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
backends/libcommuni/singleircnetworkplugin.cpp
View file

@ -58,6 +58,7 @@ void SingleIRCNetworkPlugin::handleLoginRequest(const std::string &user, const s
session->setRealName(FROM_UTF8(legacyName));
session->setHost(FROM_UTF8(m_server));
session->setPort(6667);
session->setEncoding( "utf-8" );
if (!password.empty()) {
std::string identify = m_identify;

5
include/transport/adhoccommand.h
View file

@ -41,6 +41,8 @@ class AdHocCommand {
virtual boost::shared_ptr<Swift::Command> handleRequest(boost::shared_ptr<Swift::Command> payload) = 0;
void addFormField(Swift::FormField::ref field);
const std::string &getId() {
return m_id;
}
@ -57,9 +59,10 @@ class AdHocCommand {
Component *m_component;
Swift::JID m_initiator;
Swift::JID m_to;
std::vector<Swift::FormField::ref> m_fields;
std::string m_id;
private:
std::string m_id;
// This is used to remove AdHocCommand after long inactivity to prevent memory leaks
// caused by users which disconnect before they finish the command.
// AdHocManager uses this to garbage collect old AdHocCommands.

6
include/transport/admininterface.h
View file

@ -30,13 +30,16 @@ class Component;
class StorageBackend;
class UserManager;
class NetworkPluginServer;
class UserRegistration;
class AdminInterface {
public:
AdminInterface(Component *component, UserManager *userManager, NetworkPluginServer *server = NULL, StorageBackend *storageBackend = NULL);
AdminInterface(Component *component, UserManager *userManager, NetworkPluginServer *server = NULL, StorageBackend *storageBackend = NULL, UserRegistration *userRegistration = NULL);
~AdminInterface();
void handleQuery(Swift::Message::ref message);
private:
void handleMessageReceived(Swift::Message::ref message);
@ -44,6 +47,7 @@ class AdminInterface {
StorageBackend *m_storageBackend;
UserManager *m_userManager;
NetworkPluginServer *m_server;
UserRegistration *m_userRegistration;
};
}

10
include/transport/networkpluginserver.h
View file

@ -42,6 +42,7 @@ class VCardResponder;
class RosterResponder;
class BlockResponder;
class DummyReadBytestream;
class AdminInterface;
class NetworkPluginServer {
public:
@ -63,6 +64,10 @@ class NetworkPluginServer {
virtual ~NetworkPluginServer();
void setAdminInterface(AdminInterface *adminInterface) {
m_adminInterface = adminInterface;
}
int getBackendCount() {
return m_clients.size();
}
@ -84,6 +89,7 @@ class NetworkPluginServer {
private:
void handleNewClientConnection(boost::shared_ptr<Swift::Connection> c);
void handleSessionFinished(Backend *c);
void handlePongReceived(Backend *c);
void handleDataRead(Backend *c, boost::shared_ptr<Swift::SafeByteArray> data);
void handleConnectedPayload(const std::string &payload);
@ -100,7 +106,8 @@ class NetworkPluginServer {
void handleStatsPayload(Backend *c, const std::string &payload);
void handleFTStartPayload(const std::string &payload);
void handleFTFinishPayload(const std::string &payload);
void handleFTDataPayload(Backend *b ,const std::string &payload);
void handleFTDataPayload(Backend *b, const std::string &payload);
void handleQueryPayload(Backend *b, const std::string &payload);
void handleUserCreated(User *user);
void handleRoomJoined(User *user, const Swift::JID &who, const std::string &room, const std::string &nickname, const std::string &password);
@ -144,6 +151,7 @@ class NetworkPluginServer {
std::map<unsigned long, FileTransferManager::Transfer> m_filetransfers;
FileTransferManager *m_ftManager;
std::vector<std::string> m_crashedBackends;
AdminInterface *m_adminInterface;
};
}

1
include/transport/protocol.proto
View file

@ -158,6 +158,7 @@ message WrapperMessage {
TYPE_FT_CONTINUE = 28;
TYPE_EXIT = 29;
TYPE_BACKEND_CONFIG = 30;
TYPE_QUERY = 31;
}
required Type type = 1;
optional bytes payload = 2;

7
include/transport/settingsadhoccommand.h
View file

@ -34,12 +34,19 @@ class Component;
class SettingsAdHocCommand : public AdHocCommand {
public:
typedef enum { Init, WaitingForResponse } State;
SettingsAdHocCommand(Component *component, const Swift::JID &initiator, const Swift::JID &to);
/// Destructor.
virtual ~SettingsAdHocCommand();
virtual boost::shared_ptr<Swift::Command> handleRequest(boost::shared_ptr<Swift::Command> payload);
private:
boost::shared_ptr<Swift::Command> getForm();
boost::shared_ptr<Swift::Command> handleResponse(boost::shared_ptr<Swift::Command> payload);
State m_state;
};
class SettingsAdHocCommandFactory : public AdHocCommandFactory {

28
spectrum/src/main.cpp
View file

@ -207,6 +207,16 @@ int main(int argc, char **argv)
std::cerr << "Can't create service.working_dir directory " << CONFIG_STRING(&config, "service.working_dir") << ".\n";
return 1;
}
// create directories
try {
boost::filesystem::create_directories(
boost::filesystem::path(CONFIG_STRING(&config, "service.portfile")).parent_path().string()
);
}
catch (...) {
std::cerr << "Can't create service.portfile directory " << CONFIG_STRING(&config, "service.portfile") << ".\n";
return 1;
}
#ifndef WIN32
if (!CONFIG_STRING(&config, "service.group").empty() ||!CONFIG_STRING(&config, "service.user").empty() ) {
@ -223,6 +233,20 @@ int main(int argc, char **argv)
chown(CONFIG_STRING(&config, "service.working_dir").c_str(), pw->pw_uid, gr->gr_gid);
}
char backendport[20];
FILE* port_file_f;
port_file_f = fopen(CONFIG_STRING(&config, "service.portfile").c_str(), "w+");
if (port_file_f == NULL) {
std::cerr << "Cannot create port_file file " << CONFIG_STRING(&config, "service.portfile").c_str() << ". Exiting\n";
exit(1);
}
sprintf(backendport,"%s\n",CONFIG_STRING(&config, "service.backend_port").c_str());
if (fwrite(backendport,1,strlen(backendport),port_file_f) < strlen(backendport)) {
std::cerr << "Cannot write to port file " << CONFIG_STRING(&config, "service.portfile") << ". Exiting\n";
exit(1);
}
fclose(port_file_f);
if (!no_daemon) {
// daemonize
daemonize(CONFIG_STRING(&config, "service.working_dir").c_str(), CONFIG_STRING(&config, "service.pidfile").c_str());
@ -309,7 +333,9 @@ int main(int argc, char **argv)
NetworkPluginServer plugin(&transport, &config, &userManager, &ftManager);
AdminInterface adminInterface(&transport, &userManager, &plugin, storageBackend);
AdminInterface adminInterface(&transport, &userManager, &plugin, storageBackend, userRegistration);
plugin.setAdminInterface(&adminInterface);
StatsResponder statsResponder(&transport, &userManager, &plugin, storageBackend);
statsResponder.start();

4
spectrum_manager/src/CMakeLists.txt
View file

@ -1,9 +1,9 @@
cmake_minimum_required(VERSION 2.6)
FILE(GLOB SRC *.cpp)
ADD_EXECUTABLE(spectrum2_manager ${SRC} ../../src/config.cpp ../../src/util.cpp)
ADD_EXECUTABLE(spectrum2_manager ${SRC} ../../src/config.cpp ../../src/util.cpp ${CMAKE_CURRENT_SOURCE_DIR}/../../include/transport/protocol.pb.cc)
target_link_libraries(spectrum2_manager ${SWIFTEN_LIBRARY})
target_link_libraries(spectrum2_manager ${SWIFTEN_LIBRARY} ${PROTOBUF_LIBRARIES})
INSTALL(TARGETS spectrum2_manager RUNTIME DESTINATION bin)

245
spectrum_manager/src/main.cpp
View file

@ -1,5 +1,6 @@
#include "managerconfig.h"
#include "transport/config.h"
#include "transport/protocol.pb.h"
#include "Swiften/Swiften.h"
#include "Swiften/EventLoop/SimpleEventLoop.h"
@ -12,6 +13,11 @@
#include "signal.h"
#include "sys/wait.h"
#define WRAP(MESSAGE, TYPE) pbnetwork::WrapperMessage wrap; \
wrap.set_type(TYPE); \
wrap.set_payload(MESSAGE); \
wrap.SerializeToString(&MESSAGE);
using namespace Transport;
@ -19,33 +25,7 @@ using namespace boost::filesystem;
using namespace boost;
static int finished;
static std::string *m;
static void handleDisconnected(Swift::Client *client, const boost::optional<Swift::ClientError> &, const std::string &server) {
std::cout << "[ DISCONNECTED ] " << server << "\n";
if (--finished == 0) {
exit(0);
}
}
static void handleConnected(Swift::Client *client, const std::string &server) {
boost::shared_ptr<Swift::Message> message(new Swift::Message());
message->setTo(server);
message->setFrom(client->getJID());
message->setBody(*m);
client->sendMessage(message);
}
static void handleMessageReceived(Swift::Client *client, Swift::Message::ref message, const std::string &server) {
std::string body = message->getBody();
boost::replace_all(body, "\n", "\n[ OK ] " + server + ": ");
std::cout << "[ OK ] " << server << ": " << body << "\n";
if (--finished == 0) {
exit(0);
}
}
std::string _data;
static std::string searchForBinary(const std::string &binary) {
std::vector<std::string> path_list;
@ -103,6 +83,22 @@ static unsigned long exec_(std::string path, std::string config, std::string jid
return (unsigned long) pid;
}
static int getPort(const std::string &portfile) {
path p(portfile);
if (!exists(p) || is_directory(p)) {
return 0;
}
std::ifstream f(p.string().c_str(), std::ios_base::in);
std::string port;
f >> port;
if (port.empty())
return 0;
return boost::lexical_cast<int>(port);
}
static int isRunning(const std::string &pidfile) {
path p(pidfile);
if (!exists(p) || is_directory(p)) {
@ -288,7 +284,75 @@ static int show_status(ManagerConfig *config) {
return ret;
}
static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactories &networkFactories, const std::string &message) {
static void handleDataRead(boost::shared_ptr<Swift::Connection> m_conn, boost::shared_ptr<Swift::SafeByteArray> data) {
_data += std::string(data->begin(), data->end());
// Parse data while there are some
while (_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 (_data.size() >= 4) {
expected_size = *((unsigned int*) &_data[0]);
expected_size = ntohl(expected_size);
// If we don't have whole wrapper message, wait for next
// handleDataRead call.
if (_data.size() - 4 < expected_size)
return;
}
else {
return;
}
// Parse wrapper message and erase it from buffer.
pbnetwork::WrapperMessage wrapper;
if (wrapper.ParseFromArray(&_data[4], expected_size) == false) {
std::cout << "PARSING ERROR " << expected_size << "\n";
_data.erase(_data.begin(), _data.begin() + 4 + expected_size);
continue;
}
_data.erase(_data.begin(), _data.begin() + 4 + expected_size);
if (wrapper.type() == pbnetwork::WrapperMessage_Type_TYPE_QUERY) {
pbnetwork::BackendConfig payload;
if (payload.ParseFromString(wrapper.payload()) == false) {
std::cout << "PARSING ERROR\n";
// TODO: ERROR
continue;
}
std::cout << payload.config() << "\n";
exit(0);
}
}
}
static void handleConnected(boost::shared_ptr<Swift::Connection> m_conn, const std::string &msg, bool error) {
if (error) {
std::cerr << "Can't connect the server\n";
exit(50);
}
else {
pbnetwork::BackendConfig m;
m.set_config(msg);
std::string message;
m.SerializeToString(&message);
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_QUERY);
uint32_t size = htonl(message.size());
char *header = (char *) &size;
// send header together with wrapper message
m_conn->write(Swift::createSafeByteArray(std::string(header, 4) + message));
}
}
static void ask_local_server(ManagerConfig *config, Swift::BoostNetworkFactories &networkFactories, const std::string &jid, const std::string &message) {
path p(CONFIG_STRING(config, "service.config_directory"));
try {
@ -302,6 +366,7 @@ static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactorie
exit(7);
}
bool found = false;
directory_iterator end_itr;
for (directory_iterator itr(p); itr != end_itr; ++itr) {
if (is_regular(itr->path()) && extension(itr->path()) == ".cfg") {
@ -311,22 +376,34 @@ static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactorie
continue;
}
if (CONFIG_VECTOR(&cfg, "service.admin_jid").empty() || CONFIG_STRING(&cfg, "service.admin_password").empty()) {
std::cerr << itr->path().string() << ": service.admin_jid or service.admin_password empty. This server can't be queried over XMPP.\n";
if (CONFIG_STRING(&cfg, "service.jid") != jid) {
continue;
}
finished++;
Swift::Client *client = new Swift::Client(CONFIG_VECTOR(&cfg, "service.admin_jid")[0], CONFIG_STRING(&cfg, "service.admin_password"), &networkFactories);
client->setAlwaysTrustCertificates();
client->onConnected.connect(boost::bind(&handleConnected, client, CONFIG_STRING(&cfg, "service.jid")));
client->onDisconnected.connect(bind(&handleDisconnected, client, _1, CONFIG_STRING(&cfg, "service.jid")));
client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, CONFIG_STRING(&cfg, "service.jid")));
Swift::ClientOptions opt;
opt.allowPLAINWithoutTLS = true;
client->connect(opt);
found = true;
boost::shared_ptr<Swift::Connection> m_conn;
m_conn = networkFactories.getConnectionFactory()->createConnection();
m_conn->onDataRead.connect(boost::bind(&handleDataRead, m_conn, _1));
m_conn->onConnectFinished.connect(boost::bind(&handleConnected, m_conn, message, _1));
m_conn->connect(Swift::HostAddressPort(Swift::HostAddress(CONFIG_STRING(&cfg, "service.backend_host")), getPort(CONFIG_STRING(&cfg, "service.portfile"))));
// finished++;
// Swift::Client *client = new Swift::Client(CONFIG_VECTOR(&cfg, "service.admin_jid")[0], CONFIG_STRING(&cfg, "service.admin_password"), &networkFactories);
// client->setAlwaysTrustCertificates();
// client->onConnected.connect(boost::bind(&handleConnected, client, CONFIG_STRING(&cfg, "service.jid")));
// client->onDisconnected.connect(bind(&handleDisconnected, client, _1, CONFIG_STRING(&cfg, "service.jid")));
// client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, CONFIG_STRING(&cfg, "service.jid")));
// Swift::ClientOptions opt;
// opt.allowPLAINWithoutTLS = true;
// client->connect(opt);
}
}
if (!found) {
std::cerr << "Config file for Spectrum instance with this JID was not found\n";
exit(20)
}
}
catch (const filesystem_error& ex) {
std::cerr << "boost filesystem error\n";
@ -334,24 +411,71 @@ static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactorie
}
}
// static void ask_local_servers(ManagerConfig *config, Swift::BoostNetworkFactories &networkFactories, const std::string &message) {
// path p(CONFIG_STRING(config, "service.config_directory"));
//
// try {
// if (!exists(p)) {
// std::cerr << "Config directory " << CONFIG_STRING(config, "service.config_directory") << " does not exist\n";
// exit(6);
// }
//
// if (!is_directory(p)) {
// std::cerr << "Config directory " << CONFIG_STRING(config, "service.config_directory") << " does not exist\n";
// exit(7);
// }
//
// directory_iterator end_itr;
// for (directory_iterator itr(p); itr != end_itr; ++itr) {
// if (is_regular(itr->path()) && extension(itr->path()) == ".cfg") {
// Config cfg;
// if (cfg.load(itr->path().string()) == false) {
// std::cerr << "Can't load config file " << itr->path().string() << ". Skipping...\n";
// continue;
// }
//
// if (CONFIG_VECTOR(&cfg, "service.admin_jid").empty() || CONFIG_STRING(&cfg, "service.admin_password").empty()) {
// std::cerr << itr->path().string() << ": service.admin_jid or service.admin_password empty. This server can't be queried over XMPP.\n";
// continue;
// }
//
// finished++;
// Swift::Client *client = new Swift::Client(CONFIG_VECTOR(&cfg, "service.admin_jid")[0], CONFIG_STRING(&cfg, "service.admin_password"), &networkFactories);
// client->setAlwaysTrustCertificates();
// client->onConnected.connect(boost::bind(&handleConnected, client, CONFIG_STRING(&cfg, "service.jid")));
// client->onDisconnected.connect(bind(&handleDisconnected, client, _1, CONFIG_STRING(&cfg, "service.jid")));
// client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, CONFIG_STRING(&cfg, "service.jid")));
// Swift::ClientOptions opt;
// opt.allowPLAINWithoutTLS = true;
// client->connect(opt);
// }
// }
// }
// catch (const filesystem_error& ex) {
// std::cerr << "boost filesystem error\n";
// exit(5);
// }
// }
int main(int argc, char **argv)
{
ManagerConfig config;
std::string config_file;
std::string command;
std::vector<std::string> command;
boost::program_options::variables_map vm;
boost::program_options::options_description desc("Usage: spectrum [OPTIONS] <COMMAND>\nCommands:\n"
boost::program_options::options_description desc("Usage: spectrum [OPTIONS] <COMMAND>\n"
" spectrum [OPTIONS] <instance_JID> <other>\nCommands:\n"
" start - start all local Spectrum2 instances\n"
" stop - stop all local Spectrum2 instances\n"
" stop - stop all local Spectrum2 instances\n"
" status - status of local Spectrum2 instances\n"
" <other> - send command to all local + remote Spectrum2 instances and print output\n"
" <other> - send command to local Spectrum2 instance and print output\n"
"Allowed options");
desc.add_options()
("help,h", "Show help output")
("config,c", boost::program_options::value<std::string>(&config_file)->default_value("/etc/spectrum2/spectrum_manager.cfg"), "Spectrum manager config file")
("command", boost::program_options::value<std::string>(&command)->default_value(""), "Command")
("command", boost::program_options::value<std::vector<std::string> >(&command), "Command")
;
try
{
@ -388,37 +512,32 @@ int main(int argc, char **argv)
return 1;
}
if (command == "start") {
if (command[0] == "start") {
start_all_instances(&config);
}
else if (command == "stop") {
else if (command[0] == "stop") {
stop_all_instances(&config);
}
else if (command == "status") {
else if (command[0] == "status") {
return show_status(&config);
}
else {
if (command.size() < 2) {
std::cout << desc << "\n";
return 11;
}
Swift::SimpleEventLoop eventLoop;
Swift::BoostNetworkFactories networkFactories(&eventLoop);
std::string message = command;
m = &message;
std::string jid = command[0];
command.erase(command.begin());
std::string cmd = boost::algorithm::join(command, " ");
ask_local_servers(&config, networkFactories, message);
ask_local_server(&config, networkFactories, jid, cmd);
// std::string message = command;
// m = &message;
std::vector<std::string> servers = CONFIG_VECTOR(&config, "servers.server");
for (std::vector<std::string>::const_iterator it = servers.begin(); it != servers.end(); it++) {
finished++;
Swift::Client *client = new Swift::Client(CONFIG_STRING(&config, "service.admin_username") + "@" + *it, CONFIG_STRING(&config, "service.admin_password"), &networkFactories);
client->setAlwaysTrustCertificates();
client->onConnected.connect(boost::bind(&handleConnected, client, *it));
client->onDisconnected.connect(bind(&handleDisconnected, client, _1, *it));
client->onMessageReceived.connect(bind(&handleMessageReceived, client, _1, *it));
Swift::ClientOptions opt;
opt.allowPLAINWithoutTLS = true;
client->connect(opt);
// std::cout << *it << "\n";
}
// ask_local_server(&config, networkFactories, message);
eventLoop.run();
}

4
src/adhoccommand.cpp
View file

@ -45,4 +45,8 @@ AdHocCommand::AdHocCommand(Component *component, const Swift::JID &initiator, co
AdHocCommand::~AdHocCommand() {
}
void AdHocCommand::addFormField(Swift::FormField::ref field) {
m_fields.push_back(field);
}
}

84
src/admininterface.cpp
View file

@ -27,6 +27,7 @@
#include "transport/usermanager.h"
#include "transport/networkpluginserver.h"
#include "transport/logging.h"
#include "transport/userregistration.h"
#include "storageresponder.h"
#include "transport/memoryusage.h"
#include <boost/foreach.hpp>
@ -43,11 +44,12 @@ static std::string getArg(const std::string &body) {
return body.substr(body.find(" ") + 1);
}
AdminInterface::AdminInterface(Component *component, UserManager *userManager, NetworkPluginServer *server, StorageBackend *storageBackend) {
AdminInterface::AdminInterface(Component *component, UserManager *userManager, NetworkPluginServer *server, StorageBackend *storageBackend, UserRegistration *userRegistration) {
m_component = component;
m_storageBackend = storageBackend;
m_userManager = userManager;
m_server = server;
m_userRegistration = userRegistration;
m_component->getStanzaChannel()->onMessageReceived.connect(bind(&AdminInterface::handleMessageReceived, this, _1));
}
@ -55,22 +57,7 @@ AdminInterface::AdminInterface(Component *component, UserManager *userManager, N
AdminInterface::~AdminInterface() {
}
void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
if (!message->getTo().getNode().empty())
return;
std::vector<std::string> const &x = CONFIG_VECTOR(m_component->getConfig(),"service.admin_jid");
if (std::find(x.begin(), x.end(), message->getFrom().toBare().toString()) == x.end()) {
LOG4CXX_WARN(logger, "Message not from admin user, but from " << message->getFrom().toBare().toString());
return;
}
// Ignore empty messages
if (message->getBody().empty()) {
return;
}
void AdminInterface::handleQuery(Swift::Message::ref message) {
LOG4CXX_INFO(logger, "Message from admin received");
message->setTo(message->getFrom());
message->setFrom(m_component->getJID());
@ -267,6 +254,46 @@ void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
int msgCount = m_userManager->getMessagesToXMPP();
message->setBody(boost::lexical_cast<std::string>(msgCount));
}
else if (message->getBody().find("register ") == 0 && m_userRegistration) {
std::string body = message->getBody();
std::vector<std::string> args;
boost::split(args, body, boost::is_any_of(" "));
if (args.size() == 4) {
UserInfo res;
res.jid = args[1];
res.uin = args[2];
res.password = args[3];
res.language = "en";
res.encoding = "utf-8";
res.vip = 0;
if (m_userRegistration->registerUser(res)) {
message->setBody("User registered.");
}
else {
message->setBody("Registration failed: User is already registered");
}
}
else {
message->setBody("Bad argument count. See 'help'.");
}
}
else if (message->getBody().find("unregister ") == 0 && m_userRegistration) {
std::string body = message->getBody();
std::vector<std::string> args;
boost::split(args, body, boost::is_any_of(" "));
if (args.size() == 2) {
if (m_userRegistration->unregisterUser(args[1])) {
message->setBody("User unregistered.");
}
else {
message->setBody("Registration failed: User is not registered");
}
}
else {
message->setBody("Bad argument count. See 'help'.");
}
}
else if (message->getBody().find("help") == 0) {
std::string help;
help += "General:\n";
@ -277,6 +304,10 @@ void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
help += " online_users_count - number of online users\n";
help += " online_users_per_backend - shows online users per backends\n";
help += " has_online_user <bare_JID> - returns 1 if user is online\n";
if (m_userRegistration) {
help += " register <bare_JID> <legacyName> <password> - registers the new user\n";
help += " unregister <bare_JID> - unregisters existing user\n";
}
help += "Messages:\n";
help += " messages_from_xmpp - get number of messages received from XMPP users\n";
help += " messages_to_xmpp - get number of messages sent to XMPP users\n";
@ -299,6 +330,25 @@ void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
else {
message->setBody("Unknown command. Try \"help\"");
}
}
void AdminInterface::handleMessageReceived(Swift::Message::ref message) {
if (!message->getTo().getNode().empty())
return;
std::vector<std::string> const &x = CONFIG_VECTOR(m_component->getConfig(),"service.admin_jid");
if (std::find(x.begin(), x.end(), message->getFrom().toBare().toString()) == x.end()) {
LOG4CXX_WARN(logger, "Message not from admin user, but from " << message->getFrom().toBare().toString());
return;
}
// Ignore empty messages
if (message->getBody().empty()) {
return;
}
handleQuery(message);
m_component->getStanzaChannel()->sendMessage(message);
}

1
src/config.cpp
View file

@ -75,6 +75,7 @@ bool Config::load(std::istream &ifs, boost::program_options::options_description
("service.backend", value<std::string>()->default_value("libpurple_backend"), "Backend")
("service.protocol", value<std::string>()->default_value(""), "Protocol")
("service.pidfile", value<std::string>()->default_value("/var/run/spectrum2/$jid.pid"), "Full path to pid file")
("service.portfile", value<std::string>()->default_value("/var/run/spectrum2/$jid.port"), "File to store backend_port to. It's used by spectrum2_manager.")
("service.working_dir", value<std::string>()->default_value("/var/lib/spectrum2/$jid"), "Working dir")
("service.allowed_servers", value<std::vector<std::string> >()->multitoken(), "Only users from these servers can connect")
("service.server_mode", value<bool>()->default_value(false), "True if Spectrum should behave as server")

105
src/networkpluginserver.cpp
View file

@ -32,6 +32,7 @@
#include "transport/rosterresponder.h"
#include "transport/memoryreadbytestream.h"
#include "transport/logging.h"
#include "transport/admininterface.h"
#include "blockresponder.h"
#include "Swiften/Swiften.h"
#include "Swiften/Server/ServerStanzaChannel.h"
@ -227,6 +228,7 @@ NetworkPluginServer::NetworkPluginServer(Component *component, Config *config, U
m_config = config;
m_component = component;
m_isNextLongRun = false;
m_adminInterface = NULL;
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));
@ -317,18 +319,14 @@ void NetworkPluginServer::handleNewClientConnection(boost::shared_ptr<Swift::Con
client->res = 0;
client->init_res = 0;
client->shared = 0;
client->willDie = 0;
// Until we receive first PONG from backend, backend is in willDie state.
client->willDie = true;
// Backend does not accept new clients automatically if it's long-running
client->acceptUsers = !m_isNextLongRun;
client->longRun = m_isNextLongRun;
LOG4CXX_INFO(logger, "New" + (client->longRun ? std::string(" long-running") : "") + " 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_front(client);
c->onDisconnected.connect(boost::bind(&NetworkPluginServer::handleSessionFinished, this, client));
@ -339,28 +337,6 @@ void NetworkPluginServer::handleNewClientConnection(boost::shared_ptr<Swift::Con
// in first ::pingTimeout call, because it can be called right after this function
// and backend wouldn't have any time to response to ping.
client->pongReceived = -1;
// 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() == 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::handleSessionFinished(Backend *c) {
@ -752,6 +728,68 @@ void NetworkPluginServer::handleFTDataNeeded(Backend *b, unsigned long ftid) {
send(b->connection, message);
}
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_clients.size() == 1) {
// first backend connected, start the server, we're ready.
m_component->start();
}
// 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() == 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);
}
}
}
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 vcard;
vcard.set_config(msg->getBody());
std::string message;
vcard.SerializeToString(&message);
WRAP(message, pbnetwork::WrapperMessage_Type_TYPE_QUERY);
send(b->connection, message);
}
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());
@ -802,7 +840,7 @@ void NetworkPluginServer::handleDataRead(Backend *c, boost::shared_ptr<Swift::Sa
handleConvMessagePayload(wrapper.payload(), true);
break;
case pbnetwork::WrapperMessage_Type_TYPE_PONG:
c->pongReceived = true;
handlePongReceived(c);
break;
case pbnetwork::WrapperMessage_Type_TYPE_PARTICIPANT_CHANGED:
handleParticipantChangedPayload(wrapper.payload());
@ -843,6 +881,9 @@ void NetworkPluginServer::handleDataRead(Backend *c, boost::shared_ptr<Swift::Sa
case pbnetwork::WrapperMessage_Type_TYPE_BUDDY_REMOVED:
handleBuddyRemovedPayload(wrapper.payload());
break;
case pbnetwork::WrapperMessage_Type_TYPE_QUERY:
handleQueryPayload(c, wrapper.payload());
break;
default:
return;
}
@ -896,7 +937,11 @@ void NetworkPluginServer::pingTimeout() {
// pong has been received OR backend just connected and did not have time to answer the ping
// request.
if ((*it)->pongReceived || (*it)->pongReceived == -1) {
sendPing((*it));
// 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.");

8
src/rosterstorage.cpp
View file

@ -86,6 +86,13 @@ RosterStorage::~RosterStorage() {
}
void RosterStorage::storeBuddy(Buddy *buddy) {
if (!buddy) {
return;
}
if (buddy->getName().empty()) {
return;
}
m_buddies[buddy->getName()] = buddy;
m_storageTimer->start();
}
@ -129,6 +136,7 @@ bool RosterStorage::storeBuddies() {
// }
}
m_buddies.clear();
m_storageBackend->commitTransaction();
return true;
}

36
src/settingsadhoccommand.cpp
View file

@ -31,15 +31,47 @@ namespace Transport {
DEFINE_LOGGER(logger, "SettingsAdHocCommand");
SettingsAdHocCommand::SettingsAdHocCommand(Component *component, const Swift::JID &initiator, const Swift::JID &to) : AdHocCommand(component, initiator, to) {
m_state = Init;
}
SettingsAdHocCommand::~SettingsAdHocCommand() {
}
boost::shared_ptr<Swift::Command> SettingsAdHocCommand::getForm() {
boost::shared_ptr<Swift::Command> response(new Swift::Command("settings", m_id, Swift::Command::Executing));
boost::shared_ptr<Swift::Form> form(new Swift::Form());
BOOST_FOREACH(Swift::FormField::ref field, m_fields) {
form->addField(field);
}
response->setForm(form);
return response;
}
boost::shared_ptr<Swift::Command> SettingsAdHocCommand::handleResponse(boost::shared_ptr<Swift::Command> payload) {
boost::shared_ptr<Swift::Command> response;
response->setStatus(Swift::Command::Completed);
return response;
}
boost::shared_ptr<Swift::Command> SettingsAdHocCommand::handleRequest(boost::shared_ptr<Swift::Command> payload) {
boost::shared_ptr<Swift::Command> response;
switch (m_state) {
case Init:
response = getForm();
m_state = WaitingForResponse;
break;
case WaitingForResponse:
response = handleResponse(payload);
break;
default:
break;
}
return response;
}

6
src/sqlite3backend.cpp
View file

@ -199,7 +199,11 @@ bool SQLite3Backend::exec(const std::string &query) {
char *errMsg = 0;
int rc = sqlite3_exec(m_db, query.c_str(), NULL, 0, &errMsg);
if (rc != SQLITE_OK) {
LOG4CXX_ERROR(logger, errMsg << " during statement " << query);
// This error is OK, because we try to create buddies table every time
// to detect if DB is created properly.
if (errMsg != "table buddies already exists") {
LOG4CXX_ERROR(logger, errMsg << " during statement " << query);
}
sqlite3_free(errMsg);
return false;
}

3
src/transport.cpp
View file

@ -191,6 +191,9 @@ void Component::setBuddyFeatures(std::list<std::string> &features) {
void Component::start() {
if (m_component && !m_component->isAvailable()) {
LOG4CXX_INFO(logger, "Connecting XMPP server " << CONFIG_STRING(m_config, "service.server") << " port " << CONFIG_INT(m_config, "service.port"));
if (CONFIG_INT(m_config, "service.port") == 5222) {
LOG4CXX_WARN(logger, "Port 5222 is usually used for client connections, not for component connections! Are you sure you are using right port?");
}
m_reconnectCount++;
m_component->connect(CONFIG_STRING(m_config, "service.server"), CONFIG_INT(m_config, "service.port"));
m_reconnectTimer->stop();