include/anna/comm/ClientSocket.hpp

   1 // ANNA - Anna is Not Nothingness Anymore
   2 //
   3 // (c) Copyright 2005-2014 Eduardo Ramos Testillano & Francisco Ruiz Rayo
   4 //
   5 // http://redmine.teslayout.com/projects/anna-suite
   6 //
   7 // Redistribution and use in source and binary forms, with or without
   8 // modification, are permitted provided that the following conditions
   9 // are met:
  10 //
  11 //     * Redistributions of source code must retain the above copyright
  12 // notice, this list of conditions and the following disclaimer.
  13 //     * Redistributions in binary form must reproduce the above
  14 // copyright notice, this list of conditions and the following disclaimer
  15 // in the documentation and/or other materials provided with the
  16 // distribution.
  17 //     *  Neither the name of the copyright holder nor the names of its
  18 // contributors may be used to endorse or promote products derived from
  19 // this software without specific prior written permission.
  20 //
  21 // THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22 // "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23 // LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  24 // A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  25 // OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  26 // SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  27 // LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  28 // DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  29 // THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  30 // (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  31 // OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  32 //
  33 // Authors: eduardo.ramos.testillano@gmail.com
  34 //          cisco.tierra@gmail.com
  35
  36
  37 #ifndef anna_comm_ClientSocket_hpp
  38 #define anna_comm_ClientSocket_hpp
  39
  40 #include <anna/comm/Socket.hpp>
  41 #include <anna/comm/Buffer.hpp>
  42 #include <anna/comm/Transport.hpp>
  43 #include <anna/core/util/Millisecond.hpp>
  44
  45 namespace anna {
  46
  47 namespace comm {
  48
  49 class Communicator;
  50 class DatagramSocket;
  51 class Transport;
  52 class Message;
  53 class Server;
  54 class CongestionController;
  55 class Receiver;
  56
  57 namespace handler {
  58 class MetaClientSocket;
  59 class DatagramSocket;
  60 }
  61
  62 /**
  63    Implements client sockets (aka @ref Socket) established by our application with a remote endpoint
  64    where a ServerSocket is accepting connection requests.
  65 */
  66 class ClientSocket : public Socket {
  67 public:
  68   /**
  69      Numero de milisegundos por defecto que espera antes de dar por fallida una conexion con
  70      otro proceso servidor.
  71   */
  72   static const Millisecond DefaultMaxConnectionDelay;
  73
  74   /**
  75      Numero de milisegundos por defecto que queda bloqueado un proceso a la espera de poder
  76      escribir en un socket cuyo buffer de salida esta lleno.
  77   */
  78   static const Millisecond DefaultMaxWriteDelay;
  79
  80   /**
  81      Crea un socket cliente liberado.
  82      \param transportFactory factoria de protocolos de transporte a usar por este sockets.
  83      \param domain Dominio del socket.
  84      \param type Tipo de socket.
  85      \warning La factoria de protocolos debe estar disponible mientras el Socket esta activo.
  86   */
  87   ClientSocket(TransportFactory* transportFactory = NULL, Domain::_v domain = Socket::Domain::Inet, Type::_v type = Socket::Type::Stream) :
  88     Socket(domain, type, transportFactory),
  89     a_data(true),
  90     a_reader(true),
  91     a_ignoreIncomingMessages(false) {
  92     initialize();
  93   }
  94
  95   /**
  96      Crea un socket cliente que sera conectado remotamente a la direccion y puerto indicado.
  97
  98      \param remoteAddress direccion de red remota a la que conectar.
  99      \param transportFactory factoria de protocolos de transporte a usar por este sockets.
 100      \param type Tipo de socket.
 101      \warning La factoria de protocolos debe estar disponible mientras el Socket esta activo.
 102   */
 103   ClientSocket(const INetAddress& remoteAddress, TransportFactory* transportFactory = NULL, const Type::_v type = Socket::Type::Stream) :
 104     Socket(Socket::Domain::Inet, type, transportFactory),
 105     a_remoteAccessPoint(remoteAddress),
 106     a_data(true),
 107     a_reader(true),
 108     a_ignoreIncomingMessages(false) {
 109     initialize();
 110   }
 111
 112   /**
 113      Crea un socket cliente que sera conectado al archivo indicado.
 114
 115      \param path Ruta del archivo que vamos a usar para transferir datos ataves de este socket.
 116      \param type Tipo de socket.
 117   */
 118   ClientSocket(const std::string& path, const Type::_v type = Socket::Type::Stream) :
 119     Socket(path, type),
 120     a_remoteAccessPoint(path),
 121     a_data(true),
 122     a_reader(true),
 123     a_ignoreIncomingMessages(false) {
 124     initialize();
 125   }
 126
 127   /**
 128      Crea un socket cliente conectado al servidor indicado por la direccion y puerto remoto. Ademas el socket cliente
 129      seria conectado localmente (ver @ref Socket::bind) a la direccion y puerto locales indicados.
 130
 131      \param transportFactory factoria de protocolos de transporte a usar por este sockets.
 132      \param remoteAddress direccion del servidor.
 133      \param localAddress Puede ser usado para limitar la direccion por la que atendiende peticiones un servidor de socket
 134      instalado en una maquina con mas de una direccion.
 135      compartido por mas de un proceso activo.
 136      \param type Tipo de socket.
 137      \warning La factoria de protocolos debe estar disponible mientras el Socket esta activo.
 138   */
 139   ClientSocket(const INetAddress& remoteAddress,  const INetAddress& localAddress, TransportFactory* transportFactory = NULL, const Type::_v type = Socket::Type::Stream) :
 140     Socket(localAddress, type, transportFactory),
 141     a_remoteAccessPoint(remoteAddress),
 142     a_data(true),
 143     a_reader(true),
 144     a_ignoreIncomingMessages(false) {
 145     initialize();
 146   }
 147
 148   /**
 149      Destructor.
 150   */
 151   virtual ~ClientSocket() { close(); }
 152
 153   /**
 154      Devuelve la direccion remota del socket.
 155      \return La direccion remota del socket.
 156   */
 157   const AccessPoint& getRemoteAccessPoint() const throw() { return a_remoteAccessPoint; }
 158
 159   /**
 160      Devuelve el estado de conexion de este socket.
 161      \return \em true si el socket mantiene la conexion realizada mediante el metodo #connect o
 162      \em false en otro caso.
 163   */
 164   bool isConnected() const throw() { return (a_status & Status::Connected) != 0; }
 165
 166   /**
 167      Devuelve el estado del contenido del socket.
 168      \return \em true si el socket mantiene el sincronismo o \em false en otro caso.
 169   */
 170   bool isSynchronized() const throw() { return (a_status & Status::Corrupt) == 0; }
 171
 172   /**
 173      Devuelve el estado del contenido del socket.
 174      \return \em true si el socket NO mantiene el sincronismo o \em false en otro caso.
 175   */
 176   bool isCorrupt() const throw() { return (a_status & Status::Corrupt); }
 177
 178   /**
 179    * Devuelve el estado de cierre del socket.
 180    * \return \em true si el socket está a la espera de ser cerrado o \em false en otro caso.
 181    */
 182   bool isClosedPending() const throw() { return (a_status & Status::ClosePending) != 0; }
 183
 184   /**
 185      Devuelve la instancia de anna::comm::Server asociado a una conexion remota iniciada por
 186      nuestra aplicacion.
 187      \return la instancia de anna::comm::Server asociado a una conexion remota iniciada por
 188      nuestra aplicacion.
 189      \warning Puede ser NULL en caso de que este ClientSocket corresponda a una conexion local
 190      establecida por un proceso remoto contra un ServerSocket de nuestra aplicacion.
 191   */
 192   Server* getServer() throw(RuntimeException);
 193
 194   /**
 195      Devuelve el numero maximo de milisegundos esperados para obtener conexion al invocar
 196      al metodo #connect.
 197      \return el numero maximo de milisegundos esperados para obtener conexion al invocar
 198      al metodo #connect.
 199   */
 200   const Millisecond &getMaxConnectionDelay() const throw() { return a_msMaxConnectionDelay; }
 201
 202   /**
 203      Devuelve el numero maximo de milisegundos que queda bloqueado el proceso/thread a la espera
 204      de escribir en un socket cuyo buffer de salida esta lleno.
 205      \return Devuelve el numero maximo de milisegundos que queda bloqueado el proceso/thread a la espera
 206      de escribir en un socket cuyo buffer de salida esta lleno.
 207   */
 208   const Millisecond &getMaxWriteDelay() const throw() { return a_msMaxWriteDelay; }
 209
 210   /**
 211      Obtiene toda la informacion referente a este socket a partir de la conexion realizada atraves del \em fd
 212      recibido como parametro.
 213      \param fd File descriptor correspondiente a una conexion local.
 214      \warning Exclusivamente uso interno.
 215   */
 216   virtual void setfd(const int fd) throw(RuntimeException);
 217
 218   /**
 219      Establece el numero maximo de milisegundos esperados para obtener la conexion al
 220      invocar al metodo #connect.
 221      \param msMaxConnectionDelay Numero de milisegundos esperados para obtener conexion.
 222
 223      \see anna::comm::Server::setMaxConnectionDelay.
 224   */
 225   void setMaxConnectionDelay(const Millisecond &msMaxConnectionDelay)
 226   throw() {
 227     a_msMaxConnectionDelay = msMaxConnectionDelay;
 228   }
 229
 230   /**
 231      Establece el numero maximo de milisegundos que queda bloqueado el proceso/thread a la espera
 232      de escribir en un socket cuyo buffer de salida esta lleno.
 233
 234      \param msMaxWriteDelay Numero de milisegundos esperados en caso de que el buffer del socket se llene.
 235   */
 236   void setMaxWriteDelay(const Millisecond &msMaxWriteDelay) throw() { a_msMaxWriteDelay = msMaxWriteDelay; }
 237
 238   /**
 239    * Devuelve \em true si el indicador que ignora los mensajes entrantes está activo, o \em false en otro caso.
 240    * \return \em true si el indicador que ignora los mensajes entrantes está activo, o \em false en otro caso.
 241    */
 242   bool getIgnoreIncomingMessages() const throw() { return a_ignoreIncomingMessages; }
 243
 244   /**
 245    * Establece el indicador que provoca ignorar los mensajes entrantes.
 246    * \param ignoreIncomingMessages \em true si el indicador que ignora los mensajes entrantes está activo, o \em false en otro caso.
 247    */
 248   void setIgnoreIncomingMessages(const bool ignoreIncomingMessages) throw() { a_ignoreIncomingMessages = ignoreIncomingMessages; }
 249
 250   /**
 251      Intenta la conexion remota con la direccion indicada en el constructor.
 252   */
 253   virtual void connect() throw(RuntimeException);
 254
 255   /**
 256      Envia el mensaje recibido como parametro. Al bloque de datos correspondiente a la
 257      codificacion del mensaje se incorpora la informacion necesaria para el protocolo
 258      de la capa de transporte indicado en el constuctor.
 259
 260      \param message Mensaje que vamos codificar para enviar a la capa de transporte.
 261   */
 262   void send(Message& message) throw(RuntimeException);
 263
 264   /**
 265      Envia el mensaje recibido como parametro. Al bloque de datos correspondiente a la
 266      codificacion del mensaje se incorpora la informacion necesaria para el protocolo
 267      de la capa de transporte indicado en el constuctor.
 268
 269      \param message Mensaje que vamos codificar para enviar a la capa de transporte.
 270   */
 271   void send(Message* message) throw(RuntimeException);
 272
 273   /**
 274      Espera la llegada de mensajes por este ClientSocket durante un numero de milisegundos
 275      recibido como parametro o hasta que llegue un mensaje. Un ejemplo de uso, que deberia
 276      ser completado con el control de condiciones de error podria ser:
 277
 278      \code
 279
 280      void f (anna::ClientSocket* clientSocket)
 281         throw (anna::RuntimeException)
 282      {
 283         anna::Guard (clientSocket);
 284
 285         if (clientSocket->wait (2000) == ClientSocket::Notify::ReceiveData) {
 286            const Message* data;
 287
 288            while ((data = clientSocket->fetch ()) != NULL) {
 289               ...
 290               ..... procesa los datos ....
 291               ...
 292            }
 293         }
 294      }
 295
 296      \endcode
 297
 298      \param timeout Numero de milisegundos en que va a quedar bloqueado a la espera de
 299      obtener el primer mensaje.
 300      \param receive Un valor \em true indica que se tratara el mensaje detectado, un \em false
 301      indica que solo esperara la llegada de un bloque de datos, pero no se procesara en absoluto, solo
 302      devolvera Notify::ReceiveData.
 303
 304      \return El resultado de la espera. En caso de no haber recibido ningun mensaje
 305      devolvera comm::Socket::Notify::None.
 306
 307      \warning \li El hilo de ejecucion que invoque a este metodo queda bloqueado durante el
 308      tiempo que indica el 'timeout' o hasta que llegue un mensaje.
 309      \li La invocacion al metodo #receive y al #fetch deben estar protegidas por la misma
 310      seccion critica.
 311   */
 312   Notify::_v wait(const Millisecond &timeout, const bool receive = true) throw(RuntimeException);
 313
 314   /**
 315      Elimina el contenido del buffer de recepcion.
 316   */
 317   void forgot() throw();
 318
 319   /**
 320      Obtiene la capa de transporte asociada a este ClientSocket.
 321      \return la capa de transporte asociada a este ClientSocket.
 322   */
 323   Transport* getTransport() throw(RuntimeException) {
 324     return (a_transport != NULL) ? a_transport : reserveTransport();
 325   }
 326
 327   /**
 328      Obtiene el receptor asociado a este ClientSocket. Puede ser NULL
 329      \return el receptor asociado a este ClientSocket.Puede ser NULL
 330      \warning Exclusivamente uso interno.
 331   */
 332   Receiver* getReceiver() throw(RuntimeException) {
 333     return (a_receiver != NULL) ? a_receiver : ((a_receiverFactory == NULL) ? NULL : reserveReceiver());
 334   }
 335
 336   /**
 337    * Establece el receptor externo que tratara los mensajes recibidos por este socket.
 338    * Si el receptor se establece directamente por el usuario, sin tener una factoria de receptores
 339    * intermedia, invoca directamente al metodo anna::comm::Receiver::initialize.
 340    * \param receive La instancia del receptor externo que tratar los mensajes de este socket.
 341    */
 342   void setReceiver(Receiver* receive) throw(RuntimeException);
 343
 344   /**
 345      Activa la solicitud de cierre del socket, que se llevara a cabo cuando el nucleo considere
 346      que se puede realizar de forma segura, desde el punto de vista de la informacion bloqueada
 347      por secciones criticas en los distintos componentes, que hacen uso del socket.
 348   */
 349   void requestClose() throw();
 350
 351   /**
 352      Devuelve el estado de la solicitud de cierre del socket.
 353      \return el estado de la solicitud de cierre del socket.
 354   */
 355   bool hasRequestedClose() const throw() { return (a_status & Status::ClosePending) != 0; }
 356
 357   /**
 358      Devuelve una cadena con la informacin referente a este socket.
 359      \return Una cadena con la informacin referente a este socket.
 360   */
 361   virtual std::string asString() const throw();
 362
 363   /**
 364      Devuelve un nodo XML con la informacin referente a este objeto.
 365      \param parent Nodo XML a partir del cual introducir la informacin.
 366      \return Un nodo XML con la informacin referente a este objeto.
 367   */
 368   virtual xml::Node* asXML(xml::Node* parent) const throw(RuntimeException);
 369
 370   /**
 371      Devuelve el nombre logico de esta clase.
 372      \return el nombre logico de esta clase.
 373   */
 374   static const char* className() throw() { return "anna::comm::ClientSocket"; }
 375
 376 protected:
 377   struct Status {
 378     enum _v { None = 0, Connected = 1, Corrupt = 2, ClosePending = 4, Working = 8 };
 379
 380     static std::string asString(const int status) throw();
 381   };
 382
 383   Transport* a_transport;
 384   AccessPoint a_remoteAccessPoint;
 385
 386   /**
 387      Devuelve el numero de bytes contenido en el buffer intermedio de recepcion.
 388      \return El numero de bytes contenido en el buffer intermedio de recepcion.
 389   */
 390   int getBufferSize() const throw() { return a_data.getSize() - a_offset; }
 391
 392   /**
 393      Devuelve el numero maximo de bytes que puede tener en el buffer intermedio de recepcion.
 394      \return el numero maximo de bytes que puede tener en el buffer intermedio de recepcion.
 395   */
 396   int getReceiveBufferSize() const throw() { return a_rcvBufferSize; }
 397
 398   /**
 399      Obtiene los parametros de funcionamiento del Socket.
 400      \warning Exclusivamente uso interno.
 401   */
 402   void getSocketOptions() throw(RuntimeException);
 403
 404   /**
 405      Recupera el ultimo mensaje recibido.
 406      El invocador debera haber establecido la forma de asegurar que hay que hay datos
 407       disponibles. Ademas debera establecer el tratamiento adecuado para cada uno
 408      de los posibles resultados.
 409      \return El resultado de la operacion.
 410      \warning La invocacion al metodo #receive y al #fetch deben estar protegidas por la
 411      misma seccion critica. Por ejemplo:
 412
 413      \code
 414
 415      void f (anna::ClientSocket* clientSocket)
 416         throw (anna::RuntimeException)
 417      {
 418         anna::Guard (clientSocket);
 419
 420         if (clientSocket->receive () == ClientSocket::Notify::ReceiveData) {
 421            const Message* data;
 422
 423            while ((data = clientSocket->fetch ()) != NULL) {
 424               ...
 425               ..... procesa los datos ....
 426               ...
 427            }
 428         }
 429      }
 430
 431      \endcode
 432   */
 433   Notify::_v receive() throw(RuntimeException);
 434
 435   /**
 436      Recupera el ultimo bloque de datos recuperado mediante el metodo #receive o #wait.
 437      El bloque de datos deberia ser interpretado segun las reglas de la capa de transporte
 438      asociada a este ClientSocket.
 439
 440      \return El ultimo bloque recuperado mediante el metodo #receive o #wait. Puede ser NULL si no hay
 441      disponible ningun bloque.
 442
 443      \warning Exclusivamente uso interno. Debe invocarse con una seccion
 444      critica activa sobre el ClientSocket.
 445   */
 446   const DataBlock* fetch() throw(RuntimeException);
 447
 448   void activate(const Status::_v v) throw() { a_status |= v; }
 449   void deactivate(const Status::_v v) throw() { a_status &= ~v; }
 450   void deactivate(const int v) throw() { a_status &= ~v; }
 451
 452   virtual int do_connect(const sockaddr*, const int len) throw(RuntimeException);
 453   virtual void do_write(const DataBlock&) throw(RuntimeException);
 454   virtual int do_read(const char* data, const int size) throw(RuntimeException);
 455   virtual void do_close() throw();
 456
 457 private:
 458   struct PendingBytes {
 459     Millisecond validUntil;
 460     int bytesToRead;
 461
 462     PendingBytes() { validUntil = 0; bytesToRead = 0; }
 463   };
 464
 465   //-----------------------------------------------------------------------------------------
 466   // a_status : Combinacion de bit's que indica el estado de la instancia.
 467   // a_expectedSize: Numero de bytes esperados. valdra -1 si todavia no hay informacion
 468   //                 para calcularlo.
 469   // a_data: Ultimo bloque de datos leido. Puede contener un numero indeterminado de mensajes
 470   // a_buffer: Referencia al espacio que ocupa el mensaje que estamos tratanto. Apunta
 471   // a un espacio dentro de 'a_data'-
 472   // a_reader: Buffer usado para leer los datos del Socket.
 473   // a_offset: Desplazamiento que hay que aplicar sobre los datos contenidos en 'a_data'.
 474   // a_cachedServer: Ultimo server calculado en el metodo getServer.
 475   //-----------------------------------------------------------------------------------------
 476   int a_status;
 477   int a_expectedSize;
 478   DataBlock a_data;
 479   Buffer a_buffer;
 480   DataBlock a_reader;
 481   int a_offset;
 482   comm::Server* a_cachedServer;
 483   Millisecond a_msMaxConnectionDelay;
 484   Millisecond a_msMaxWriteDelay;
 485   int a_rcvBufferSize;
 486   Receiver* a_receiver;
 487   mutable PendingBytes a_pendingBytes;
 488   bool a_ignoreIncomingMessages;
 489
 490   void initialize() throw();
 491   void calculeExpectedSize(const DataBlock&) throw();
 492   Transport* reserveTransport() throw(RuntimeException);
 493   Transport* unsafe_reserveTransport() throw(RuntimeException);
 494   Receiver* reserveReceiver() throw(RuntimeException);
 495   int getTotalPendingBytes() const throw();
 496
 497   friend class Communicator;
 498   // unsafe_reserveTransport
 499
 500
 501   friend class CongestionController;
 502   friend class handler::MetaClientSocket;
 503   friend class handler::DatagramSocket;
 504 };
 505
 506 }
 507 }
 508
 509 #endif
 510