Cleaning socket watcher

Author: hamidr

CMakeLists.txt

@@ -30,6 +30,7 @@ link_directories("/usr/local/lib")
 
 
 add_library(event_loop
+  ${PROJECT_SOURCE_DIR}/event_loop/socket_watcher.cpp
   ${PROJECT_SOURCE_DIR}/event_loop/event_loop_ev.cpp)
 
 add_library(network

includes/event_loop/event_loop_ev.h

@@ -1,23 +1,15 @@
 #pragma once
 
-#include <ev.h>
-#include <functional>
+#include <event_loop/socket_watcher.h>
 #include <memory>
-#include <unordered_map>
-#include <queue>
 
 namespace async_redis {
   namespace event_loop
   {
     class event_loop_ev
     {
-      using socket_id = int;
-      using string = std::string;
-      struct socket_queue;
-
     public:
-      using action              = std::function<void()>;
-      using socket_identifier_t = std::shared_ptr<socket_queue>;
+      using socket_identifier_t = std::shared_ptr<socket_watcher>;
 
     private:
       struct timer_watcher
@@ -32,50 +24,22 @@ namespace async_redis {
         }
       };
 
-      struct socket_queue
-      {
-        event_loop_ev& loop_;
-
-        ev_io write_watcher;
-        ev_io read_watcher;
-
-        std::queue<action> write_handlers;
-        std::queue<action> read_handlers;
-
-        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);
-
-          write_watcher.data = this;
-          read_watcher.data = this;
-        }
-
-        void stop() {
-          loop_.stop(write_watcher);
-          loop_.stop(read_watcher);
-        }
-      };
 
     public:
       event_loop_ev();
       event_loop_ev(struct ev_loop *);
+
       void run();
 
       socket_identifier_t watch(int);
       void unwatch(socket_identifier_t&);
 
-      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 );
+      void async_write(socket_identifier_t& id, action&& cb);
+      void async_read(socket_identifier_t& id, action&& cb);
+      void async_timeout(double time, action&& cb );
 
     private:
-      static void read_handler(EV_P_ ev_io* w, int revents);
-      static void write_handler(EV_P_ ev_io* w, int revents);
       static void timer_handler(EV_P_ ev_timer* w, int revents);
-      void stop(ev_io&);
-      void start(ev_io&);
 
     private:
       struct ev_loop* loop_;

includes/event_loop/socket_watcher.h

@@ -0,0 +1,47 @@
+#pragma once
+
+#include <ev.h>
+#include <queue>
+#include <functional>
+
+namespace async_redis {
+namespace event_loop {
+
+typedef std::function<void()> action;
+
+class socket_watcher
+{
+ public:
+  socket_watcher(struct ev_loop* loop, int fd);
+
+  void stop();
+
+  void start_reading_with(action&& cb);
+  void start_writing_with(action&& cb);
+
+ private:
+  void call_read();
+  void call_write();
+
+  static void read_handler(EV_P_ ev_io* w, int revents);
+  static void write_handler(EV_P_ ev_io* w, int revents);
+
+ private:
+  void start_reading();
+  void start_writing();
+
+  void stop_reading();
+  void stop_writing();
+
+ private:
+  struct ev_loop* loop_;
+
+  ev_io write_watcher_;
+  ev_io read_watcher_;
+
+  std::queue<action> write_handlers_;
+  std::queue<action> read_handlers_;
+};
+
+}
+}

includes/monitor.hpp

@@ -2,6 +2,7 @@
 
 #include <parser/array_parser.h>
 
+#include <unordered_map>
 #include <list>
 #include <cassert>
 

includes/network/async_socket.hpp

@@ -3,6 +3,7 @@
 #include <sys/socket.h>
 
 #include <string>
