add pipelining support

Author: Cylix

README.md

@@ -1,10 +1,12 @@
 # cpp_redis
 cpp_redis is C++11 Asynchronous Redis Client.
 
-Network is based on raw sockets API. Thus, the library is really lightweight.
+Network is based on raw sockets API, making the library really lightweight.
+Commands pipelining is supported.
 
 ## Requirements
 * C++11
+* Mac or Linux (no support for Windows platforms)
 
 ## Compiling
 The library uses `cmake`. In order to build the library, follow these steps:
@@ -52,11 +54,19 @@ Throws redis_error if client is not connected to any server.
 #### bool is_connected(void)
 Returns whether the client is connected or not.
 
-#### void send(const std::vector<std::string>& redis_cmd, const reply_callback& callback = nullptr)
+#### redis_client& send(const std::vector<std::string>& redis_cmd, const reply_callback& callback = nullptr)
 Send a command and set the callback which has to be called when the reply has been received.
 If `nullptr` is passed as callback, command is executed and no callback will be called.
 Reply callback is an `std::function<void(reply&)>`.
 
+The command is not effectively sent immediately, but stored inside an internal buffer until `commit()` is called.
+
+### redis_client& commit(void)
+Send all the commands that have been stored by calling `send()` since the last `commit()` call to the redis server.
+
+That is, pipelining is supported in a very simple and efficient way: `client.send(...).send(...).send(...).commit()` will send the 3 commands at once (instead of sending 3 network requests, one for each command, as it would have been done without pipelining).
+
+
 ### Example
 
 ```cpp
@@ -88,6 +98,7 @@ main(void) {
   client.send({"GET", "hello"}, [] (cpp_redis::reply& reply) {
     std::cout << reply.as_string() << std::endl;
   });
+  client.commit();
 
   signal(SIGINT, &sigint_handler);
   while (not should_exit);
@@ -114,20 +125,33 @@ Throws redis_error if client is not connected to any server.
 #### bool is_connected(void)
 Returns whether the client is connected or not.
 
-#### void subscribe(const std::string& channel, const subscribe_callback& callback)
+#### redis_subscriber& subscribe(const std::string& channel, const subscribe_callback& callback)
 Subscribe to the given channel and call subscribe_callback each time a message is published in this channel.
 subscribe_callback is an `std::function<void(const std::string&, const std::string&)>`.
 
-#### void psubscribe(const std::string& pattern, const subscribe_callback& callback)
+The command is not effectively sent immediately, but stored inside an internal buffer until `commit()` is called.
+
+#### redis_subscriber& psubscribe(const std::string& pattern, const subscribe_callback& callback)
 PSubscribe to the given pattern and call subscribe_callback each time a message is published in a channel matching the pattern.
 subscribe_callback is an `std::function<void(const std::string&, const std::string&)>`.
 
-#### void unsubscribe(const std::string& channel)
+The command is not effectively sent immediately, but stored inside an internal buffer until `commit()` is called.
+
+#### redis_subscriber& unsubscribe(const std::string& channel)
 Unsubscribe from the given channel.
 
-#### void punsubscribe(const std::string& pattern)
+The command is not effectively sent immediately, but stored inside an internal buffer until `commit()` is called.
+
+#### redis_subscriber& punsubscribe(const std::string& pattern)
 Unsubscribe from the given pattern.
 
+The command is not effectively sent immediately, but stored inside an internal buffer until `commit()` is called.
+
+### redis_subscriber& commit(void)
+Send all the commands that have been stored by calling `send()` since the last `commit()` call to the redis server.
+
+That is, pipelining is supported in a very simple and efficient way: `sub.subscribe(...).psubscribe(...).unsubscribe(...).commit()` will send the 3 commands at once (instead of sending 3 network requests, one for each command, as it would have been done without pipelining).
+
 ### Example
 
 ```cpp
