ClientSocket.hpp

Go to the documentation of this file.

// ANNA - Anna is Not Nothingness Anymore                                                         //
//                                                                                                //
// (c) Copyright 2005-2015 Eduardo Ramos Testillano & Francisco Ruiz Rayo                         //
//                                                                                                //
// See project site at http://redmine.teslayout.com/projects/anna-suite                           //
// See accompanying file LICENSE or copy at http://www.teslayout.com/projects/public/anna.LICENSE //


#ifndef anna_comm_ClientSocket_hpp
#define anna_comm_ClientSocket_hpp

#include <anna/comm/Socket.hpp>
#include <anna/comm/Buffer.hpp>
#include <anna/comm/Transport.hpp>
#include <anna/core/util/Millisecond.hpp>

namespace anna {

namespace comm {

class Communicator;
class DatagramSocket;
class Transport;
class Message;
class Server;
class CongestionController;
class Receiver;

namespace handler {
class MetaClientSocket;
class DatagramSocket;
}

class ClientSocket : public Socket {
public:
  static const Millisecond DefaultMaxConnectionDelay;

  static const Millisecond DefaultMaxWriteDelay;

  ClientSocket(TransportFactory* transportFactory = NULL, Domain::_v domain = Socket::Domain::Inet, Type::_v type = Socket::Type::Stream) :
    Socket(domain, type, transportFactory),
    a_data(true),
    a_reader(true),
    a_ignoreIncomingMessages(false) {
    initialize();
  }

  ClientSocket(const INetAddress& remoteAddress, TransportFactory* transportFactory = NULL, const Type::_v type = Socket::Type::Stream) :
    Socket(Socket::Domain::Inet, type, transportFactory),
    a_remoteAccessPoint(remoteAddress),
    a_data(true),
    a_reader(true),
    a_ignoreIncomingMessages(false) {
    initialize();
  }

  ClientSocket(const std::string& path, const Type::_v type = Socket::Type::Stream) :
    Socket(path, type),
    a_remoteAccessPoint(path),
    a_data(true),
    a_reader(true),
    a_ignoreIncomingMessages(false) {
    initialize();
  }

  ClientSocket(const INetAddress& remoteAddress,  const INetAddress& localAddress, TransportFactory* transportFactory = NULL, const Type::_v type = Socket::Type::Stream) :
    Socket(localAddress, type, transportFactory),
    a_remoteAccessPoint(remoteAddress),
    a_data(true),
    a_reader(true),
    a_ignoreIncomingMessages(false) {
    initialize();
  }

  virtual ~ClientSocket() { close(); }

  const AccessPoint& getRemoteAccessPoint() const { return a_remoteAccessPoint; }

  bool isConnected() const { return (a_status & Status::Connected) != 0; }

  bool isSynchronized() const { return (a_status & Status::Corrupt) == 0; }

  bool isCorrupt() const { return (a_status & Status::Corrupt); }

  bool isClosedPending() const { return (a_status & Status::ClosePending) != 0; }

  Server* getServer() noexcept(false);

  const Millisecond &getMaxConnectionDelay() const { return a_msMaxConnectionDelay; }

  const Millisecond &getMaxWriteDelay() const { return a_msMaxWriteDelay; }

  virtual void setfd(const int fd) noexcept(false);

  void setMaxConnectionDelay(const Millisecond &msMaxConnectionDelay)
  {
    a_msMaxConnectionDelay = msMaxConnectionDelay;
  }

  void setMaxWriteDelay(const Millisecond &msMaxWriteDelay) { a_msMaxWriteDelay = msMaxWriteDelay; }

  bool getIgnoreIncomingMessages() const { return a_ignoreIncomingMessages; }

  void setIgnoreIncomingMessages(const bool ignoreIncomingMessages) { a_ignoreIncomingMessages = ignoreIncomingMessages; }

  virtual void connect() noexcept(false);

  void send(Message& message) noexcept(false);

  void send(Message* message) noexcept(false);

  Notify::_v wait(const Millisecond &timeout, const bool receive = true) noexcept(false);

  void forgot() ;

  Transport* getTransport() noexcept(false) {
    return (a_transport != NULL) ? a_transport : reserveTransport();
  }

  Receiver* getReceiver() noexcept(false) {
    return (a_receiver != NULL) ? a_receiver : ((a_receiverFactory == NULL) ? NULL : reserveReceiver());
  }

  void setReceiver(Receiver* receive) noexcept(false);

  void requestClose() ;

  bool hasRequestedClose() const { return (a_status & Status::ClosePending) != 0; }

  virtual std::string asString() const ;

  virtual xml::Node* asXML(xml::Node* parent) const noexcept(false);

  static const char* className() { return "anna::comm::ClientSocket"; }

protected:
  struct Status {
    enum _v { None = 0, Connected = 1, Corrupt = 2, ClosePending = 4, Working = 8 };

    static std::string asString(const int status) ;
  };

  Transport* a_transport;
  AccessPoint a_remoteAccessPoint;

  int getBufferSize() const { return a_data.getSize() - a_offset; }

  int getReceiveBufferSize() const { return a_rcvBufferSize; }

  void getSocketOptions() noexcept(false);

  Notify::_v receive() noexcept(false);

  const DataBlock* fetch() noexcept(false);

  void activate(const Status::_v v) { a_status |= v; }
  void deactivate(const Status::_v v) { a_status &= ~v; }
  void deactivate(const int v) { a_status &= ~v; }

  virtual int do_connect(const sockaddr*, const int len) noexcept(false);
  virtual void do_write(const DataBlock&) noexcept(false);
  virtual int do_read(const char* data, const int size) noexcept(false);
  virtual void do_close() ;

private:
  struct PendingBytes {
    Millisecond validUntil;
    int bytesToRead;

    PendingBytes() { validUntil = 0; bytesToRead = 0; }
  };

  //-----------------------------------------------------------------------------------------
  // a_status : Combinacion de bit's que indica el estado de la instancia.
  // a_expectedSize: Numero de bytes esperados. valdra -1 si todavia no hay informacion
  //                 para calcularlo.
  // a_data: Ultimo bloque de datos leido. Puede contener un numero indeterminado de mensajes
  // a_buffer: Referencia al espacio que ocupa el mensaje que estamos tratanto. Apunta
  // a un espacio dentro de 'a_data'-
  // a_reader: Buffer usado para leer los datos del Socket.
  // a_offset: Desplazamiento que hay que aplicar sobre los datos contenidos en 'a_data'.
  // a_cachedServer: Ultimo server calculado en el metodo getServer.
  //-----------------------------------------------------------------------------------------
  int a_status;
  int a_expectedSize;
  DataBlock a_data;
  Buffer a_buffer;
  DataBlock a_reader;
  int a_offset;
  comm::Server* a_cachedServer;
  Millisecond a_msMaxConnectionDelay;
  Millisecond a_msMaxWriteDelay;
  int a_rcvBufferSize;
  Receiver* a_receiver;
  mutable PendingBytes a_pendingBytes;
  bool a_ignoreIncomingMessages;

  void initialize() ;
  void calculeExpectedSize(const DataBlock&) ;
  Transport* reserveTransport() noexcept(false);
  Transport* unsafe_reserveTransport() noexcept(false);
  Receiver* reserveReceiver() noexcept(false);
  int getTotalPendingBytes() const ;

  friend class Communicator;
  // unsafe_reserveTransport


  friend class CongestionController;
  friend class handler::MetaClientSocket;
  friend class handler::DatagramSocket;
};

}
}

#endif