diff --git a/EPoll.cpp b/EPoll.cpp index db7bad6..4cbd9a0 100644 --- a/EPoll.cpp +++ b/EPoll.cpp @@ -2,6 +2,7 @@ #include "EPoll.h" #include "Command.h" #include "Exception.h" +#include namespace core { diff --git a/Socket.cpp b/Socket.cpp index a17e063..3e89ac0 100644 --- a/Socket.cpp +++ b/Socket.cpp @@ -6,6 +6,8 @@ namespace core { + void sigpipe_handler(int unused) {} + Socket::Socket(EPoll &ePoll, std::string text) : ePoll(ePoll), text(text) { coreutils::Log(coreutils::LOG_DEBUG_2) << "Socket object created [" << text << "]."; buffer = (char *)malloc(4096); @@ -110,7 +112,7 @@ namespace core { if((len = ::read(getDescriptor(), buffer.getData(), buffer.getLength())) >= 0) { coreutils::ZString zbuffer(buffer.getData(), len); -// coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer; + coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer; onDataReceived(zbuffer); } else { @@ -140,8 +142,8 @@ namespace core { outlock.lock(); ::write(descriptor, fifo.front().c_str(), fifo.front().length()); fifo.pop(); - if(shutDown && !needsToWrite()) - delete this; +// if(shutDown && !needsToWrite()) +// delete this; outlock.unlock(); } } diff --git a/Subscription.cpp b/Subscription.cpp index 97e09be..cd00ec8 100644 --- a/Subscription.cpp +++ b/Subscription.cpp @@ -3,33 +3,39 @@ #include "Log.h" #include -namespace core { +namespace core +{ + Subscription::Subscription(std::string id, std::string mode) + : id(id), mode(mode), owner(NULL) {} - Subscription::Subscription(std::string id, std::string mode) - : id(id), mode(mode), owner(NULL) {} + Subscription::Subscription(std::string id, TCPSession &session, std::string mode) + : id(id), mode(mode), owner(&session) {} - Subscription::Subscription(std::string id, TCPSession &session, std::string mode) - : id(id), mode(mode), owner(&session) {} - - Subscription::~Subscription() { + Subscription::~Subscription() + { coreutils::Log(coreutils::LOG_DEBUG_3) << "Subscription destructor...."; std::stringstream out; out << "cancel:" << id << std::endl; - for(auto subscriber : subscribers) { + for (auto subscriber : subscribers) + { subscriber->write(out.str()); } } - int Subscription::subscribe(TCPSession &session) { - onSubscribe(session); + int Subscription::subscribe(TCPSession &session) + { + onSubscribe(session); subscribers.push_back(&session); - return 1; + return 1; } - int Subscription::unsubscribe(TCPSession &session) { - for(auto subscriber = subscribers.begin(); subscriber < subscribers.end(); ++subscriber) { - if(*subscriber == &session) { + int Subscription::unsubscribe(TCPSession &session) + { + for (auto subscriber = subscribers.begin(); subscriber < subscribers.end(); ++subscriber) + { + if (*subscriber == &session) + { subscribers.erase(subscriber++); return 1; } @@ -37,23 +43,27 @@ namespace core { return 0; } - int Subscription::process(coreutils::ZString &request, std::stringstream &out) { + int Subscription::process(coreutils::ZString &request, std::stringstream &out, TCPSession &session) + { out << "event:" << request[1] << ":" << request[2] << std::endl; return 1; } - int Subscription::event(std::stringstream &out) { - for(auto subscriber = subscribers.begin(); subscriber < subscribers.end(); ++subscriber) - (*subscriber)->write(out.str()); + int Subscription::event(std::stringstream &out) + { + for (auto subscriber = subscribers.begin(); subscriber < subscribers.end(); ++subscriber) + (*subscriber)->write(out.str()); return 1; } - bool Subscription::ifSubscriber(TCPSession &session) { + bool Subscription::ifSubscriber(TCPSession &session) + { return (std::find(subscribers.begin(), subscribers.end(), &session) != subscribers.end()); } - int Subscription::onSubscribe(TCPSession &session) { + int Subscription::onSubscribe(TCPSession &session) + { return 0; } - + } diff --git a/Subscription.h b/Subscription.h index 31336e3..371b6c4 100644 --- a/Subscription.h +++ b/Subscription.h @@ -5,11 +5,13 @@ #include #include -namespace core { +namespace core +{ class TCPSession; - class Subscription { + class Subscription + { public: Subscription(std::string id, std::string mode = "*AUTHOR"); @@ -19,13 +21,14 @@ namespace core { int subscribe(TCPSession &session); int unsubscribe(TCPSession &session); - virtual int process(coreutils::ZString &request, std::stringstream &out); + virtual int process(coreutils::ZString &request, std::stringstream &out, TCPSession &session); + virtual int onSubscribe(TCPSession &session); int event(std::stringstream &out); bool ifSubscriber(TCPSession &session); - + // int processCommand(coreutils::ZString &request, TCPSession &session) override; std::string id; @@ -33,7 +36,6 @@ namespace core { TCPSession *owner; std::vector subscribers; - }; } diff --git a/SubscriptionManager.cpp b/SubscriptionManager.cpp index dcc754b..d5575e8 100644 --- a/SubscriptionManager.cpp +++ b/SubscriptionManager.cpp @@ -64,7 +64,7 @@ namespace core { return 1; } else if(request[0].equals("event")) { std::stringstream out; - subscription->process(request, out); + subscription->process(request, out, session); if(subscription->mode == "*ANYONE") { subscription->event(out); return 1; diff --git a/TCPServer.cpp b/TCPServer.cpp index 9bacc86..61dc17e 100644 --- a/TCPServer.cpp +++ b/TCPServer.cpp @@ -70,9 +70,11 @@ namespace core { void TCPServer::removeFromSessionList(TCPSession *session) { std::vector::iterator cursor; + lock.lock(); for(cursor = sessions.begin(); cursor < sessions.end(); ++cursor) if(*cursor == session) sessions.erase(cursor); + lock.unlock(); } void TCPServer::sessionErrorHandler(std::string errorString, std::stringstream &out) { diff --git a/output/main b/output/main index b3e266c..71318cc 100755 Binary files a/output/main and b/output/main differ