Add asio support #19

Author: hamidr

CMakeLists.txt

@@ -23,21 +23,12 @@ set(PROJECT_INCLUDE_DIR ${CMAKE_SOURCE_DIR}/includes)
 
 include_directories(${PROJECT_INCLUDE_DIR})
 include_directories("/usr/include/")
-include_directories("/usr/local/include")
+include_directories(${CMAKE_SOURCE_DIR}/libs/asio/asio/include/)
 
 link_directories("/usr/lib")
 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
-  ${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
@@ -59,18 +50,19 @@ if(CMAKE_COMPILER_IS_GNUCXX)
     set(CMAKE_EXE_LINKER_FLAGS "-s")  ## Strip binary
 endif()
 
-target_link_libraries(event_loop ev)
-target_link_libraries(async_redis event_loop parser network)
+target_link_libraries(async_redis parser pthread)
 
-install(TARGETS event_loop
+install(TARGETS async_redis
   LIBRARY DESTINATION /usr/local/lib/
-  ARCHIVE DESTINATION /usr/local/lib/)
+  ARCHIVE DESTINATION /usr/local/lib/
+  )
 
 install(TARGETS parser
   LIBRARY DESTINATION /usr/local/lib/
   ARCHIVE DESTINATION /usr/local/lib/
   )
 
+
 install(DIRECTORY ${PROJECT_INCLUDE_DIR}/ DESTINATION /usr/local/include)
 
 add_executable (a1.out ${CMAKE_SOURCE_DIR}/test/main.cpp)

includes/connection.hpp

@@ -2,26 +2,28 @@
 
 #include <queue>
 #include <functional>
-#include <memory>
 #include <tuple>
 
+#include <asio/io_context.hpp>
+#include <asio/ip/tcp.hpp>
+
 #include <parser/base_resp_parser.h>
-#include <network/async_socket.hpp>
 
 namespace async_redis
 {
   class connection
   {
-    using async_socket    = network::async_socket;
+    connection(const connection&) = delete;
+    connection& operator = (const connection&) = delete;
 
   public:
+    using connect_handler_t    = std::function<void(bool)>;
     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);
+    connection(asio::io_context&);
 
-    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);
+    void connect(connect_handler_t handler, const std::string& ip, int port);
 
     bool is_connected() const;
     void disconnect();
@@ -30,13 +32,12 @@ namespace async_redis
 
   private:
     void do_read();
-    void reply_received(ssize_t len);
+    void reply_received(const asio::error_code& ec, size_t len);
 
   private:
-    std::unique_ptr<async_socket> socket_;
+    asio::ip::tcp::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/monitor.hpp

@@ -1,19 +1,24 @@
 #pragma once
 
-#include <network/async_socket.hpp>
 #include <parser/base_resp_parser.h>
 
 #include <unordered_map>
 #include <list>
 #include <string>
 
+#include <asio/io_context.hpp>
+#include <asio/ip/tcp.hpp>
+
 using std::string;
 
 namespace async_redis
 {
   class monitor
   {
-    using async_socket    = network::async_socket;
+    monitor(const monitor&) = delete;
+    monitor& operator = (const monitor&) = delete;
+
+    using connect_handler_t = std::function<void(bool)>;
 
   public:
     enum EventState {
@@ -26,10 +31,8 @@ namespace async_redis
     using parser_t     = parser::base_resp_parser::parser;
     using watcher_cb_t = std::function<void (const string&, parser_t, EventState)>;
 
-    monitor(event_loop::event_loop_ev &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);
+    monitor(asio::io_context &event_loop);
+    void connect(connect_handler_t handler, const std::string& ip, int port);
 
     bool is_connected() const;
     bool is_watching() const;
@@ -41,6 +44,7 @@ namespace async_redis
     bool punsubscribe(const std::list<string>& channels, watcher_cb_t&& cb);
 
   private:
+    void do_read();
     bool send_and_receive(string&& data);
     void handle_message_event(parser_t& channel, parser_t& value);
     void handle_subscribe_event(parser_t& channel, parser_t& clients);
@@ -51,15 +55,14 @@ namespace async_redis
     void handle_event(parser_t&& request);
     void report_disconnect();
 
-    void stream_received(ssize_t len);
+    void stream_received(const asio::error_code& ec, size_t len);
 
   private:
     parser_t parser_;
     std::unordered_map<std::string, watcher_cb_t> watchers_;
     std::unordered_map<std::string, watcher_cb_t> pwatchers_;
 
-    std::unique_ptr<async_socket> socket_;
-    event_loop::event_loop_ev &io_;
+    asio::ip::tcp::socket socket_;
     enum { max_data_size = 1024 };
     char data_[max_data_size];
     bool is_watching_ = false;

includes/network/async_socket.hpp

@@ -12,14 +12,17 @@ namespace async_redis
 
   class redis_client
   {
+    redis_client(const redis_client&) = delete;
+    redis_client& operator = (const redis_client&) = delete;
+
     using reply_cb_t   = connection::reply_cb_t;
-    using connect_cb_t = network::async_socket::connect_handler_t;
+    using connect_cb_t = connection::connect_handler_t;
 
   public:
     class connect_exception : std::exception {};
     using parser_t = connection::parser_t;
 
-    redis_client(event_loop::event_loop_ev &eventIO, int n = 1);
+    redis_client(asio::io_context &io, uint n = 1);
     bool is_connected() const;
 
     template <typename ...Args>

includes/network/tcp_socket.hpp

@@ -1,17 +1,15 @@
 #pragma once
-#include <network/async_socket.hpp>
 #include <parser/base_resp_parser.h>
 
-#include <monitor.hpp>
 #include <functional>
+
+#include <monitor.hpp>
 #include <connection.hpp>
-// #include <memory>
 
 namespace async_redis {
   class sentinel
   {
-    using socket_t     = ::async_redis::network::async_socket;
-    using connect_cb_t = socket_t::connect_handler_t;
+    using connect_cb_t = connection::connect_handler_t;
 
   public:
     using parser_t     = parser::base_resp_parser::parser;
@@ -21,7 +19,7 @@ namespace async_redis {
       Watching
     };
 
-    sentinel(event_loop::event_loop_ev &event_loop);
+    sentinel(asio::io_context &io);
 
     bool is_connected() const;
     bool connect(const string& ip, int port, connect_cb_t&& connector);
@@ -54,7 +52,7 @@ namespace async_redis {
 
   private:
     int connected_ = 0;
-    std::unique_ptr<monitor> stream_;
-    std::unique_ptr<connection> conn_;
+    monitor stream_;
+    connection conn_;
   };
 }