add ability to setup its own network module and to use it in place of tacopie

Author: Cylix

includes/cpp_redis/network/redis_connection.hpp

@@ -23,13 +23,13 @@
 #pragma once
 
 #include <functional>
+#include <memory>
 #include <mutex>
 #include <string>
 #include <vector>
 
 #include <cpp_redis/builders/reply_builder.hpp>
-
-#include <tacopie/tacopie>
+#include <cpp_redis/network/tcp_client_iface.hpp>
 
 #ifndef __CPP_REDIS_READ_SIZE
 #define __CPP_REDIS_READ_SIZE 4096
@@ -67,7 +67,7 @@ class redis_connection {
 
 private:
   //! receive & disconnection handlers
-  void tcp_client_receive_handler(const tacopie::tcp_client::read_result& result);
+  void tcp_client_receive_handler(const tcp_client_iface::read_result& result);
   void tcp_client_disconnection_handler(void);
 
   std::string build_command(const std::vector<std::string>& redis_cmd);
@@ -77,7 +77,7 @@ class redis_connection {
 
 private:
   //! tcp client for redis connection
-  tacopie::tcp_client m_client;
+  std::shared_ptr<cpp_redis::network::tcp_client_iface> m_client;
 
   //! reply callback
   reply_callback_t m_reply_callback;
@@ -95,6 +95,8 @@ class redis_connection {
   std::mutex m_buffer_mutex;
 };
 
+extern std::function<std::shared_ptr<tcp_client_iface>()> get_tcp_client;
+
 } //! network
 
 } //! cpp_redis

includes/cpp_redis/network/tcp_client.hpp

@@ -0,0 +1,102 @@
+// MIT License
+//
+// Copyright (c) 2016-2017 Simon Ninon <simon.ninon@gmail.com>
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+#pragma once
+
+#include <cpp_redis/network/tcp_client_iface.hpp>
+#include <cpp_redis/redis_error.hpp>
+
+#include <tacopie/tacopie>
+
+namespace cpp_redis {
+
+namespace network {
+
+class tcp_client : public tcp_client_iface {
+public:
+  //! ctor & dtor
+  tcp_client(void)  = default;
+  ~tcp_client(void) = default;
+
+public:
+  //! start & stop the tcp client
+  void
+  connect(const std::string& addr, std::uint32_t port) {
+    m_client.connect(addr, port);
+  }
+
+  void
+  disconnect(bool wait_for_removal = false) {
+    m_client.disconnect(wait_for_removal);
+  }
+
+  //! returns whether the client is currently connected or not
+  bool
+  is_connected(void) const {
+    return m_client.is_connected();
+  }
+
+public:
+  //! async read & write operations
+  void
+  async_read(read_request& request) {
+    auto callback = std::move(request.async_read_callback);
+
+    m_client.async_read({request.size, [=](tacopie::tcp_client::read_result& result) {
+                           if (not callback) {
+                             return;
+                           }
+
+                           read_result converted_result = {result.success, std::move(result.buffer)};
+                           callback(converted_result);
+                         }});
+  }
+
+  void
+  async_write(write_request& request) {
+    auto callback = std::move(request.async_write_callback);
+
+    m_client.async_write({std::move(request.buffer), [=](tacopie::tcp_client::write_result& result) {
+                            if (not callback) {
+                              return;
+                            }
+
+                            write_result converted_result = {result.success, result.size};
+                            callback(converted_result);
+                          }});
+  }
+
+public:
+  //! set on disconnection handler
+  void
+  set_on_disconnection_handler(const disconnection_handler_t& disconnection_handler) {
+    m_client.set_on_disconnection_handler(disconnection_handler);
+  }
+
+private:
+  //! tcp client for redis connection
+  tacopie::tcp_client m_client;
+};
+
+} //! network
+
+} //! cpp_redis

includes/cpp_redis/network/tcp_client_iface.hpp

