Refactoring finished

Author: hamidr

CMakeLists.txt

@@ -46,6 +46,12 @@ add_library(parser
   ${PROJECT_SOURCE_DIR}/parser/number_parser.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)
@@ -54,6 +60,7 @@ if(CMAKE_COMPILER_IS_GNUCXX)
 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/
@@ -67,4 +74,4 @@ install(TARGETS parser
 install(DIRECTORY ${PROJECT_INCLUDE_DIR}/ DESTINATION /usr/local/include)
 
 add_executable (a1.out ${CMAKE_SOURCE_DIR}/test/main.cpp)
-target_link_libraries(a1.out parser event_loop network)
+target_link_libraries(a1.out async_redis)

includes/connection.hpp

@@ -6,122 +6,31 @@
 #include <tuple>
 
 #include <parser/base_resp_parser.h>
-#include <network/tcp_socket.hpp>
-#include <network/unix_socket.hpp>
+#include <network/async_socket.hpp>
 
 namespace async_redis
 {
   class connection
   {
     using async_socket    = network::async_socket;
-    using tcp_socket      = network::tcp_socket;
-    using unix_socket     = network::unix_socket;
 
   public:
     using parser_t        = parser::base_resp_parser::parser;
     using reply_cb_t      = std::function<void (parser_t)>;
 
-    connection(event_loop::event_loop_ev& event_loop)
-      : event_loop_(event_loop) {
-    }
+    connection(event_loop::event_loop_ev& event_loop);
 
-    void connect(async_socket::connect_handler_t handler, const std::string& ip, int port)
-    {
-      if (!socket_ || !socket_->is_valid())
-        socket_ = std::make_unique<tcp_socket>(event_loop_);
+    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);
 
-      static_cast<tcp_socket*>(socket_.get())->async_connect(ip, port, handler);
-    }
-
-    void connect(async_socket::connect_handler_t handler, const std::string& path)
-    {
-      if (!socket_ || !socket_->is_valid())
-        socket_ = std::make_unique<unix_socket>(event_loop_);
-
-      static_cast<unix_socket*>(socket_.get())->async_connect(path, handler);
-    }
-
-    bool is_connected() const
-    { return socket_ && socket_->is_connected(); }
-
-    inline int pressure() const
-    { return req_queue_.size(); }
-
-    void disconnect() {
-      socket_->close();
-      //TODO: check the policy! Should we free queue or retry again?
-      decltype(req_queue_) free_me;
-      free_me.swap(req_queue_);
-    }
-
-    bool pipelined_send(std::string&& pipelined_cmds, std::vector<reply_cb_t>&& callbacks)
-    {
-      if (!is_connected())
-        return false;
-
-      return
-        socket_->async_write(pipelined_cmds, [this, cbs = std::move(callbacks)](ssize_t sent_chunk_len) {
-          if (sent_chunk_len == 0)
-            return disconnect();
-
-          if (!req_queue_.size() && cbs.size())
-            do_read();
-
-          for(auto &&cb : cbs)
-            req_queue_.emplace(std::move(cb), nullptr);
-        });
-    }
-
-    bool send(const std::string&& command, const reply_cb_t& reply_cb)
-    {
-      if (!is_connected())
-        return false;
-
-      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)
-          do_read();
-      });
-    }
+    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:
-    inline
-    void do_read() {
-      socket_->async_read(data_, max_data_size, std::bind(&connection::reply_received, this, std::placeholders::_1));
-    }
-
-    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);
-
-        bool is_finished = false;
-        acc += parser::base_resp_parser::append_chunk(parser, data_ + acc, len - acc, is_finished);
-
-        if (!is_finished)
-          break;
-
-        cb(parser);
-        req_queue_.pop(); //free the resources
-      }
-
-      if (req_queue_.size())
-        do_read();
-    }
+    void do_read();
+    void reply_received(ssize_t len);
 
   private:
     std::unique_ptr<async_socket> socket_;