Mods to improve stability for Zephora.

EPoll.cpp

@@ -2,6 +2,7 @@
 #include "EPoll.h"
 #include "Command.h"
 #include "Exception.h"
+#include <signal.h>
 
 namespace core {
 

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();
       }
    }

Subscription.cpp

@@ -3,8 +3,8 @@
 #include "Log.h"
 #include <algorithm>
 
-namespace core {
-
+namespace core
+{
 
    Subscription::Subscription(std::string id, std::string mode)
        : id(id), mode(mode), owner(NULL) {}
@@ -12,24 +12,30 @@ namespace core {
    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) {
+   int Subscription::subscribe(TCPSession &session)
+   {
       onSubscribe(session);
       subscribers.push_back(&session);
       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,22 +43,26 @@ 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)
+   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;
    }
 

Subscription.h

@@ -5,11 +5,13 @@
 #include <vector>
 #include <string>
 
-namespace core {
+namespace core
+{
 
    class TCPSession;
 
-   class Subscription {
+   class Subscription
+   {
 
    public:
       Subscription(std::string id, std::string mode = "*AUTHOR");
@@ -19,7 +21,8 @@ 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);
@@ -33,7 +36,6 @@ namespace core {
       TCPSession *owner;
 
       std::vector<TCPSession *> subscribers;
-		
    };
 }
 

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;

TCPServer.cpp

@@ -70,9 +70,11 @@ namespace core {
 
    void TCPServer::removeFromSessionList(TCPSession *session) {
       std::vector<TCPSession *>::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) {