@@ -0,0 +1,89 @@
+// MIT License
+//
+// Copyright (c) 2016-2017 Simon Ninon <simon.ninon@gmail.com>
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+#pragma once
+
+namespace cpp_redis {
+
+namespace network {
+
+class tcp_client_iface {
+public:
+  //! ctor & dtor
+  tcp_client_iface(void)          = default;
+  virtual ~tcp_client_iface(void) = default;
+
+public:
+  //! start & stop the tcp client
+  virtual void connect(const std::string& addr, std::uint32_t port) = 0;
+  virtual void disconnect(bool wait_for_removal = false) = 0;
+
+  //! returns whether the client is currently connected or not
+  virtual bool is_connected(void) const = 0;
+
+public:
+  //! structure to store read requests result
+  struct read_result {
+    bool success;
+    std::vector<char> buffer;
+  };
+
+  //! structure to store write requests result
+  struct write_result {
+    bool success;
+    std::size_t size;
+  };
+
+public:
+  //! async read & write completion callbacks
+  typedef std::function<void(read_result&)> async_read_callback_t;
+  typedef std::function<void(write_result&)> async_write_callback_t;
+
+public:
+  //! structure to store read requests information
+  struct read_request {
+    std::size_t size;
+    async_read_callback_t async_read_callback;
+  };
+
+  //! structure to store write requests information
+  struct write_request {
+    std::vector<char> buffer;
+    async_write_callback_t async_write_callback;
+  };
+
+public:
+  //! async read & write operations
+  virtual void async_read(read_request& request)   = 0;
+  virtual void async_write(write_request& request) = 0;
+
+public:
+  //! disconnection handle
+  typedef std::function<void()> disconnection_handler_t;
+
+  //! set on disconnection handler
+  virtual void set_on_disconnection_handler(const disconnection_handler_t& disconnection_handler) = 0;
+};
+
+} //! network
+
+} //! cpp_redis

includes/cpp_redis/redis_client.hpp

@@ -22,6 +22,7 @@
 
 #pragma once
 
+#include <atomic>
 #include <condition_variable>
 #include <functional>
 #include <mutex>

sources/network/redis_connection.cpp

@@ -22,20 +22,28 @@
 
 #include <cpp_redis/logger.hpp>
 #include <cpp_redis/network/redis_connection.hpp>
+#include <cpp_redis/network/tcp_client.hpp>
 #include <cpp_redis/redis_error.hpp>
 
