source/comm/handler/MetaClientSocket.cpp

   1 // ANNA - Anna is Not Nothingness Anymore                                                         //
   2 //                                                                                                //
   3 // (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
   4 //                                                                                                //
   5 // See project site at http://redmine.teslayout.com/projects/anna-suite                           //
   6 // See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //
   7
   8
   9 #include <anna/core/tracing/Logger.hpp>
  10 #include <anna/core/tracing/TraceMethod.hpp>
  11
  12 #include <anna/comm/Communicator.hpp>
  13 #include <anna/comm/ClientSocket.hpp>
  14 #include <anna/comm/Transport.hpp>
  15 #include <anna/comm/Device.hpp>
  16 #include <anna/comm/Receiver.hpp>
  17
  18 #include <anna/comm/handler/MetaClientSocket.hpp>
  19
  20 using namespace std;
  21 using namespace anna;
  22
  23 /*
  24  * Este método se ejecuta desde un único thread (Tx). Cada socket tiene asociado un thread que lo trata.
  25  */
  26 void comm::handler::MetaClientSocket::apply()
  27 noexcept(false) {
  28   LOGMETHOD(TraceMethod tm(Logger::Local7, "comm::handler::MetaClientSocket", "apply", ANNA_FILE_LOCATION));
  29   comm::ClientSocket* clientSocket = getClientSocket();
  30
  31   if(clientSocket == NULL) {
  32     string msg("comm::handler::MetaClientSocket::apply | fd: ");
  33     msg += functions::asString(getfd());
  34     msg += " | getClientSocket returns NULL";
  35     Logger::warning(msg, ANNA_FILE_LOCATION);
  36     a_communicator->detach(this);
  37     return;
  38   }
  39
  40   /*
  41    * En MT se ha comprobado que hay alguna combinaci�n en la que se llega a tratar un socket que ha sido cerrado previamente.
  42    */
  43   if(clientSocket->isOpened() == false) {
  44     string msg("comm::handler::MetaClientSocket::apply | fd: ");
  45     msg += functions::asString(getfd());
  46     msg += " | Detected incoming socket overload";
  47     Logger::error(msg, ANNA_FILE_LOCATION);
  48     requestStop();
  49     return;
  50   }
  51
  52   bool mustDetach(false);
  53   {
  54     Guard guard(clientSocket, "comm::ClientSocket from comm::handler::MetaClientSocket::apply");
  55     comm::ClientSocket::Notify::_v notify = clientSocket->receive();
  56
  57     if(notify == comm::ClientSocket::Notify::ReceiveData) {
  58       const Message* message;
  59       const DataBlock* dataBlock;
  60       int messageCounter = 0;
  61       Transport* transport = clientSocket->getTransport();
  62       Receiver* receiver = clientSocket->getReceiver();
  63       clientSocket->activate(comm::ClientSocket::Status::Working);
  64
  65       while(canContinue() == true && (dataBlock = clientSocket->fetch()) != NULL) {
  66         try {
  67           try {
  68             message = transport->decode(*dataBlock);                          // (4)
  69           } catch(RuntimeException& ex) {
  70             ex.trace();
  71             clientSocket->activate(comm::ClientSocket::Status::Corrupt);
  72             message = NULL;
  73             break;
  74           }
  75
  76           if(message != NULL) {
  77             messageCounter ++;
  78
  79             if(receiver == NULL)
  80               a_communicator->eventReceiveMessage(*clientSocket, *message);
  81             else
  82               receiver->apply(*clientSocket, *message);
  83           }
  84         } catch(RuntimeException& ex) {
  85           ex.trace();
  86         }
  87       }
  88
  89       LOGLOCAL6(
  90         string msg("comm::handler::MetaClientSocket::apply | fd: ");
  91         msg += functions::asString(getfd());
  92         msg += functions::asText(" | Message Counter: ", messageCounter);
  93         Logger::write(Logger::Local6, msg, ANNA_FILE_LOCATION);
  94       );
  95       clientSocket->deactivate(comm::ClientSocket::Status::Working);
  96
  97       if(clientSocket->hasRequestedClose() == true || clientSocket->isCorrupt()) {
  98         if(clientSocket->isCorrupt()) {
  99           string msg(clientSocket->asString());
 100           msg += " | ClientSocket was closed because of wrong message";
 101           Logger::error(msg, ANNA_FILE_LOCATION);
 102           a_communicator->eventDiscardConnection(*clientSocket);
 103         }
 104
 105         mustDetach = true;
 106       }
 107     } else if(notify == comm::ClientSocket::Notify::Close || notify == comm::ClientSocket::Notify::Corrupt) {  // (2)
 108       if(clientSocket->isCorrupt()) {
 109         string msg(clientSocket->asString());
 110         msg += " | ClientSocket was closed because of wrong message";
 111         Logger::error(msg, ANNA_FILE_LOCATION);
 112         a_communicator->eventDiscardConnection(*clientSocket);
 113       }
 114
 115       mustDetach = true;
 116     }
 117   }
 118
 119   /*
 120    * Termina llamando al XXX::finalize sobre-escrito por la clase heredada de esta
 121    */
 122   if(mustDetach == true)
 123     a_communicator->detach(this);
 124 }
 125
 126 //virtual
 127 bool comm::handler::MetaClientSocket::testClose()
 128 noexcept(false) {
 129   LOGMETHOD(TraceMethod tm(Logger::Local7, "comm::handler::MetaClientSocket", "testClose", ANNA_FILE_LOCATION));
 130   comm::ClientSocket* clientSocket = getClientSocket();
 131
 132   if(clientSocket == NULL) {
 133     string msg("comm::handler::MetaClientSocket::testClose | fd: ");
 134     msg += functions::asString(getfd());
 135     msg += " | getClientSocket returns NULL";
 136     Logger::warning(msg, ANNA_FILE_LOCATION);
 137     a_communicator->detach(this);
 138     return false;
 139   }
 140
 141   bool mustClose;
 142   {
 143     Guard guard(clientSocket, "comm::ClientSocket from comm::handler::MetaClientSocket::testClose");
 144     mustClose = clientSocket->hasRequestedClose();
 145   }
 146
 147   /*
 148    * Termina llamando al XXX::finalize sobre-escrito por la clase heredada de esta
 149    */
 150   if(mustClose == true)
 151     a_communicator->detach(this);
 152
 153   return mustClose;
 154 }
 155
 156 /*
 157  * Este metodo se invoca cuando algun elemento externo detecta la caída de toda la red 'address'
 158  *
 159  * Por lo que se puede invocar desde cualquier otro thread no reconocido.
 160  */
 161 void comm::handler::MetaClientSocket::breakAddress(const in_addr_t& address)
 162 {
 163   comm::ClientSocket* clientSocket = getClientSocket();
 164
 165   if(clientSocket == NULL) {
 166     string msg(asString());
 167     msg += " | getClientSocket returns NULL";
 168     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 169   }
 170
 171   {
 172     Guard guard(clientSocket, "comm::ClientSocket from comm::handler::MetaClientSocket::breakAddress");
 173     const comm::AccessPoint& accessPoint = clientSocket->getRemoteAccessPoint();
 174     const comm::Device* device = accessPoint.getINetAddress().getDevice(false);
 175
 176     if(device == NULL)
 177       return;
 178
 179     if(device->getAddress() != address)
 180       return;
 181
 182     LOGDEBUG(
 183       string msg("comm::handler::MetaClientSocket::breakAddress | ");
 184       msg += clientSocket->asString();
 185       Logger::debug(msg, ANNA_FILE_LOCATION);
 186     );
 187   }
 188
 189   /*
 190    * Termina llamando al XXX::finalize sobre-escrito por la clase heredada de esta
 191    */
 192   a_communicator->detach(clientSocket);
 193 }
 194