Merge pull request #20 from hamidr/refactoring

Refactored!

Author: hamidr

CMakeLists.txt

@@ -28,18 +28,39 @@ include_directories("/usr/local/include")
 link_directories("/usr/lib")
 link_directories("/usr/local/lib")
 
-add_library(event_loop 
+
+add_library(event_loop
+  ${PROJECT_SOURCE_DIR}/event_loop/socket_watcher.cpp
   ${PROJECT_SOURCE_DIR}/event_loop/event_loop_ev.cpp)
 
-add_library(parser 
+add_library(network
+  ${PROJECT_SOURCE_DIR}/network/async_socket.cpp
+  ${PROJECT_SOURCE_DIR}/network/unix_socket.cpp
+  ${PROJECT_SOURCE_DIR}/network/tcp_socket.cpp)
+
+add_library(parser
+  ${PROJECT_SOURCE_DIR}/parser/base_resp_parser.cpp
   ${PROJECT_SOURCE_DIR}/parser/array_parser.cpp
   ${PROJECT_SOURCE_DIR}/parser/bulk_string_parser.cpp
   ${PROJECT_SOURCE_DIR}/parser/error_parser.cpp
   ${PROJECT_SOURCE_DIR}/parser/number_parser.cpp
-  ${PROJECT_SOURCE_DIR}/parser/redis_response.cpp
   ${PROJECT_SOURCE_DIR}/parser/simple_string_parser.cpp)
 
+add_library(async_redis
+  ${PROJECT_SOURCE_DIR}/connection.cpp
+  ${PROJECT_SOURCE_DIR}/monitor.cpp
+  ${PROJECT_SOURCE_DIR}/sentinel.cpp
+  ${PROJECT_SOURCE_DIR}/redis_client.cpp)
+
+
+## Compiler flags
+if(CMAKE_COMPILER_IS_GNUCXX)
+    set(CMAKE_CXX_FLAGS "-O2")        ## Optimize
+    set(CMAKE_EXE_LINKER_FLAGS "-s")  ## Strip binary
+endif()
+
 target_link_libraries(event_loop ev)
+target_link_libraries(async_redis event_loop parser network)
 
 install(TARGETS event_loop
   LIBRARY DESTINATION /usr/local/lib/
@@ -52,5 +73,5 @@ install(TARGETS parser
 
 install(DIRECTORY ${PROJECT_INCLUDE_DIR}/ DESTINATION /usr/local/include)
 
-add_executable (test ${CMAKE_SOURCE_DIR}/test/main.cpp)
-target_link_libraries(test parser event_loop ev)
+add_executable (a1.out ${CMAKE_SOURCE_DIR}/test/main.cpp)
+target_link_libraries(a1.out async_redis)

examples/tcp_server.hpp

@@ -10,34 +10,33 @@
 namespace async_redis {
 namespace tcp_server {
 
-  template<typename InputOutputHandler>
   class tcp_server
   {
   public:
-    using tcp_socket   = async_redis::network::tcp_socket<InputOutputHandler>;
+    using tcp_socket   = async_redis::network::tcp_socket;
+    using async_socket   = async_redis::network::async_socket;
 
-    tcp_server(InputOutputHandler &event_loop)
-      : loop_(event_loop) {
-      listener_ = std::make_shared<tcp_socket>(event_loop);
+    tcp_server(event_loop::event_loop_ev &event_loop)
+      : loop_(event_loop), listener_(event_loop) {
     }
 
     void listen(int port) {
-      if (!listener_->bind("127.0.0.1", port) || !listener_->listen())
+      if (!listener_.bind("127.0.0.1", port) || !listener_.listen())
         throw;
 
       auto receiver = std::bind(&tcp_server::accept, this, std::placeholders::_1);
-      listener_->template async_accept<tcp_socket>(receiver);
+      listener_.async_accept(receiver);
     }
 
-    void accept(std::shared_ptr<tcp_socket> socket) {
+    void accept(std::shared_ptr<async_socket> socket) {
       auto receiver = std::bind(&tcp_server::chunk_received, this, std::placeholders::_1, socket);
       socket->async_read(buffer_, max_buffer_length, receiver);
 
       conns_.emplace(socket, nullptr);
     }
 
   private:
-    void chunk_received(int len, std::shared_ptr<tcp_socket>& socket)
+    void chunk_received(int len, std::shared_ptr<async_socket>& socket)
     {
       std::string command;
 
@@ -78,10 +77,10 @@ namespace tcp_server {
     }
 
   private:
-    using socket_t = std::shared_ptr<tcp_socket>;
+    using socket_t = std::shared_ptr<async_socket>;
 
-    socket_t listener_;
-    InputOutputHandler& loop_;
+    tcp_socket listener_;
+    event_loop::event_loop_ev& loop_;
     std::unordered_map<socket_t, void*> conns_;
     enum { max_buffer_length = 1024 };
     char buffer_[max_buffer_length];

includes/connection.hpp

@@ -5,109 +5,39 @@
 #include <memory>
 #include <tuple>
 
-namespace async_redis {
-  namespace redis_impl
-  {
-    using std::string;
-
-    template<typename InputOutputHandler, typename SocketType, typename ParserPolicy>
-    class connection
-    {
-    public:
-      using parser_t        = typename ParserPolicy::parser;
-      using reply_cb_t      = std::function<void (parser_t)>;
-
-      connection(InputOutputHandler &event_loop)
-        : event_loop_(event_loop) {
-        socket_ = std::make_unique<SocketType>(event_loop);
-      }
-
-      template<typename ...Args>
-      inline void connect(Args... args) {
-        if (!socket_->is_valid())
-          socket_ = std::make_unique<SocketType>(event_loop_);
-
-        socket_->template async_connect<SocketType>(0, std::forward<Args>(args)...);
-      }
-
-      bool is_connected() const
-      { return socket_ && socket_->is_connected(); }
-
-      inline int pressure() const
-      { return req_queue_.size(); }
-
-      void disconnect() {
-        socket_->close();
-        decltype(req_queue_) free_me;
-        free_me.swap(req_queue_);
-      }
-
-      bool pipelined_send(string&& pipelined_cmds, std::vector<reply_cb_t>&& callbacks)
-      {
-        return
-          socket_->async_write(pipelined_cmds, [this, cbs = std::move(callbacks)](ssize_t sent_chunk_len) {
-            if (sent_chunk_len == 0)
-              return disconnect();
+#include <parser/base_resp_parser.h>
+#include <network/async_socket.hpp>
 
-            if (!req_queue_.size() && cbs.size())
-              socket_->async_read(data_, max_data_size, std::bind(&connection::reply_received, this, std::placeholders::_1));
-
-            for(auto &&cb : cbs)
-              req_queue_.emplace(std::move(cb), nullptr);
-          });
-      }
-
-      bool send(const string&& command, const reply_cb_t& reply_cb)
-      {
-        bool read_it = !req_queue_.size();
-        req_queue_.emplace(reply_cb, nullptr);
-
-        return
-          socket_->async_write(std::move(command), [this, read_it](ssize_t sent_chunk_len) {
-            if (sent_chunk_len == 0)
-              return disconnect();
-
-          if (read_it)
-            socket_->async_read(data_, max_data_size, std::bind(&connection::reply_received, this, std::placeholders::_1));
-        });
-      }
-
-    protected:
-      void reply_received(ssize_t len)
-      {
-        if (len == 0)
-          return disconnect();
-
-        ssize_t acc = 0;
-        while (acc < len && req_queue_.size())
-        {
-          auto& request = req_queue_.front();
-
-          auto &cb = std::get<0>(request);
-          auto &parser = std::get<1>(request);
+namespace async_redis
+{
+  class connection
+  {
+    using async_socket    = network::async_socket;
 
-          bool is_finished = false;
-          acc += ParserPolicy(parser).append_chunk(data_ + acc, len - acc, is_finished);
+  public:
+    using parser_t        = parser::base_resp_parser::parser;
+    using reply_cb_t      = std::function<void (parser_t)>;
 
-          if (!is_finished)
-            break;
+    connection(event_loop::event_loop_ev& event_loop);
 
-          cb(parser);
-          req_queue_.pop(); //free the resources
-        }
+    void connect(async_socket::connect_handler_t handler, const std::string& ip, int port);
+    void connect(async_socket::connect_handler_t handler, const std::string& path);
 
-        if (req_queue_.size())
-          socket_->async_read(data_, max_data_size, std::bind(&connection::reply_received, this, std::placeholders::_1));
-      }
+    bool is_connected() const;
+    void disconnect();
+    bool pipelined_send(std::string&& pipelined_cmds, std::vector<reply_cb_t>&& callbacks);
+    bool send(const std::string&& command, const reply_cb_t& reply_cb);
 
-    private:
-      std::unique_ptr<SocketType> socket_;
-      std::queue<std::tuple<reply_cb_t, parser_t>> req_queue_;
+  private:
+    void do_read();
+    void reply_received(ssize_t len);
 
-      InputOutputHandler& event_loop_;
-      enum { max_data_size = 1024 };
-      char data_[max_data_size];
-    };
+  private:
+    std::unique_ptr<async_socket> socket_;
+    std::queue<std::tuple<reply_cb_t, parser_t>> req_queue_;
 
-  }
+    event_loop::event_loop_ev& event_loop_;
+    enum { max_data_size = 1024 };
+    char data_[max_data_size];
+  };
 }

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_;
+};
+
+}
+}