@@ -159,6 +183,7 @@ main(void) {
   sub.psubscribe("*", [] (const std::string& chan, const std::string& msg) {
     std::cout << "PMESSAGE " << chan << ": " << msg << std::endl;
   });
+  sub.commit();
 
   signal(SIGINT, &sigint_handler);
   while (not should_exit);

examples/redis_client.cpp

@@ -26,6 +26,7 @@ main(void) {
   client.send({"GET", "hello"}, [] (cpp_redis::reply& reply) {
     std::cout << reply.as_string() << std::endl;
   });
+  client.commit();
 
   signal(SIGINT, &sigint_handler);
   while (not should_exit);

examples/redis_subscriber.cpp

@@ -26,6 +26,7 @@ main(void) {
   sub.psubscribe("*", [] (const std::string& chan, const std::string& msg) {
     std::cout << "PMESSAGE " << chan << ": " << msg << std::endl;
   });
+  sub.commit();
 
   signal(SIGINT, &sigint_handler);
   while (not should_exit);

includes/cpp_redis/network/redis_connection.hpp

@@ -33,7 +33,10 @@ class redis_connection {
   bool is_connected(void);
 
   //! send cmd
-  void send(const std::vector<std::string>& redis_cmd);
+  redis_connection& send(const std::vector<std::string>& redis_cmd);
+
+  //! commit pipelined transaction
+  redis_connection& commit(void);
 
 private:
   //! receive & disconnection handlers
@@ -54,6 +57,12 @@ class redis_connection {
 
   //! reply builder
   builders::reply_builder m_builder;
+
+  //! internal buffer used for pipelining
+  std::string m_buffer;
+
+  //! protect internal buffer against race conditions
+  std::mutex m_buffer_mutex;
 };
 
 } //! network

includes/cpp_redis/redis_client.hpp

@@ -30,7 +30,10 @@ class redis_client {
 
   //! send cmd
   typedef std::function<void(reply&)> reply_callback_t;
-  void send(const std::vector<std::string>& redis_cmd, const reply_callback_t& callback = nullptr);
+  redis_client& send(const std::vector<std::string>& redis_cmd, const reply_callback_t& callback = nullptr);
+
+  //! commit pipelined transaction
+  redis_client& commit(void);
 
 private:
   //! receive & disconnection handlers

includes/cpp_redis/redis_subscriber.hpp

@@ -30,10 +30,13 @@ class redis_subscriber {
 
     //! subscribe - unsubscribe
     typedef std::function<void(const std::string&, const std::string&)> subscribe_callback_t;
-    void subscribe(const std::string& channel, const subscribe_callback_t& callback);
-    void psubscribe(const std::string& pattern, const subscribe_callback_t& callback);
-    void unsubscribe(const std::string& channel);
-    void punsubscribe(const std::string& pattern);
+    redis_subscriber& subscribe(const std::string& channel, const subscribe_callback_t& callback);
+    redis_subscriber& psubscribe(const std::string& pattern, const subscribe_callback_t& callback);
+    redis_subscriber& unsubscribe(const std::string& channel);
+    redis_subscriber& punsubscribe(const std::string& pattern);
+
+    //! commit pipelined transaction
+    redis_subscriber& commit(void);
 
 private:
     void connection_receive_handler(network::redis_connection&, reply& reply);

sources/network/redis_connection.cpp

@@ -47,9 +47,22 @@ redis_connection::build_command(const std::vector<std::string>& redis_cmd) {
   return cmd;
 }
 
-void
+redis_connection&
 redis_connection::send(const std::vector<std::string>& redis_cmd) {
-  m_client.send(build_command(redis_cmd));
+  std::lock_guard<std::mutex> lock(m_buffer_mutex);
+  m_buffer += build_command(redis_cmd);
+
+  return *this;
+}
+
+//! commit pipelined transaction
+redis_connection&
+redis_connection::commit(void) {
+  std::lock_guard<std::mutex> lock(m_buffer_mutex);
+  m_client.send(m_buffer);
+  m_buffer.clear();
+
+  return *this;
 }
 
 bool

sources/redis_client.cpp

@@ -29,12 +29,22 @@ redis_client::is_connected(void) {
   return m_client.is_connected();
 }
 
-void
+redis_client&
 redis_client::send(const std::vector<std::string>& redis_cmd, const reply_callback_t& callback) {
   m_client.send(redis_cmd);
 
   std::lock_guard<std::mutex> lock(m_callbacks_mutex);
   m_callbacks.push(callback);
+
+  return *this;
+}
+
+//! commit pipelined transaction
+redis_client&
+redis_client::commit(void) {
+  m_client.commit();
+
+  return *this;
 }
 
 void

sources/redis_subscriber.cpp

@@ -30,44 +30,59 @@ redis_subscriber::is_connected(void) {
   return m_client.is_connected();
 }
 
-void
+redis_subscriber&
 redis_subscriber::subscribe(const std::string& channel, const subscribe_callback_t& callback) {
   std::lock_guard<std::mutex> lock(m_subscribed_channels_mutex);
 
   m_subscribed_channels[channel] = callback;
   m_client.send({ "SUBSCRIBE", channel });
+
+  return *this;
 }
 
-void
+redis_subscriber&
 redis_subscriber::psubscribe(const std::string& pattern, const subscribe_callback_t& callback) {
   std::lock_guard<std::mutex> lock(m_psubscribed_channels_mutex);
 
   m_psubscribed_channels[pattern] = callback;
   m_client.send({ "PSUBSCRIBE", pattern });
+
+  return *this;
 }
 
-void
+redis_subscriber&
 redis_subscriber::unsubscribe(const std::string& channel) {
   std::lock_guard<std::mutex> lock(m_subscribed_channels_mutex);
 
   auto it = m_subscribed_channels.find(channel);
   if (it == m_subscribed_channels.end())
-    return ;
+    return *this;
 
   m_client.send({ "UNSUBSCRIBE", channel });
   m_subscribed_channels.erase(it);
+
+  return *this;
 }
 
-void
+redis_subscriber&
 redis_subscriber::punsubscribe(const std::string& pattern) {
   std::lock_guard<std::mutex> lock(m_psubscribed_channels_mutex);
 
   auto it = m_psubscribed_channels.find(pattern);
   if (it == m_psubscribed_channels.end())
-    return ;
+    return *this;
 
   m_client.send({ "PUNSUBSCRIBE", pattern });
   m_psubscribed_channels.erase(it);
+
+  return *this;
+}
+
+redis_subscriber&
+redis_subscriber::commit(void) {
+  m_client.commit();
+
+  return *this;
 }
 
 void