+#include <functional>
 #include <event_loop/event_loop_ev.h>
 
 namespace async_redis {
@@ -21,9 +22,9 @@ namespace async_redis {
       using socket_t = struct sockaddr;
 
       using socket_identifier_t  = event_loop::event_loop_ev::socket_identifier_t;
-      using recv_cb_t         = std::function<void (ssize_t)>;
-      using ready_cb_t        = std::function<void (ssize_t)>;
-      using connect_handler_t = std::function<void (bool)>;
+      using recv_cb_t            = std::function<void (ssize_t)>;
+      using ready_cb_t           = std::function<void (ssize_t)>;
+      using connect_handler_t    = std::function<void (bool)>;
 
       async_socket(event_loop::event_loop_ev& io);
 
@@ -36,8 +37,8 @@ namespace async_redis {
       bool listen(int backlog = 0);
       int accept();
       bool close();
-      bool async_write(const string& data, const ready_cb_t& cb);
-      bool async_read(char *buffer, int max_len, const recv_cb_t& cb);
+      bool async_write(const string& data, ready_cb_t cb);
+      bool async_read(char *buffer, int max_len, recv_cb_t cb);
       void async_accept(const std::function<void(std::shared_ptr<async_socket>)>& cb);
 
       bool is_connected() const;

src/event_loop/event_loop_ev.cpp

@@ -18,94 +18,33 @@ void event_loop_ev::run()
   ev_run (loop_, 0);
 }
 
-void event_loop_ev::async_write(socket_identifier_t& watcher, const action& cb)
+void event_loop_ev::async_write(socket_identifier_t& watcher, action&& cb)
 {
-  auto &handlers = watcher->write_handlers;
-  handlers.push(cb);
-
-  if (watcher->write_handlers.size() == 1) {
-    ev_io *w = &watcher->write_watcher;
-    ev_io_start(loop_, w);
-  }
+  watcher->start_writing_with(std::move(cb));
 }
 
-void event_loop_ev::async_read(socket_identifier_t& watcher, const action& cb)
+void event_loop_ev::async_read(socket_identifier_t& watcher, action&& cb)
 {
-  auto &handlers = watcher->read_handlers;
-  handlers.push(cb);
-
-  if (watcher->read_handlers.size() == 1) {
-    ev_io *w = &watcher->read_watcher;
-    ev_io_start(loop_, w);
-  }
+  watcher->start_reading_with(std::move(cb));
 }
 
-void event_loop_ev::async_timeout(double time, const action& cb )
+void event_loop_ev::async_timeout(double time, action&& cb )
 {
   timer_watcher *w = new timer_watcher(time, cb);
   ev_timer_start (loop_, &w->timer);
 }
 
-void event_loop_ev::read_handler(EV_P_ ev_io* w, int revents)
-{
-  if (revents & EV_ERROR)
-    return;
-
-  socket_queue* sq = reinterpret_cast<socket_queue*>(w->data);
-  auto &handlers = sq->read_handlers;
-
-  if (handlers.size())
-  {
-    auto &action = handlers.front();
-    action();
-    handlers.pop();
-  }
-
-  if (!handlers.size())
-    ev_io_stop(loop, &sq->read_watcher);
-}
-
-void event_loop_ev::write_handler(EV_P_ ev_io* w, int revents)
-{
-  if (revents & EV_ERROR)
-    return;
-
-  socket_queue* sq = reinterpret_cast<socket_queue*>(w->data);
-
-  auto &handlers = sq->write_handlers;
-
-  if (handlers.size())
-  {
-    auto &action = handlers.front();
-    action();
-    handlers.pop();
-  }
-
-  if (!handlers.size())
-    ev_io_stop(loop, &sq->write_watcher);
-}
-
 void event_loop_ev::timer_handler(EV_P_ ev_timer* w, int revents)
 {
   timer_watcher *watcher = reinterpret_cast<timer_watcher*>(w);
   watcher->timeout_cb();
   delete watcher;
 }
 
-void event_loop_ev::stop(ev_io& io)
-{
-  ev_clear_pending(loop_, &io);
-  ev_io_stop(loop_, &io);
-}
-
-void event_loop_ev::start(ev_io& io)
-{
-  ev_io_start(loop_, &io);
-}
 
 event_loop_ev::socket_identifier_t event_loop_ev::watch(int fd)
 {
-  return std::make_shared<event_loop_ev::socket_queue>(*this, fd);
+  return std::make_shared<socket_watcher>(loop_, fd);
 }
 
 void event_loop_ev::unwatch(socket_identifier_t& id)

src/event_loop/socket_watcher.cpp

@@ -0,0 +1,111 @@
+#include "../../includes/event_loop/socket_watcher.h"
+
+namespace async_redis {
+namespace event_loop {
+
+socket_watcher::socket_watcher(struct ev_loop* loop, int fd)
+  : loop_(loop)
+{
+  ev_io_init(&read_watcher_, &socket_watcher::read_handler, fd, EV_READ);
+  ev_io_init(&write_watcher_, &socket_watcher::write_handler, fd, EV_WRITE);
+
+  write_watcher_.data = this;
+  read_watcher_.data = this;
+}
+
+  void socket_watcher::start_writing_with(action&& cb)
+{
+  write_handlers_.push(std::move(cb));
+
+  if (write_handlers_.size() != 1)
+    return;
+
+  start_writing();
+}
+
+void socket_watcher::start_reading_with(action&& cb)
+{
+  read_handlers_.push(std::move(cb));
+
+  if (read_handlers_.size() != 1)
+    return;
+
+  start_reading();
+}
+
+void socket_watcher::call_read()
+{
+  if (read_handlers_.size())
+  {
+    auto &action = read_handlers_.front();
+    action();
+    read_handlers_.pop();
+  }
+
+  if (!read_handlers_.size())
+    stop_reading();
+}
+
+
+void socket_watcher::call_write()
+{
+  if (write_handlers_.size())
+    {
+      auto &action = write_handlers_.front();
+      action();
+      write_handlers_.pop();
+    }
+
+  if (!write_handlers_.size())
+    stop_writing();
+}
+
+void socket_watcher::write_handler(EV_P_ ev_io* w, int revents)
+{
+  if (revents & EV_ERROR)
+    return;
+
+  socket_watcher* sq = reinterpret_cast<socket_watcher*>(w->data);
+  sq->call_write();
+}
+
+void socket_watcher::read_handler(EV_P_ ev_io* w, int revents)
+{
+  if (revents & EV_ERROR)
+    return;
+
+  socket_watcher* sq = reinterpret_cast<socket_watcher*>(w->data);
+  sq->call_read();
+}
+
+
+void socket_watcher::stop_reading()
+{
+  ev_clear_pending(loop_, &read_watcher_);
+  ev_io_stop(loop_, &read_watcher_);
+}
+
+void socket_watcher::stop_writing()
+{
+  ev_clear_pending(loop_, &write_watcher_);
+  ev_io_stop(loop_, &write_watcher_);
+}
+
+void socket_watcher::start_writing()
+{
+  ev_io_start(loop_, &write_watcher_);
+}
+
+void socket_watcher::start_reading()
+{
+  ev_io_start(loop_, &read_watcher_);
+}
+
+void socket_watcher::stop()
+{
+  stop_writing();
+  stop_reading();
+}
+
+}
+}