Rename CommandAsync->Command, cancel functionality

Renamed CommandAsync to Command to make things simpler, and refactored into its own file. command() now returns a pointer to the created Command object, which the client can pass into cancel() to stop any delayed or repeating calls. This is very important for an asynchronous client.

Rename CommandAsync->Command, cancel functionality
Renamed CommandAsync to Command to make things simpler, and refactored into its own file. command() now returns a pointer to the created Command object, which the client can pass into cancel() to stop any delayed or repeating calls. This is very important for an asynchronous client.
Hayk Martirosyan
1 parent be5fad76
Showing 5 changed files with 132 additions and 69 deletions
README.md
examples/simple_loop.cpp
src/command.hpp
src/redisx.cpp
src/redisx.hpp
@@ -2,3 +2,6 @@ redisx
 ======
 Asynchronous, minimalistic, and highly effective C++11 bindings for Redis
+Asynchronous, minimalistic, and wicked fast C++11 bindings for Redis
+
+Work in progress, details coming soon.
@@ -8,8 +8,8 @@
 using namespace std;
 double time_s() {
-  unsigned long ms = chrono::system_clock::now().time_since_epoch() / chrono::milliseconds(1);
-  return (double)ms / 1000;
+  unsigned long ms = chrono::system_clock::now().time_since_epoch() / chrono::microseconds(1);
+  return (double)ms / 1e6;
 }
 int main(int argc, char* argv[]) {
@@ -20,28 +20,31 @@ int main(int argc, char* argv[]) {
   string cmd_str = "SET alaska rules!";
   double freq = 10000; // Hz
-  double t_end = 5;
-
-  double dt = 1 / freq;
-  double t0 = time_s();
-  int count = 0;
+  double dt = 1 / freq; // s
+  double t = 5; // s
   cout << "Running \"" << cmd_str << "\" at dt = " << dt
-      << "s for " << t_end << "s..." << endl;
+      << "s for " << t << "s..." << endl;
-  rdx.command<const string &>(
+  int count = 0;
+  redisx::Command<const string&>* c = rdx.command<const string&>(
       cmd_str,
-      [&count, &rdx, t0, t_end](const string &cmd, const string &value) {
+      [&count](const string &cmd, const string &value) {
         count++;
-        if(time_s() - t0 >= t_end) rdx.stop();
       },
       dt,
       dt
   );
-  rdx.block_until_stopped();
-  double actual_freq = (double)count / t_end;
-  cout << "Sent " << count << " commands in " << t_end<< "s, "
+  double t0 = time_s();
+  this_thread::sleep_for(chrono::microseconds((int)(t*1e6)));
+  rdx.cancel<const string&>(c);
+  rdx.stop();
+
+  double t_elapsed = time_s() - t0;
+  double actual_freq = (double)count / t_elapsed;
+
+  cout << "Sent " << count << " commands in " << t_elapsed << "s, "
        << "that's " << actual_freq << " commands/s." << endl;
   return 0;
+/**
+* Redis C++11 wrapper.
+*/
+
+#pragma once
+
+#include <iostream>
+#include <string>
+#include <functional>
+
+#include <hiredis/adapters/libev.h>
+#include <hiredis/async.h>
+
+namespace redisx {
+
+template<class ReplyT>
+class Command {
+public:
+  Command(
+      redisAsyncContext* c,
+      const std::string& cmd,
+      const std::function<void(const std::string&, ReplyT)>& callback,
+      double repeat, double after
+  );
+
+  redisAsyncContext* c;
+  const std::string cmd;
+  const std::function<void(const std::string&, ReplyT)> callback;
+  double repeat;
+  double after;
+  bool done;
+  ev_timer* timer;
+  std::mutex timer_guard;
+
+  void invoke(ReplyT reply);
+
+  void free_if_done();
+  ev_timer* get_timer() {
+    std::lock_guard<std::mutex> lg(timer_guard);
+    return timer;
+  }
+};
+
+template<class ReplyT>
+Command<ReplyT>::Command(
+    redisAsyncContext* c,
+    const std::string& cmd,
+    const std::function<void(const std::string&, ReplyT)>& callback,
+    double repeat, double after
+) : c(c), cmd(cmd), callback(callback), repeat(repeat), after(after), done(false) {
+  timer_guard.lock();
+}
+
+template<class ReplyT>
+void Command<ReplyT>::invoke(ReplyT reply) {
+  if(callback != NULL) callback(cmd, reply);
+  if((repeat == 0)) done = true;
+}
+
+template<class ReplyT>
+void Command<ReplyT>::free_if_done() {
+  if(done) {
+    std::cout << "Deleting Command: " << cmd << std::endl;
+    delete this;
+  };
+}
+} // End namespace redis
@@ -13,7 +13,7 @@ namespace redisx {
 // Global mutex to manage waiting for connected state
 mutex connected_lock;
-// Map of ev_timer events to pointers to CommandAsync objects
+// Map of ev_timer events to pointers to Command objects
 // Used to get the object back from the timer watcher callback
 unordered_map<ev_timer*, void*> timer_callbacks;
@@ -99,7 +99,7 @@ void Redis::block_until_stopped() {
 * true if succeeded, false otherwise.
 */
 template<class ReplyT>
-bool submit_to_server(CommandAsync<ReplyT>* cmd_obj) {
+bool submit_to_server(Command<ReplyT>* cmd_obj) {
   if (redisAsyncCommand(cmd_obj->c, command_callback<ReplyT>, (void*)cmd_obj, cmd_obj->cmd.c_str()) != REDIS_OK) {
     cerr << "[ERROR] Async command \"" << cmd_obj->cmd << "\": " << cmd_obj->c->errstr << endl;
     cmd_obj->free_if_done();
@@ -110,7 +110,7 @@ bool submit_to_server(CommandAsync&lt;ReplyT&gt;* cmd_obj) {
 template<class ReplyT>
 void submit_command_callback(struct ev_loop* loop, ev_timer* timer, int revents) {
-  auto cmd_obj = (CommandAsync<ReplyT>*)timer_callbacks.at(timer);
+  auto cmd_obj = (Command<ReplyT>*)timer_callbacks.at(timer);
   submit_to_server<ReplyT>(cmd_obj);
 }
@@ -121,7 +121,7 @@ bool Redis::process_queued_command(void* cmd_ptr) {
   auto it = command_map.find(cmd_ptr);
   if(it == command_map.end()) return false;
-  CommandAsync<ReplyT>* cmd_obj = it->second;
+  Command<ReplyT>* cmd_obj = it->second;
   command_map.erase(cmd_ptr);
   if((cmd_obj->repeat == 0) && (cmd_obj->after == 0)) {
@@ -131,11 +131,14 @@ bool Redis::process_queued_command(void* cmd_ptr) {
     timer_callbacks[cmd_obj->timer] = (void*)cmd_obj;
     ev_timer_init(cmd_obj->timer, submit_command_callback<ReplyT>, cmd_obj->after, cmd_obj->repeat);
     ev_timer_start(EV_DEFAULT_ cmd_obj->timer);
+
+    cmd_obj->timer_guard.unlock();
   }
   return true;
 }
+
 void Redis::process_queued_commands() {
   lock_guard<mutex> lg(queue_guard);
@@ -162,15 +165,15 @@ long Redis::num_commands_processed() {
 // ----------------------------
-template<> unordered_map<void*, CommandAsync<const redisReply*>*>& Redis::get_command_map() { return commands_redis_reply; }
+template<> unordered_map<void*, Command<const redisReply*>*>& Redis::get_command_map() { return commands_redis_reply; }
 template<>
-void invoke_callback(CommandAsync<const redisReply*>* cmd_obj, redisReply* reply) {
+void invoke_callback(Command<const redisReply*>* cmd_obj, redisReply* reply) {
   cmd_obj->invoke(reply);
 }
-template<> unordered_map<void*, CommandAsync<const string&>*>& Redis::get_command_map() { return commands_string_r; }
+template<> unordered_map<void*, Command<const string&>*>& Redis::get_command_map() { return commands_string_r; }
 template<>
-void invoke_callback(CommandAsync<const string&>* cmd_obj, redisReply* reply) {
+void invoke_callback(Command<const string&>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_STRING && reply->type != REDIS_REPLY_STATUS) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-string reply." << endl;
     return;
@@ -179,9 +182,9 @@ void invoke_callback(CommandAsync&lt;const string&amp;&gt;* cmd_obj, redisReply* reply) {
   cmd_obj->invoke(reply->str);
 }
-template<> unordered_map<void*, CommandAsync<const char*>*>& Redis::get_command_map() { return commands_char_p; }
+template<> unordered_map<void*, Command<const char*>*>& Redis::get_command_map() { return commands_char_p; }
 template<>
-void invoke_callback(CommandAsync<const char*>* cmd_obj, redisReply* reply) {
+void invoke_callback(Command<const char*>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_STRING && reply->type != REDIS_REPLY_STATUS) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-string reply." << endl;
     return;
@@ -189,9 +192,9 @@ void invoke_callback(CommandAsync&lt;const char*&gt;* cmd_obj, redisReply* reply) {
   cmd_obj->invoke(reply->str);
 }
-template<> unordered_map<void*, CommandAsync<int>*>& Redis::get_command_map() { return commands_int; }
+template<> unordered_map<void*, Command<int>*>& Redis::get_command_map() { return commands_int; }
 template<>
-void invoke_callback(CommandAsync<int>* cmd_obj, redisReply* reply) {
+void invoke_callback(Command<int>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_INTEGER) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-integer reply." << endl;
     return;
@@ -199,9 +202,9 @@ void invoke_callback(CommandAsync&lt;int&gt;* cmd_obj, redisReply* reply) {
   cmd_obj->invoke((int)reply->integer);
 }
-template<> unordered_map<void*, CommandAsync<long long int>*>& Redis::get_command_map() { return commands_long_long_int; }
+template<> unordered_map<void*, Command<long long int>*>& Redis::get_command_map() { return commands_long_long_int; }
 template<>
-void invoke_callback(CommandAsync<long long int>* cmd_obj, redisReply* reply) {
+void invoke_callback(Command<long long int>* cmd_obj, redisReply* reply) {
   if(reply->type != REDIS_REPLY_INTEGER) {
     cerr << "[ERROR] " << cmd_obj->cmd << ": Received non-integer reply." << endl;
     return;
@@ -20,37 +20,9 @@
 #include <hiredis/async.h>
 #include <hiredis/adapters/libev.h>
-namespace redisx {
+#include "command.hpp"
-template<class ReplyT>
-class CommandAsync {
-public:
-  CommandAsync(
-      redisAsyncContext* c,
-      const std::string& cmd,
-      const std::function<void(const std::string&, ReplyT)>& callback,
-      double repeat, double after
-  ) : c(c), cmd(cmd), callback(callback), repeat(repeat), after(after), done(false) {}
-
-  redisAsyncContext* c;
-  const std::string cmd;
-  const std::function<void(const std::string&, ReplyT)> callback;
-  double repeat;
-  double after;
-  bool done;
-  ev_timer* timer;
-
-  void invoke(ReplyT reply) {
-    if(callback != NULL) callback(cmd, reply);
-    if((repeat == 0)) done = true;
-  }
-  void free_if_done() {
-    if(done) {
-      std::cout << "Deleting CommandAsync: " << cmd << std::endl;
-      delete this;
-    };
-  }
-};
+namespace redisx {
 class Redis {
@@ -65,7 +37,7 @@ public:
   void block_until_stopped();
   template<class ReplyT>
-  void command(
+  Command<ReplyT>* command(
       const std::string& cmd,
       const std::function<void(const std::string&, ReplyT)>& callback = NULL,
       double repeat = 0.0,
@@ -74,6 +46,9 @@ public:
   void command(const char* command);
+  template<class ReplyT>
+  bool cancel(Command<ReplyT>* cmd_obj);
+
   long num_commands_processed();
 //  void get(const char* key, std::function<void(const std::string&, const char*)> callback);
@@ -105,14 +80,14 @@ private:
   std::thread event_loop_thread;
-  std::unordered_map<void*, CommandAsync<const redisReply*>*> commands_redis_reply;
-  std::unordered_map<void*, CommandAsync<const std::string&>*> commands_string_r;
-  std::unordered_map<void*, CommandAsync<const char*>*> commands_char_p;
-  std::unordered_map<void*, CommandAsync<int>*> commands_int;
-  std::unordered_map<void*, CommandAsync<long long int>*> commands_long_long_int;
+  std::unordered_map<void*, Command<const redisReply*>*> commands_redis_reply;
+  std::unordered_map<void*, Command<const std::string&>*> commands_string_r;
+  std::unordered_map<void*, Command<const char*>*> commands_char_p;
+  std::unordered_map<void*, Command<int>*> commands_int;
+  std::unordered_map<void*, Command<long long int>*> commands_long_long_int;
   template<class ReplyT>
-  std::unordered_map<void*, CommandAsync<ReplyT>*>& get_command_map();
+  std::unordered_map<void*, Command<ReplyT>*>& get_command_map();
   std::queue<void*> command_queue;
   std::mutex queue_guard;
@@ -126,7 +101,7 @@ private:
 template<class ReplyT>
 void invoke_callback(
-    CommandAsync<ReplyT>* cmd_obj,
+    Command<ReplyT>* cmd_obj,
     redisReply* reply
 );
@@ -134,7 +109,7 @@ template&lt;class ReplyT&gt;
 void command_callback(redisAsyncContext *c, void *r, void *privdata) {
   redisReply *reply = (redisReply *) r;
-  auto *cmd_obj = (CommandAsync<ReplyT> *) privdata;
+  auto *cmd_obj = (Command<ReplyT> *) privdata;
   if (reply->type == REDIS_REPLY_ERROR) {
     std::cerr << "[ERROR] " << cmd_obj->cmd << ": " << reply->str << std::endl;
@@ -153,7 +128,7 @@ void command_callback(redisAsyncContext *c, void *r, void *privdata) {
 }
 template<class ReplyT>
-void Redis::command(
+Command<ReplyT>* Redis::command(
     const std::string& cmd,
     const std::function<void(const std::string&, ReplyT)>& callback,
     double repeat,
@@ -161,9 +136,21 @@ void Redis::command(
 ) {
   std::lock_guard<std::mutex> lg(queue_guard);
-  auto* cmd_obj = new CommandAsync<ReplyT>(c, cmd, callback, repeat, after);
+  auto* cmd_obj = new Command<ReplyT>(c, cmd, callback, repeat, after);
   get_command_map<ReplyT>()[(void*)cmd_obj] = cmd_obj;
   command_queue.push((void*)cmd_obj);
+  return cmd_obj;
+}
+
+template<class ReplyT>
+bool Redis::cancel(Command<ReplyT>* cmd_obj) {
+
+  // TODO erase from global timer_callbacks
+
+  if((cmd_obj->repeat != 0) || (cmd_obj->after != 0))
+    ev_timer_stop(EV_DEFAULT_ cmd_obj->get_timer());
+
+  return true;
 }
 } // End namespace redis