source/comm/Server.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/xml/Node.hpp>
  13
  14 #include <anna/comm/Communicator.hpp>
  15 #include <anna/comm/Server.hpp>
  16 #include <anna/comm/ClientSocket.hpp>
  17 #include <anna/comm/Host.hpp>
  18 #include <anna/comm/Device.hpp>
  19 #include <anna/comm/functions.hpp>
  20 #include <anna/comm/internal/RemoteConnection.hpp>
  21
  22 using namespace std;
  23 using namespace anna;
  24
  25 comm::Server::Server(const string& name, const comm::Host& host, const int remotePort, const bool autoRecovery, comm::TransportFactory* transportFactory) :
  26   comm::Resource(name),
  27   a_host(host),
  28   a_remotePort(remotePort),
  29   a_autoRecovery(autoRecovery),
  30   a_transportFactory(transportFactory),
  31   a_clientSocket(NULL),
  32   a_msMaxConnectionDelay(ClientSocket::DefaultMaxConnectionDelay),
  33   a_msMaxWriteDelay(ClientSocket::DefaultMaxWriteDelay),
  34   a_receiverFactory(NULL),
  35   a_ignoreIncomingMessages(false),
  36   a_sequence(0)
  37   // La a_sequence se usa para diferenciar en las trazas y demás las conexiones que puede haber contra un mismo (IP, port)
  38   // Se establece desde comm::Host
  39
  40 {;}
  41
  42 //----------------------------------------------------------------------------------------
  43 // (1) Para asegurar que el comm::handler::RemoteConnection no lo mete en la lista de
  44 //     servidores pendientes de recuperar.
  45 //----------------------------------------------------------------------------------------
  46 comm::Server::~Server() {
  47   if(a_clientSocket == NULL)
  48     return;
  49
  50   bool* autoRecovery = const_cast <bool*>(&a_autoRecovery);
  51   *autoRecovery = false;
  52
  53   try {
  54     Communicator* communicator = functions::component <Communicator> (ANNA_FILE_LOCATION);
  55     communicator->detach(a_clientSocket);
  56   } catch(Exception& ex) {
  57     ex.trace();
  58   }
  59 }
  60
  61 /*
  62  * Se invoca desde com::handle::RemoteConnection::finalize
  63  * [Tx] -> Communicator
  64  *
  65  * El socket se cierra protegido por esta SSCC para asegurar que los recursos se usan/bloquean en el mismo
  66  * orden que el usado en el Server::connect.
  67  */
  68 void comm::Server::reset()
  69 noexcept(false) {
  70 // La SSCC se establece en el método que invoca a éste
  71 //   Guard guard (*this, "comm::Server::reset");
  72   if(a_clientSocket == NULL)
  73     return;
  74
  75   a_clientSocket->close();
  76   delete a_clientSocket;
  77   a_clientSocket = NULL;
  78 }
  79
  80 void comm::Server::setReceiverFactory(comm::ReceiverFactory& receiverFactory)
  81 {
  82   a_receiverFactory = &receiverFactory;
  83
  84   if(a_clientSocket != NULL)
  85     a_clientSocket->setReceiverFactory(receiverFactory);
  86 }
  87
  88 /*
  89  * El punto conflictivo de llamada es desde el comm::ConnectionRecover::tryRecover
  90  *
  91  *    [T1] -> Communicator
  92  *
  93  * (1) La instancia RemoteConnection se libera en el handler::RemoteConnection::finalize.
  94  * (2) La espera maxima solo se aplica si ha sido modificada .
  95  * (3) La espera para la escritura sobre buffer de salida llenos solo se aplica si ha sido modificada.
  96  *
  97  * Lista de llamadas en 1.11.07
  98  *
  99 [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (90) | thr: 0xb7819b20 | Guard::lock | Reference: (0x9f86090): comm::Host::createServer
 100
 101    [19/11/2009 13:22:07] Information | comm.Host.cc (91) | thr: 0xb7819b20 | comm::Host::createServer | comm::Server { ...
 102
 103    [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (90) | thr: 0xb7819b20 | Guard::lock | Reference: (0x9f865b0): comm::Server::connect
 104
 105       [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (90) | thr: 0xb7819b20 | Guard::lock | Reference: (0xbfd67ea8): comm::Communicator::attach (RemoteConnection)
 106
 107          [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (90) | thr: 0xb7819b20 | Guard::lock | Reference: (0x9f86608): anna::comm::ClientSocket (connect)
 108
 109             [19/11/2009 13:22:07] Debug | comm.Socket.cc (171) | thr: 0xb7819b20 | anna::comm::Socket::open | comm::ClientSocket { comm::Socket { Domain: Inet | Type: Stream | comm::TransportFactory { Name: anna::comm::SureTransport | OverQuotaSize: 0 bytes } | fd: 4 | LocalPoint: { <null>  } | Bound: false } | RcvBufferSize: -1 bytes | Status: None | MaxConDelay: 200 ms | Reserved: 0 | Pending: 0 | Offset: 0 | ExpectedSize: -1 | Punto remoto: { comm::INetAddress { comm::Device { IP: 127.0.0.1 | Status: Up } | Port: 2000 }  } }
 110
 111             [19/11/2009 13:22:07] Debug | comm.Socket.cc (219) | thr: 0xb7819b20 | anna::comm::Socket::setBlockingMode | Current: false | Previous: true
 112
 113             [19/11/2009 13:22:07] Debug | comm.ClientSocket.cc (286) | thr: 0xb7819b20 | anna::comm::ClientSocket::connect | comm::ClientSocket { comm::Socket { Domain: Inet | Type: Stream | comm::TransportFactory { Name: anna::comm::SureTransport | OverQuotaSize: 0 bytes } | fd: 4 | LocalPoint: { <null>  } | Bound: false } | RcvBufferSize: 87380 bytes | Status: Connected | MaxConDelay: 200 ms | Reserved: 0 | Pending: 0 | Offset: 0 | ExpectedSize: -1 | Punto remoto: { comm::INetAddress { comm::Device { IP: 127.0.0.1 | Status: Up } | Port: 2000 }  } }
 114
 115          [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (78) | thr: 0xb7819b20 | Guard::deactivate | Reference: (0x9f86608): anna::comm::ClientSocket (connect)
 116
 117          [19/11/2009 13:22:07] Notice | comm.Communicator.cc (714) | thr: 0xb7819b20 | comm::Communicator::eventCreateConnection | comm::Server { anna::Resource { Nombre: 127.0.0.1:2000 | Habilitado: true | Disponible: true } | Auto-Recovery: false | comm::ClientSocket { comm::Socket { Domain: Inet | Type: Stream | comm::TransportFactory { Name: anna::comm::SureTransport | OverQuotaSize: 0 bytes } | fd: 4 | LocalPoint: { <null>  } | Bound: false } | RcvBufferSize: 87380 bytes | Status: Connected | MaxConDelay: 200 ms | Reserved: 0 | Pending: 0 | Offset: 0 | ExpectedSize: -1 | Punto remoto: { comm::INetAddress { comm::Device { IP: 127.0.0.1 | Status: Up } | Port: 2000 }  } } }
 118
 119          [19/11/2009 13:22:07] Debug | comm.Communicator.cc (213) | thr: 0xb7819b20 | comm::Communicator::attach | comm::handler::RemoteConnection { comm::Handler { anna::Runnable { Id: Handler4 | Running: false | RequestedStop: false } | fd: 4 } | comm::RemoteConnection { comm::ClientSocket { comm::Socket { Domain: Inet | Type: Stream | comm::TransportFactory { Name: anna::comm::SureTransport | OverQuotaSize: 0 bytes } | fd: 4 | LocalPoint: { <null>  } | Bound: false } | RcvBufferSize: 87380 bytes | Status: Connected | MaxConDelay: 200 ms | Reserved: 0 | Pending: 0 | Offset: 0 | ExpectedSize: -1 | Punto remoto: { comm::INetAddress { comm::Device { IP: 127.0.0.1 | Status: Up } | Port: 2000 }  } } } }
 120
 121       [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (78) | thr: 0xb7819b20 | Guard::deactivate | Reference: (0xbfd67ea8): comm::Communicator::attach (RemoteConnection)
 122
 123       [19/11/2009 13:22:07] Debug | comm.Server.cc (163) | thr: 0xb7819b20 | anna::comm::Server::connect | comm::Server { anna::Resource { Nombre: 127.0.0.1:2000 | Habilitado: true | Disponible: true } | Auto-Recovery: false | comm::ClientSocket { comm::Socket { Domain: Inet | Type: Stream | comm::TransportFactory { Name: anna::comm::SureTransport | OverQuotaSize: 0 bytes } | fd: 4 | LocalPoint: { <null>  } | Bound: false } | RcvBufferSize: 87380 bytes | Status: Connected | MaxConDelay: 200 ms | Reserved: 0 | Pending: 0 | Offset: 0 | ExpectedSize: -1 | Punto remoto: { comm::INetAddress { comm::Device { IP: 127.0.0.1 | Status: Up } | Port: 2000 }  } } }
 124
 125    [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (78) | thr: 0xb7819b20 | Guard::deactivate | Reference: (0x9f865b0): comm::Server::connect
 126
 127 [19/11/2009 13:22:07] Local7 | mt.db/Guard.cc (78) | thr: 0xb7819b20 | Guard::deactivate | Reference: (0x9f86090): comm::Host::createServer
 128
 129 */
 130 void comm::Server::connect()
 131 noexcept(false) {
 132   LOGMETHOD(TraceMethod tm(Logger::Local7, "comm::Server", "connect", ANNA_FILE_LOCATION));
 133   Communicator* communicator = functions::component <Communicator> (ANNA_FILE_LOCATION);
 134   /*
 135    * Para asegurar el mismo orden de bloqueo a la hora de tratar la desconexión.
 136    */
 137   Guard guard(communicator, "comm::Communicator from comm::Server::connect");
 138   Guard guard2(*this, "comm::Server::connect");
 139
 140   if(a_clientSocket != NULL && a_clientSocket->isConnected() == true)
 141     return;
 142
 143   Host::const_device_iterator ii;
 144   Host::const_device_iterator maxii = a_host.device_end();
 145   int counter = 0;
 146
 147   for(ii = a_host.device_begin(); ii != maxii; ii ++)
 148     counter += (Host::device(ii)->getStatus() == Device::Status::Up);
 149
 150   if(counter > 0) {
 151     const Millisecond msMaxConnectionDelay = a_msMaxConnectionDelay / counter;
 152     RemoteConnection* remoteConnection = NULL;
 153
 154     for(ii = a_host.device_begin(); ii != maxii && a_clientSocket == NULL; ii ++) {
 155       a_clientSocket = allocateClientSocket(INetAddress(Host::device(ii), a_remotePort), a_transportFactory);
 156
 157       if(msMaxConnectionDelay != 0)
 158         a_clientSocket->setMaxConnectionDelay(msMaxConnectionDelay);            // (2)
 159
 160       if(a_clientSocket->getMaxWriteDelay() != a_msMaxWriteDelay)
 161         a_clientSocket->setMaxWriteDelay(a_msMaxWriteDelay);                    // (3)
 162
 163       if(a_receiverFactory != NULL)
 164         a_clientSocket->setReceiverFactory(*a_receiverFactory);
 165
 166       a_clientSocket->setIgnoreIncomingMessages(a_ignoreIncomingMessages);
 167       remoteConnection = new RemoteConnection(this, a_clientSocket);             // (1)
 168
 169       try {
 170         communicator->attach(remoteConnection);
 171       } catch(RuntimeException& ex) {
 172         delete a_clientSocket;
 173         delete remoteConnection;
 174         a_clientSocket = NULL;
 175         ex.trace();
 176       }
 177     }
 178   }
 179
 180   if(a_clientSocket == NULL) {
 181     string msg(asString());
 182     msg += " | Cannot connect to server";
 183     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 184   }
 185
 186   LOGDEBUG(
 187     string msg("anna::comm::Server::connect | ");
 188     msg += asString();
 189     Logger::debug(msg, ANNA_FILE_LOCATION)
 190   );
 191 }
 192
 193 //---------------------------------------------------------------------------------------
 194 // (1) Mejora de la version 1.0.8 -> Si el Server no tiene reconexion automatica pero
 195 // se intenta usar => se intenta volver a conectar antes de dar el fallo.
 196 //---------------------------------------------------------------------------------------
 197 comm::ClientSocket* comm::Server::send(Message& message)
 198 noexcept(false) {
 199   LOGMETHOD(TraceMethod tm(Logger::Local7, "comm::Server", "send", ANNA_FILE_LOCATION));
 200   Guard guard(*this, "comm::Server::send");
 201   const bool available = isAvailable();
 202   const bool enabled = isEnabled();
 203
 204   if(a_autoRecovery == false && available == false && enabled == true)     // (1)
 205     connect();
 206
 207   if(available == false || enabled == false) {
 208     string msg(asString());
 209     msg += " | Server unavailable";
 210     throw RuntimeException(msg, ANNA_FILE_LOCATION);
 211   }
 212
 213   a_clientSocket->send(message);
 214   return a_clientSocket;
 215 }
 216
 217 comm::ClientSocket* comm::Server::send(Message* message)
 218 noexcept(false) {
 219   if(message == NULL)
 220     throw RuntimeException("anna::comm::Server::send | Cannot send a NULL message", ANNA_FILE_LOCATION);
 221
 222   return send(*message);
 223 }
 224
 225 void comm::Server::setAutoRecovery(bool autoRecovery) {
 226   bool* ar = const_cast <bool*>(&a_autoRecovery);
 227   *ar = autoRecovery;
 228 }
 229
 230 bool comm::Server::isAvailable() const
 231 noexcept(false) {
 232 //   Guard guard (*this, "comm::Server::isAvailable");
 233   return (a_clientSocket == NULL) ? false : (a_clientSocket->isConnected() && (a_clientSocket->isClosedPending() == false));
 234 }
 235
 236 //  Este metodo sea re-escrito en commsec::Server::allocateClientSocket para devolver un commsec::RemoteConnection
 237 comm::ClientSocket* comm::Server::allocateClientSocket(const comm::INetAddress& in, comm::TransportFactory* transportFactory) const
 238 {
 239   return new ClientSocket(in, transportFactory);
 240 }
 241
 242 string comm::Server::asString() const
 243 {
 244   string result("comm::Server { ");
 245   result += Resource::asString();
 246   result += " |  Sequence: ";
 247   result += functions::asString(a_sequence);
 248   result += " | Auto-Recovery: ";
 249   result += functions::asString(a_autoRecovery);
 250   result += " | ";
 251
 252   if(a_clientSocket == NULL) {
 253     result += "Host: ";
 254     result += a_host.getName();
 255     result += functions::asString(" | RemotePort: %d", a_remotePort);
 256   } else
 257     result += a_clientSocket->asString();
 258
 259   return result += " }";
 260 }
 261
 262 xml::Node* comm::Server::asXML(xml::Node* parent) const
 263 noexcept(false) {
 264   xml::Node* result = parent->createChild("comm.Server");
 265   result->createAttribute("Host", a_host.getName());
 266   result->createAttribute("Sequence", a_sequence);
 267   comm::Resource::asAttribute(result);
 268   result->createAttribute("AutoRecovery", functions::asString(a_autoRecovery));
 269   result->createAttribute("RemotePort", a_remotePort);
 270   result->createAttribute("IgnoreIncomingMessages", functions::asString(a_ignoreIncomingMessages));
 271
 272   if(a_clientSocket != NULL)
 273     result->createAttribute("fd", a_clientSocket->getfd());
 274
 275   return result;
 276 }
 277
 278
 279