Separation of concerns and single responsibility for event loop; Now we can implement file io event too

hamidr · hamidr · commit 8d5b033f5f38 · 2016-10-17T11:29:42.000+03:30
diff --git a/includes/connection.hpp b/includes/connection.hpp
@@ -5,8 +5,6 @@
 #include <memory>
 #include <tuple>
 
-#include "network/unix_socket.hpp"
-
 namespace async_redis {
   namespace redis_impl
   {
@@ -21,7 +19,7 @@ namespace async_redis {
 
       connection(InputOutputHandler &event_loop)
         : event_loop_(event_loop) {
-        socket_ = std::make_shared<SocketType>(event_loop);
+        socket_ = std::make_unique<SocketType>(event_loop);
       }
 
       template<typename ...Args>
@@ -44,12 +42,10 @@ namespace async_redis {
             if (req_queue_.size() == 1)
               socket_->async_read(std::bind(&connection::reply_received, this, std::placeholders::_1, std::placeholders::_2));
           });
-
       }
 
     private:
-      void reply_received(const char* data, ssize_t len) {
-
+      void reply_received(const char* data, int len) {
         ssize_t acc = 0;
 
         while (acc < len && req_queue_.size()) {
@@ -68,7 +64,6 @@ namespace async_redis {
 
             cb(parser);
             req_queue_.pop(); //free the resources
-
           }
         }
 
@@ -77,7 +72,7 @@ namespace async_redis {
       }
 
     private:
-      std::shared_ptr<SocketType> socket_;
+      std::unique_ptr<SocketType> socket_;
       InputOutputHandler& event_loop_;
       std::queue<std::tuple<reply_cb_t, parser_t>> req_queue_;
     };
diff --git a/includes/event_loop/event_loop_ev.h b/includes/event_loop/event_loop_ev.h
@@ -6,32 +6,26 @@
 #include <unordered_map>
 #include <queue>
 
-#include "../network/async_socket.hpp"
-
 namespace async_redis {
   namespace event_loop
   {
     class event_loop_ev
     {
-      using socket_id = int; //fd;
+      using socket_id = int;
       using string = std::string;
       struct socket_queue;
 
     public:
-      using async_socket    = ::async_redis::network::async_socket;
-      using timeout_cb_t    = std::function<void()>;
-      using ready_cb_t      = async_socket::ready_cb_t;
-      using recv_cb_t       = async_socket::recv_cb_t;
-
+      using action              = std::function<void()>;
       using socket_identifier_t = std::unordered_map<socket_id, std::unique_ptr<socket_queue>>::iterator;
 
     private:
       struct timer_watcher
       {
         ev_timer timer;
-        timeout_cb_t timeout_cb;
+        action timeout_cb;
 
-        timer_watcher(double time, const timeout_cb_t& cb)
+        timer_watcher(double time, const action& cb)
           : timeout_cb(cb)
         {
           ev_timer_init (&timer, &event_loop_ev::timer_handler, time, 0.);
@@ -40,20 +34,16 @@ namespace async_redis {
 
       struct socket_queue
       {
-        async_socket& socket;
         event_loop_ev& loop_;
 
         ev_io write_watcher;
         ev_io read_watcher;
 
-        using write_action = std::tuple<string, ready_cb_t>;
-        using read_action = recv_cb_t;
-
-        std::queue<write_action> write_handlers;
-        std::queue<read_action> read_handlers;
+        std::queue<action> write_handlers;
+        std::queue<action> read_handlers;
 
-        socket_queue(event_loop_ev& loop, int fd, async_socket& s)
-          : loop_(loop), socket(s)
+        socket_queue(event_loop_ev& loop, int fd)
+          : loop_(loop)
         {
           ev_io_init(&read_watcher, &event_loop_ev::read_handler, fd, EV_READ);
           ev_io_init(&write_watcher, &event_loop_ev::write_handler, fd, EV_WRITE);
@@ -73,12 +63,12 @@ namespace async_redis {
       event_loop_ev(struct ev_loop *);
       void run();
 
-      socket_identifier_t watch(int, async_socket&);
+      socket_identifier_t watch(int);
       void unwatch(socket_identifier_t&);
 
-      void async_write(socket_identifier_t& id, const string& data, const ready_cb_t& cb);
-      void async_read(socket_identifier_t& id, const recv_cb_t& cb);
-      void async_timeout(double time, const timeout_cb_t& cb );
+      void async_write(socket_identifier_t& id, const action& cb);
+      void async_read(socket_identifier_t& id, const action& cb);
+      void async_timeout(double time, const action& cb );
 
     private:
       static void read_handler(EV_P_ ev_io* w, int revents);
diff --git a/includes/network/async_socket.hpp b/includes/network/async_socket.hpp
@@ -22,12 +22,33 @@ namespace async_redis {
     class connect_socket_exception : socket_excetion {};
     class nonblocking_socket_exception : socket_excetion {};
 
-    struct async_socket
+    template <typename InputOutputHanler>
+    class async_socket
     {
-      using recv_cb_t         = std::function<void (const char*, int )>;
+    public:
+      using socket_identifier_t  = typename InputOutputHanler::socket_identifier_t;
+      using recv_cb_t         = std::function<void (const char*, int)>;
       using ready_cb_t        = std::function<void ()>;
       using connect_handler_t = std::function<void (bool)>;
 
+      async_socket(InputOutputHanler& io)
+        : io_(io)
+      { }
+
+      async_socket(InputOutputHanler &io, int fd)
+        : io_(io)
+      {
+        fd_ = fd;
+        is_connected_ = true;
+
+        id_ = io_.watch(fd_);
+      }
+
+      inline ~async_socket() {
+        close();
+        io_.unwatch(id_);
+      }
+
       inline int send(const string& data) {
         return ::send(fd_, data.data(), data.size(), 0);
       }
@@ -52,49 +73,20 @@ namespace async_redis {
         return ::close(fd_) == 0;
       }
 
-    protected:
-      int fd_ = -1;
-    };
-
-    template <typename InputOutputHanler>
-    class async_socket_t : public async_socket
-    {
-      using socket_identifier_t  = typename InputOutputHanler::socket_identifier_t;
-
-    public:
-      async_socket_t(InputOutputHanler& io)
-        : io_(io)
-      { }
-
-      async_socket_t(InputOutputHanler &io, int fd)
-        : io_(io)
+      void async_write(const string& data, const ready_cb_t& cb)
       {
-        fd_ = fd;
-        is_connected_ = true;
-
-        id_ = io_.watch(fd_, *this);
-      }
-
-      inline ~async_socket_t() {
-        close();
-        io_.unwatch(id_);
-      }
-
-      inline
-      void async_write(const string& data, const ready_cb_t& cb) {
-        return io_.async_write(id_, data, cb);
-      }
-
-      inline
-      void async_write_then_read(const string& data, const recv_cb_t& cb) {
-        return this->async_write(data, [&, cb]() {
-            this->async_read(cb);
+        return io_.async_write(id_, [this, data, cb]() {
+            send(data);
+            cb();
           });
       }
 
-      inline
-      void async_read(const recv_cb_t& cb) {
-        return io_.async_read(id_, cb);
+      void async_read(const recv_cb_t& cb)
+      {
+        return io_.async_read(id_, [&, cb]() {
+            auto l = receive(buffer_, max_length);
+            cb(buffer_, l);
+          });
       }
 
       template <typename SocketType, typename... Args>
@@ -115,7 +107,7 @@ namespace async_redis {
       template<typename SocketType>
       void async_accept(const std::function<void(std::shared_ptr<SocketType>)>& cb)
       {
-        return async_read([&, cb](const char* data, int len) {
+        return io_.async_read(id_, [&, cb]() {
             int fd = this->accept();
             cb(std::make_shared<SocketType>(io_, fd));
             this->async_accept(cb);
@@ -135,7 +127,7 @@ namespace async_redis {
         if (-1 == fcntl(fd_, F_SETFL, fcntl(fd_, F_GETFL) | O_NONBLOCK))
           throw nonblocking_socket_exception();
 
-        id_ = io_.watch(fd_, *this);
+        id_ = io_.watch(fd_);
       }
 
       int connect_to(socket_t* socket_addr, int len) {
@@ -154,6 +146,9 @@ namespace async_redis {
       bool is_connected_ = false;
       InputOutputHanler& io_;
       socket_identifier_t id_;
+      int fd_ = -1;
+      enum { max_length = 1024 };
+      char buffer_[max_length]; //TODO: move to user land!
     };
   }
 }
diff --git a/includes/network/tcp_server.hpp b/includes/network/tcp_server.hpp
@@ -8,31 +8,12 @@
 #include <unordered_map>
 
 namespace async_redis {
-  namespace tcp_server {
+namespace tcp_server {
 
-  using std::string;
-  class test_parser
-  {
-  public:
-    using parser = std::shared_ptr<string>;
-
-    test_parser(parser &ptr)
-    {
-    }
-
-    int append_chunk(const char* chunk, ssize_t length, bool &is_finished) {
-      is_finished = true;
-      return length;
-    }
-  };
-
-
-  template<typename InputOutputHandler, typename ParserPolicy>
+  template<typename InputOutputHandler>
   class tcp_server
   {
   public:
-    using parser_t     = typename ParserPolicy::parser;
-    using receive_cb_t = std::function<void (parser_t)>;
     using tcp_socket   = async_redis::network::tcp_socket<InputOutputHandler>;
 
     tcp_server(InputOutputHandler &event_loop)
@@ -55,12 +36,8 @@ namespace async_redis {
       conns_.emplace(socket, nullptr);
     }
 
-    void data_received(parser_t& data) {
-      LOG_ME(data->second);
-    }
-
   private:
-    void chunk_received(const char* data, ssize_t len, std::shared_ptr<tcp_socket>& socket) {
+    void chunk_received(const char* data, int len, std::shared_ptr<tcp_socket>& socket) {
       ssize_t acc = 0;
       bool is_finished = false;
 
@@ -82,8 +59,8 @@ namespace async_redis {
 
     socket_t listener_;
     InputOutputHandler& loop_;
-    std::unordered_map<socket_t, parser_t> conns_;
+    std::unordered_map<socket_t, void*> conns_;
   };
 
-  }
+}
 }
diff --git a/includes/network/tcp_socket.hpp b/includes/network/tcp_socket.hpp
@@ -6,18 +6,18 @@ namespace async_redis {
   namespace network
   {
     template<typename InputOutputHanler>
-    class tcp_socket : public async_socket_t<InputOutputHanler>
+    class tcp_socket : public async_socket<InputOutputHanler>
     {
 
     public:
       tcp_socket(InputOutputHanler& io)
-        : async_socket_t<InputOutputHanler>(io)
+        : async_socket<InputOutputHanler>(io)
       {
         this->create_socket(AF_INET);
       }
 
       tcp_socket(InputOutputHanler& io, int fd)
-        : async_socket_t<InputOutputHanler>(io, fd)
+        : async_socket<InputOutputHanler>(io, fd)
       {}
 
       bool bind(const string& host, int port)
@@ -27,13 +27,14 @@ namespace async_redis {
         addr.sin_port   = ::htons(port);
         addr.sin_addr.s_addr = inet_addr(host.data());
 
-        // setsockopt (fd_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on));
 
         return this->bind_to((socket_t *)&addr, sizeof(addr)) == 0;
       }
 
       int connect(const string& host, int port)
       {
+        //TODO:
+        // setsockopt (fd_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on));
         struct sockaddr_in addr = {0};
         addr.sin_family = AF_INET;
         addr.sin_port   = ::htons(port);
diff --git a/includes/network/unix_socket.hpp b/includes/network/unix_socket.hpp
@@ -6,20 +6,20 @@ namespace async_redis {
   namespace network
   {
     template<typename InputOutputHanler>
-    class unix_socket : public async_socket_t<InputOutputHanler>
+    class unix_socket : public async_socket<InputOutputHanler>
     {
     public:
 
       inline
       unix_socket(InputOutputHanler &io)
-        : async_socket_t<InputOutputHanler>(io)
+        : async_socket<InputOutputHanler>(io)
       {
         this->create_socket(AF_UNIX);
       }
 
       inline
       unix_socket(InputOutputHanler &io, int fd)
-        : async_socket_t<InputOutputHanler>(io, fd)
+        : async_socket<InputOutputHanler>(io, fd)
       {}
 
       int connect(const string& path) {
diff --git a/includes/redis_client.hpp b/includes/redis_client.hpp
@@ -18,7 +18,7 @@ namespace async_redis {
     {
       using connection_t = connection<InputOutputHandler, SocketType, ::async_redis::parser::redis_response>;
       using reply_cb_t = typename connection_t::reply_cb_t;
-      using connect_handler_t = async_redis::network::async_socket::connect_handler_t;
+      using connect_handler_t = typename async_redis::network::async_socket<SocketType>::connect_handler_t;
 
     public:
       using parser_t = typename connection_t::parser_t;
diff --git a/src/event_loop/event_loop_ev.cpp b/src/event_loop/event_loop_ev.cpp
diff --git a/test/main.cpp b/test/main.cpp

Original file line number	Diff line number	Diff line change
`@@ -6,18 +6,18 @@ namespace async_redis {`
`6`	`6`	`namespace network`
`7`	`7`	`{`
`8`	`8`	`template<typename InputOutputHanler>`
`9`		`- class tcp_socket : public async_socket_t<InputOutputHanler>`
	`9`	`+ class tcp_socket : public async_socket<InputOutputHanler>`
`10`	`10`	`{`
`11`	`11`
`12`	`12`	`public:`
`13`	`13`	`tcp_socket(InputOutputHanler& io)`
`14`		`- : async_socket_t<InputOutputHanler>(io)`
	`14`	`+ : async_socket<InputOutputHanler>(io)`
`15`	`15`	`{`
`16`	`16`	`this->create_socket(AF_INET);`
`17`	`17`	`}`
`18`	`18`
`19`	`19`	`tcp_socket(InputOutputHanler& io, int fd)`
`20`		`- : async_socket_t<InputOutputHanler>(io, fd)`
	`20`	`+ : async_socket<InputOutputHanler>(io, fd)`
`21`	`21`	`{}`
`22`	`22`
`23`	`23`	`bool bind(const string& host, int port)`
`@@ -27,13 +27,14 @@ namespace async_redis {`
`27`	`27`	`addr.sin_port = ::htons(port);`
`28`	`28`	`addr.sin_addr.s_addr = inet_addr(host.data());`
`29`	`29`
`30`		`- // setsockopt (fd_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on));`
`31`	`30`
`32`	`31`	`return this->bind_to((socket_t *)&addr, sizeof(addr)) == 0;`
`33`	`32`	`}`
`34`	`33`
`35`	`34`	`int connect(const string& host, int port)`
`36`	`35`	`{`
	`36`	`+ //TODO:`
	`37`	`+ // setsockopt (fd_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof (on));`
`37`	`38`	`struct sockaddr_in addr = {0};`
`38`	`39`	`addr.sin_family = AF_INET;`
`39`	`40`	`addr.sin_port = ::htons(port);`
Original file line number	Diff line number	Diff line change
`@@ -6,20 +6,20 @@ namespace async_redis {`
`6`	`6`	`namespace network`
`7`	`7`	`{`
`8`	`8`	`template<typename InputOutputHanler>`
`9`		`- class unix_socket : public async_socket_t<InputOutputHanler>`
	`9`	`+ class unix_socket : public async_socket<InputOutputHanler>`
`10`	`10`	`{`
`11`	`11`	`public:`
`12`	`12`
`13`	`13`	`inline`
`14`	`14`	`unix_socket(InputOutputHanler &io)`
`15`		`- : async_socket_t<InputOutputHanler>(io)`
	`15`	`+ : async_socket<InputOutputHanler>(io)`
`16`	`16`	`{`
`17`	`17`	`this->create_socket(AF_UNIX);`
`18`	`18`	`}`
`19`	`19`
`20`	`20`	`inline`
`21`	`21`	`unix_socket(InputOutputHanler &io, int fd)`
`22`		`- : async_socket_t<InputOutputHanler>(io, fd)`
	`22`	`+ : async_socket<InputOutputHanler>(io, fd)`
`23`	`23`	`{}`
`24`	`24`
`25`	`25`	`int connect(const string& path) {`
Original file line number	Diff line number	Diff line change
`@@ -18,7 +18,7 @@ namespace async_redis {`
`18`	`18`	`{`
`19`	`19`	`using connection_t = connection<InputOutputHandler, SocketType, ::async_redis::parser::redis_response>;`
`20`	`20`	`using reply_cb_t = typename connection_t::reply_cb_t;`
`21`		`- using connect_handler_t = async_redis::network::async_socket::connect_handler_t;`
	`21`	`+ using connect_handler_t = typename async_redis::network::async_socket<SocketType>::connect_handler_t;`
`22`	`22`
`23`	`23`	`public:`
`24`	`24`	`using parser_t = typename connection_t::parser_t;`