Add asio support #19

.gitignore

@@ -35,3 +35,4 @@
 
 # build directory
 build/
+.vscode/

.gitmodules

@@ -1,3 +1,3 @@
-[submodule "libs/libevpp"]
-	path = libs/libevpp
-	url = https://github.com/hamidr/libevpp
+[submodule "libs/asio"]
+	path = libs/asio
+	url = https://github.com/chriskohlhoff/asio

CMakeLists.txt

@@ -15,8 +15,9 @@ else()
   SET( CMAKE_EXE_LINKER_FLAGS  "${CMAKE_EXE_LINKER_FLAGS} ${GCC_COVERAGE_LINK_FLAGS} -g" )
 endif()
 
-set(CMAKE_CXX_STANDARD 14)
+set(CMAKE_CXX_STANDARD 17)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
+add_definitions(-DASIO_STANDALONE)
 
 set(PROJECT_SOURCE_DIR src)
 set(PROJECT_INCLUDE_DIR includes)
@@ -25,13 +26,7 @@ include_directories(${PROJECT_INCLUDE_DIR})
 include_directories(/usr/include/)
 include_directories(/usr/local/include)
 
-link_directories(/usr/lib)
-link_directories(/usr/local/lib)
-
-add_subdirectory(libs/libevpp/)
-
-include_directories(libs/libevpp/includes)
-
+include_directories(libs/asio/asio/include/)
 
 add_library(parser
   ${PROJECT_SOURCE_DIR}/parser/base_resp_parser.cpp
@@ -54,7 +49,20 @@ if(CMAKE_COMPILER_IS_GNUCXX)
     set(CMAKE_EXE_LINKER_FLAGS "-s")  ## Strip binary
 endif()
 
-target_link_libraries(async_redis network parser)
+target_link_libraries(async_redis parser pthread)
+
+install(TARGETS async_redis
+  LIBRARY 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 (a2.out test/main.cpp)
 target_link_libraries(a2.out async_redis)

build.sh

@@ -0,0 +1,6 @@
+git submodule init
+git submodule update 
+mkdir -p build && cd build
+cmake ..
+make
+cd ..

includes/async_redis/connection.hpp

@@ -2,29 +2,27 @@
 
 #include <queue>
 #include <functional>
-#include <memory>
 #include <tuple>
 
+#include <asio/io_context.hpp>
+#include <asio/ip/tcp.hpp>
 #include <async_redis/parser/base_resp_parser.h>
-#include <libevpp/event_loop/event_loop_ev.h>
-#include <libevpp/network/async_socket.hpp>
-
-using namespace libevpp;
 
 namespace async_redis
 {
   class connection
   {
-    using async_socket    = libevpp::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();
@@ -33,14 +31,14 @@ 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];
+    bool is_connected_ = false;
   };
 }

includes/async_redis/monitor.hpp

@@ -1,20 +1,24 @@
 #pragma once
 
-#include <libevpp/network/async_socket.hpp>
 #include <async_redis/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;
-using namespace libevpp;
 
 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 {
@@ -27,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;
@@ -42,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);
@@ -52,18 +55,18 @@ 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;
+    bool is_connected_ = false;
   };
 
 }

includes/async_redis/parser/base_resp_parser.h

@@ -30,11 +30,11 @@ namespace async_redis {
       virtual int parse_append(const char*, ssize_t, bool&) = 0;
       virtual std::string to_string() const = 0;
       virtual void map(const caller_t &fn);
-      bool is_array() const;
-      bool is_number() const;
-      bool is_string() const;
-      bool is_enum() const;
-      bool is_error() const;
+      bool is_array() const noexcept;
+      bool is_number() const noexcept;
+      bool is_string() const noexcept;
+      bool is_enum() const noexcept;
+      bool is_error() const noexcept;
 
       void print();
     };

includes/async_redis/redis_client.hpp

@@ -12,20 +12,21 @@ 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);
-    bool is_connected() const;
+    redis_client(asio::io_context &io) noexcept;
+    bool is_connected() const noexcept;
 
     template <typename ...Args>
     void connect(const connect_cb_t& handler, Args... args) {
-      for(auto &conn : conn_pool_)
-        conn->connect(std::bind(&redis_client::check_conn_connected, this, handler, std::placeholders::_1), args...);
+        conn_.connect(handler, args...);
     }
 
     void disconnect();
@@ -39,27 +40,13 @@ namespace async_redis
     void ping(reply_cb_t reply);
     void publish(const string& channel, const string& msg, reply_cb_t&& reply);
     void sort(const string& hash_name, std::vector<string>&& fields, reply_cb_t&& reply);
-
-    //TODO: wtf?! doesnt make sense with multiple connections!
-    // void select(uint catalog, reply_cb_t&& reply) {
-    //   send({"select", std::to_string(catalog)}, reply);
-    // }
-
-    void commit_pipeline();
-    redis_client& pipeline_on();
-    redis_client& pipeline_off();
+    void select(uint catalog, reply_cb_t&& reply);
 
   private:
-    void send(const std::vector<string>&& elems, const reply_cb_t& reply);
-    connection& get_connection();
-    void check_conn_connected(const connect_cb_t& handler, bool res);
+    bool send(std::vector<string>&& elems, const reply_cb_t& reply) noexcept;
 
   private:
-    std::string pipeline_buffer_;
-    bool pipelined_state_ = false;
-    std::vector<reply_cb_t> pipelined_cbs_;
-    std::vector<std::unique_ptr<connection>> conn_pool_;
-    int con_rr_ctr_ = 0;
+    connection conn_;
     int connected_called_ = 0;
     bool is_connected_ = false;
   };

includes/async_redis/sentinel.hpp

@@ -1,18 +1,14 @@
 #pragma once
-#include <libevpp/network/async_socket.hpp>
 #include <async_redis/parser/base_resp_parser.h>
 
-#include <async_redis/monitor.hpp>
 #include <functional>
+#include <async_redis/monitor.hpp>
 #include <async_redis/connection.hpp>
 
-using namespace libevpp;
-
 namespace async_redis {
   class sentinel
   {
-    using socket_t     = libevpp::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;
@@ -22,7 +18,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);