include/rpc/connection.h

   1 #ifndef connection_h
   2 #define connection_h
   3
   4 #include "include/types.h"
   5 #include <arpa/inet.h>
   6 #include <netinet/in.h>
   7 #include "include/t4.h"
   8 #include "include/rpc/poll_mgr.h"
   9 #include "include/rpc/file.h"
  10 #include "include/debug.h"
  11
  12 class connection;
  13
  14 class connection_delegate {
  15     public:
  16         virtual bool got_pdu(const shared_ptr<connection> & c, const string & b) = 0;
  17         virtual ~connection_delegate();
  18 };
  19
  20 using std::chrono::steady_clock;
  21 using time_point = std::chrono::time_point<steady_clock>;
  22
  23 class connection : private aio_callback, public std::enable_shared_from_this<connection> {
  24     public:
  25         connection(connection_delegate * delegate, socket_t && f1, int lossytest=0);
  26         ~connection();
  27
  28         bool isdead() { return dead_; }
  29
  30         bool send(const string & b);
  31
  32         static shared_ptr<connection> to_dst(const sockaddr_in & dst, connection_delegate *mgr, int lossy=0);
  33
  34         const time_point create_time = steady_clock::now();
  35         socket_t fd;
  36
  37     private:
  38         void write_cb(int s);
  39         void read_cb(int s);
  40
  41         bool readpdu();
  42         bool writepdu();
  43
  44         connection_delegate * delegate_;
  45         bool dead_ = false;
  46
  47         enum charbuf_status_t { unused, inflight, error };
  48
  49         struct charbuf {
  50             charbuf_status_t status;
  51             string buf;
  52             size_t cursor; // number of bytes written or read so far
  53         };
  54
  55         charbuf wpdu_ = {unused, "", 0};
  56         charbuf rpdu_ = {unused, "", 0};
  57
  58         int lossy_ = 0;
  59
  60         std::mutex m_;
  61         cond send_complete_;
  62         cond send_wait_;
  63 };
  64
  65 class connection_listener : private aio_callback {
  66     public:
  67         connection_listener(connection_delegate * delegate, in_port_t port, int lossytest=0);
  68         ~connection_listener();
  69         inline in_port_t port() { return port_; }
  70     private:
  71         void write_cb(int) {}
  72         void read_cb(int s);
  73
  74         in_port_t port_;
  75         std::mutex m_;
  76
  77         socket_t tcp_; // listens for connections
  78         connection_delegate * delegate_;
  79         int lossy_;
  80         std::map<int, shared_ptr<connection>> conns_;
  81 };
  82 #endif