+#include <tacopie/tacopie>
+
 namespace cpp_redis {
 
 namespace network {
 
+std::function<std::shared_ptr<tcp_client_iface>()> get_tcp_client = []() -> std::shared_ptr<tcp_client_iface> {
+  return std::make_shared<tcp_client>();
+};
+
 redis_connection::redis_connection(void)
-: m_reply_callback(nullptr)
+: m_client(get_tcp_client())
+, m_reply_callback(nullptr)
 , m_disconnection_handler(nullptr) {
   __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection created");
 }
 
 redis_connection::~redis_connection(void) {
-  m_client.disconnect(true);
+  m_client->disconnect(true);
   __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection destroyed");
 }
 
@@ -46,14 +54,17 @@ redis_connection::connect(const std::string& host, std::size_t port,
   try {
     __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection attempts to connect");
 
-    //! connect client and start to read asynchronously
-    m_client.connect(host, port);
-    m_client.async_read({__CPP_REDIS_READ_SIZE, std::bind(&redis_connection::tcp_client_receive_handler, this, std::placeholders::_1)});
-    m_client.set_on_disconnection_handler(std::bind(&redis_connection::tcp_client_disconnection_handler, this));
+    //! connect client
+    m_client->connect(host, port);
+    m_client->set_on_disconnection_handler(std::bind(&redis_connection::tcp_client_disconnection_handler, this));
+
+    //! start to read asynchronously
+    tcp_client_iface::read_request request = {__CPP_REDIS_READ_SIZE, std::bind(&redis_connection::tcp_client_receive_handler, this, std::placeholders::_1)};
+    m_client->async_read(request);
 
     __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection connected");
   }
-  catch (const tacopie::tacopie_error& e) {
+  catch (const std::exception& e) {
     __CPP_REDIS_LOG(error, std::string("cpp_redis::network::redis_connection ") + e.what());
     throw redis_error(e.what());
   }
@@ -65,13 +76,13 @@ redis_connection::connect(const std::string& host, std::size_t port,
 void
 redis_connection::disconnect(bool wait_for_removal) {
   __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection attempts to disconnect");
-  m_client.disconnect(wait_for_removal);
+  m_client->disconnect(wait_for_removal);
   __CPP_REDIS_LOG(debug, "cpp_redis::network::redis_connection disconnected");
 }
 
 bool
 redis_connection::is_connected(void) {
-  return m_client.is_connected();
+  return m_client->is_connected();
 }
 
 std::string
@@ -104,9 +115,10 @@ redis_connection::commit(void) {
   std::string buffer = std::move(m_buffer);
 
   try {
-    m_client.async_write({std::vector<char>{buffer.begin(), buffer.end()}, nullptr});
+    tcp_client_iface::write_request request = {std::vector<char>{buffer.begin(), buffer.end()}, nullptr};
+    m_client->async_write(request);
   }
-  catch (const tacopie::tacopie_error& e) {
+  catch (const std::exception& e) {
     __CPP_REDIS_LOG(error, std::string("cpp_redis::network::redis_connection ") + e.what());
     throw redis_error(e.what());
   }
@@ -125,7 +137,7 @@ redis_connection::call_disconnection_handler(void) {
 }
 
 void
-redis_connection::tcp_client_receive_handler(const tacopie::tcp_client::read_result& result) {
+redis_connection::tcp_client_receive_handler(const tcp_client_iface::read_result& result) {
   if (!result.success) { return; }
 
   try {
@@ -151,9 +163,10 @@ redis_connection::tcp_client_receive_handler(const tacopie::tcp_client::read_res
   }
 
   try {
-    m_client.async_read({__CPP_REDIS_READ_SIZE, std::bind(&redis_connection::tcp_client_receive_handler, this, std::placeholders::_1)});
+    tcp_client_iface::read_request request = {__CPP_REDIS_READ_SIZE, std::bind(&redis_connection::tcp_client_receive_handler, this, std::placeholders::_1)};
+    m_client->async_read(request);
   }
-  catch (const tacopie::tacopie_error&) {
+  catch (const std::exception&) {
     //! Client disconnected in the meantime
   }
 }

tacopie

@@ -20,8 +20,11 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 // SOFTWARE.
 
+#include <thread>
+
 #include <cpp_redis/redis_client.hpp>
 #include <cpp_redis/redis_error.hpp>
+
 #include <gtest/gtest.h>
 
 TEST(RedisClient, ValidConnectionDefaultParams) {
@@ -46,7 +49,7 @@ TEST(RedisClient, InvalidConnection) {
   cpp_redis::redis_client client;
 
   EXPECT_FALSE(client.is_connected());
-  EXPECT_THROW(client.connect("invalid.url", 1234), cpp_redis::redis_error);
+  EXPECT_THROW(client.connect("invalid url", 1234), cpp_redis::redis_error);
   EXPECT_FALSE(client.is_connected());
 }
 

tests/sources/spec/redis_client_spec.cpp

@@ -20,9 +20,12 @@
 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 // SOFTWARE.
 
+#include <thread>
+
 #include <cpp_redis/redis_client.hpp>
 #include <cpp_redis/redis_error.hpp>
 #include <cpp_redis/redis_subscriber.hpp>
+
 #include <gtest/gtest.h>
 
 TEST(RedisSubscriber, ValidConnectionDefaultParams) {
@@ -47,7 +50,7 @@ TEST(RedisSubscriber, InvalidConnection) {
   cpp_redis::redis_subscriber client;
 
   EXPECT_FALSE(client.is_connected());
-  EXPECT_THROW(client.connect("invalid.url", 1234), cpp_redis::redis_error);
+  EXPECT_THROW(client.connect("invalid url", 1234), cpp_redis::redis_error);
   EXPECT_FALSE(client.is_connected());
 }