merged in publish handlers.

TCPServer.cpp

@@ -1,12 +1,14 @@
 #include "TCPServer.h"
 #include "EPoll.h"
-#include "TCPSession.h"
 #include "Exception.h"
 #include "Log.h"
+#include "TCPSession.h"
 
-namespace core {
+namespace core
+{
 
    TCPServer::TCPServer(EPoll &ePoll, IPAddress address, std::string delimiter, int depth, std::string text)
+<<<<<<< HEAD
      : TCPSocket(ePoll, text), commands(delimiter, depth) {
 
 	commands.add(subscriptions, "publish");
@@ -27,8 +29,32 @@ namespace core {
 	  throw coreutils::Exception("Error on listen to socket");
 
      }
+=======
+       : TCPSocket(ePoll, text), commands(delimiter, depth)
+   {
+>>>>>>> subscription
 
-   TCPServer::~TCPServer() {
+      commands.add(subscriptions, "publish");
+      commands.add(subscriptions, "unpublish");
+      commands.add(subscriptions, "subscribe");
+      commands.add(subscriptions, "unsubscribe");
+      commands.add(subscriptions, "catalog");
+      commands.add(subscriptions, "event");
+      commands.add(subscriptions, "invite");
+
+      setDescriptor(socket(AF_INET, SOCK_STREAM, 0));
+      int yes = 1;
+      setsockopt(getDescriptor(), SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
+
+      if (bind(getDescriptor(), address.getPointer(), address.addressLength) < 0)
+         throw coreutils::Exception("Error on bind to socket: " + std::to_string(errno));
+
+      if (listen(getDescriptor(), 20) < 0)
+         throw coreutils::Exception("Error on listen to socket");
+   }
+
+   TCPServer::~TCPServer()
+   {
       coreutils::Log(coreutils::LOG_DEBUG_2) << "Closing server socket " << getDescriptor() << ".";
       close(getDescriptor());
    }
@@ -36,8 +62,8 @@ namespace core {
    void TCPServer::onDataReceived(std::string data) {
       lock.lock();
       TCPSession *session = accept();
-      if(session)
-        sessions.push_back(session);
+      if (session)
+         sessions.push_back(session);
       lock.unlock();
    }
 
@@ -47,22 +73,24 @@ namespace core {
 
          TCPSession *session = getSocketAccept(ePoll);
          session->setDescriptor(::accept(getDescriptor(), (struct sockaddr *)&session->ipAddress.addr, &session->ipAddress.addressLength));
-	 //      if(blackList && blackList->contains(session->ipAddress.getClientAddress())) {
-	 //         session->shutdown();
-	 //         Log(LOG_WARN) << "Client at IP address " << session->ipAddress.getClientAddress() << " is blacklisted and was denied a connection.";
-	 //         return NULL;
-	 //      }
-	 //      if(whiteList && !whiteList->contains(session->ipAddress.getClientAddress())) {
-	 //         session->shutdown();
-	 //         Log(LOG_WARN) << "Client at IP address " << session->ipAddress.getClientAddress() << " is not authorized and was denied a connection.";
-	 //         return NULL;
-	 //      }
+         //      if(blackList && blackList->contains(session->ipAddress.getClientAddress())) {
+         //         session->shutdown();
+         //         Log(LOG_WARN) << "Client at IP address " << session->ipAddress.getClientAddress() << " is blacklisted and was denied a connection.";
+         //         return NULL;
+         //      }
+         //      if(whiteList && !whiteList->contains(session->ipAddress.getClientAddress())) {
+         //         session->shutdown();
+         //         Log(LOG_WARN) << "Client at IP address " << session->ipAddress.getClientAddress() << " is not authorized and was denied a connection.";
+         //         return NULL;
+         //      }
          return session;
       }
-      catch(coreutils::Exception e) {
+      catch (coreutils::Exception e)
+      {
          coreutils::Log(coreutils::LOG_EXCEPT) << "Major error on session initialization. Error is '" << e.text << "'.";
       }
-      catch(...) {
+      catch (...)
+      {
          coreutils::Log(coreutils::LOG_EXCEPT) << "Unnspecified error on session initialization.";
       }
       return NULL;
@@ -93,7 +121,8 @@ namespace core {
 
    int TCPServer::processCommand(coreutils::ZString &request, TCPSession &session) {
       int sequence = 0;
-      for(auto *sessionx : sessions) {
+      for (auto *sessionx : sessions)
+      {
          session.out << "|" << ++sequence;
          sessionx->output(session.out);
          session.out << "|" << std::endl;
@@ -101,25 +130,41 @@ namespace core {
       return 1;
    }
 
-   void TCPServer::sendToAll(std::stringstream &data) {
-      for(auto session : sessions)
-      	session->write(data.str(), NULL);
+   void TCPServer::sendToAll(std::stringstream &data)
+   {
+      for (auto session : sessions)
+         session->write(data.str());
       data.str("");
    }
 
-   void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender) {
-      for(auto session : sessions)
-	if(session != &sender)
-	  session->write(data.str(), &sender);
+   void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender)
+   {
+      for (auto session : sessions)
+         if (session != &sender)
+            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(session != &sender)
-	    session->write(data.str(), &sender);
+   void TCPServer::sendToAll(std::stringstream &data, TCPSession &sender, SessionFilter filter)
+   {
+      for (auto session : sessions)
+         if (filter.test(*session))
+            if (session != &sender)
+               session->write(data.str());
       data.str("");
    }
 
+   TCPSession *TCPServer::getSessionByAlias(void *alias)
+   {
+      coreutils::Log(coreutils::LOG_DEBUG_1) << alias;
+      for (auto session : sessions)
+         if (compareAlias(session->alias, alias))
+            return session;
+      return NULL;
+   }
+
+   bool TCPServer::compareAlias(void *alias, void *compare)
+   {
+      return alias == compare;
+   }
 }