Cleaned up socket scheduling. Very clean now.

Socket.cpp

@@ -17,7 +17,7 @@ namespace core {
    Socket::~Socket() {
       free(buffer);
       if(descriptor == -1)
-	return;
+		  return;
       onUnregister();
       ePoll.unregisterSocket(this);
       coreutils::Log(coreutils::LOG_DEBUG_3) << "Socket destroyed for socket " << descriptor << ".";
@@ -26,10 +26,9 @@ namespace core {
 
    void Socket::setDescriptor(int descriptor) {
       if((descriptor == -1) && (errno == 24)) {
-	 shutdown("Too many files open");
+			shutdown("Too many files open");
-	 throw coreutils::Exception("Too many files open. Refusing connection.");
+			throw coreutils::Exception("Too many files open. Refusing connection.");
       }
-      lock.lock();
       coreutils::Log(coreutils::LOG_DEBUG_3) << "Descriptor set to " << descriptor << " for Socket.";
       if(descriptor < 3)
         throw coreutils::Exception("Descriptor out of range", __FILE__, __LINE__);
@@ -37,7 +36,6 @@ namespace core {
       onRegister();
       ePoll.registerSocket(this);
       onRegistered();
-      lock.unlock();
    }
 
    int Socket::getDescriptor() {
@@ -63,29 +61,20 @@ namespace core {
    void Socket::onUnregistered() {}
 
    bool Socket::eventReceived(struct epoll_event event) {
-
-      lock.lock();
-
       if(event.events & EPOLLRDHUP) {
          readHangup = true;
          shutdown("hangup received");
       }
-
+      else if(event.events & EPOLLIN) {
-      if(event.events & EPOLLIN) {
          coreutils::ZString zbuffer(buffer, length);
       	receiveData(zbuffer);
       }
-
+      else if(event.events & EPOLLWRNORM) {
-      if(event.events & EPOLLWRNORM) {
 	      writeSocket();
       }
-
+      else if(event.events & EPOLLHUP) {
-      if(event.events & EPOLLHUP) {
          shutdown();
       }
-
-      lock.unlock();
-
       return !shutDown;
    }
 
@@ -107,9 +96,6 @@ namespace core {
       int len;
       int error = -1;
 
-//      for(int ix = 0; ix < buffer.getLength(); ++ix)
-//        buffer[ix] = 0;
-
       if((len = ::read(getDescriptor(), buffer.getData(), buffer.getLength())) >= 0) {
          coreutils::ZString zbuffer(buffer.getData(), len);
          coreutils::Log(coreutils::LOG_DEBUG_1) << zbuffer;
@@ -140,10 +126,9 @@ namespace core {
    void Socket::writeSocket() {
       if(fifo.size() > 0) {
       	 outlock.lock();
-         ::write(descriptor, fifo.front().c_str(), fifo.front().length());
+			if(!shutDown)
+			  ::write(descriptor, fifo.front().c_str(), fifo.front().length());
          fifo.pop();
-//	      if(shutDown && !needsToWrite())
-//	        delete this;
 	      outlock.unlock();
       }
    }
@@ -152,7 +137,7 @@ namespace core {
       outlock.lock();
       fifo.emplace(data);
       outlock.unlock();
-      ePoll.resetSocket(this);
+//      ePoll.resetSocket(this);
       return 1;
    }
 
@@ -168,8 +153,6 @@ namespace core {
       coreutils::Log(coreutils::LOG_DEBUG_2) << "Shutdown requested on socket " << descriptor << " with reason " << text << ".";
       shutDown = true;
       reset = false;
-      //      if(!needsToWrite())
-//      delete this;
    }
 
 }

Socket.h

@@ -167,7 +167,7 @@ namespace core {
 
       std::string text;
       int descriptor = -1;
-      std::mutex lock;
+//      std::mutex lock;
       std::mutex outlock;
       bool readHangup = false;
 

Subscription.cpp

@@ -14,7 +14,6 @@ namespace core
 
    Subscription::~Subscription()
    {
-      coreutils::Log(coreutils::LOG_DEBUG_3) << "Subscription destructor....";
       std::stringstream out;
       out << "cancel:" << id << std::endl;
       for (auto subscriber : subscribers)

SubscriptionManager.cpp

@@ -7,7 +7,9 @@ namespace core {
    SubscriptionManager::SubscriptionManager() {}
 	
 	int SubscriptionManager::add(Subscription &subscription) {
+		lock.lock();
 		subscriptions.insert(std::make_pair(subscription.id, &subscription));
+		lock.unlock();
 		return 1;
 	}
 	
@@ -15,6 +17,7 @@ namespace core {
       int countSubscribed = 0;
       int countPublished = 0;
 
+		lock.lock();		
       std::string temp = "";
       for(auto [key, subscription] : subscriptions) {
          if(temp != "") {
@@ -34,6 +37,7 @@ namespace core {
       }
       coreutils::Log(coreutils::LOG_DEBUG_2) << "Removed session from " << countSubscribed << " subscription(s).";
       coreutils::Log(coreutils::LOG_DEBUG_2) << "Cancelled " << countPublished << " channel(s) for session.";
+		lock.unlock();
       return countSubscribed;
    }
 

SubscriptionManager.h

@@ -23,6 +23,7 @@ namespace core {
 
     private:
       std::map<std::string, Subscription *> subscriptions;
+		std::mutex lock;
 		
    };
 }

TCPServer.cpp

@@ -72,8 +72,10 @@ namespace core {
       std::vector<TCPSession *>::iterator cursor;
 		lock.lock();
       for(cursor = sessions.begin(); cursor < sessions.end(); ++cursor)
-        if(*cursor == session)
+        if(*cursor == session) {				 
-          sessions.erase(cursor);
+			  sessions.erase(cursor);
+			  break;
+		  }		
 		lock.unlock();
    }
 
@@ -101,22 +103,22 @@ namespace core {
 	
    void TCPServer::sendToAll(std::stringstream &data) {
       for(auto session : sessions)
-	session->write(data.str());
+		  session->write(data.str());
       data.str("");
    }
 	
    void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender) {
       for(auto session : sessions)
-	if(session != &sender)
+		  if(session != &sender)
-	  session->write(data.str());
+			 session->write(data.str());
       data.str("");
    }
 	
    void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender, SessionFilter filter) {
       for(auto session : sessions)
-	if(filter.test(*session))
+		  if(filter.test(*session))
-	  if(session != &sender)
+			 if(session != &sender)
-	    session->write(data.str());
+				session->write(data.str());
       data.str("");
    }
 	

Timer.cpp

@@ -4,7 +4,7 @@ namespace core {
    
    Timer::Timer(EPoll &ePoll, double delay = 0.0f) : Socket(ePoll, "Timer") {
       setDescriptor(timerfd_create(CLOCK_REALTIME, 0));
-      ePoll.registerSocket(this);
+//      ePoll.registerSocket(this);
       setTimer(delay);
    }