Make .disconnect() a combination of .wait() and .stop()

More intuitive if disconnect() blocks like connect(). Most clients will want to use these two methods. For more fine grained control, disconnect() is just a combo of stop() and wait().

Make .disconnect() a combination of .wait() and .stop()
More intuitive if disconnect() blocks like connect(). Most clients will want to use these two methods. For more fine grained control, disconnect() is just a combo of stop() and wait().
Hayk Martirosyan
1 parent ca1d3171
Showing 15 changed files with 46 additions and 25 deletions
examples/basic.cpp
examples/basic_threaded.cpp
examples/binary_data.cpp
examples/data_types.cpp
examples/lpush_benchmark.cpp
examples/multi-client.cpp
examples/pub_sub.cpp
examples/speed_test_async.cpp
examples/speed_test_async_multi.cpp
examples/speed_test_sync.cpp
src/client.cpp
src/client.hpp
src/subscriber.cpp
src/subscriber.hpp
test/test.cpp
@@ -22,5 +22,6 @@ int main(int argc, char* argv[]) {
  
   cout << "key = \"occupation\", value = \"" << rdx.get("occupation") << "\"" << endl;
  
+  rdx.disconnect();
   return 0;
 }
@@ -42,6 +42,5 @@ int main(int argc, char* argv[]) {
   getter.join();
  
   rdx.disconnect();
-  rdx.wait();
   return 0;
 };
@@ -42,5 +42,6 @@ int main(int argc, char* argv[]) {
   else cerr << "Failed to get key! Status: " << c2.status() << endl;
   c2.free();
  
+  rdx.disconnect();
   return 0;
 }
@@ -46,7 +46,7 @@ int main(int argc, char* argv[]) {
         for (const string& s : c.reply()) cout << s << " ";
         cout << endl;
       }
-      rdx.disconnect();
+      rdx.stop();
     }
   );
  
@@ -43,7 +43,7 @@ int main(int argc, char* argv[]) {
         cout << "Total time: " << t2 - t0 << "s" <<  endl;
         cout << "Result: " << (double)len / (t2-t0) << " commands/s" << endl;
  
-        rdx.disconnect();
+        rdx.stop();
       }
     });
   }
@@ -22,5 +22,9 @@ int main(int argc, char* argv[]) {
  
   cout << "key = occupation, value = \"" << rdx3.get("occupation") << "\"" << endl;
  
+  rdx1.disconnect();
+  rdx2.disconnect();
+  rdx3.disconnect();
+
   return 0;
 }
@@ -55,8 +55,6 @@ int main(int argc, char *argv[]) {
   this_thread::sleep_for(chrono::milliseconds(10));
  
   subscriber.disconnect();
-  subscriber.wait();
   publisher.disconnect();
-  publisher.wait();
   return 0;
 }
@@ -66,6 +66,5 @@ int main(int argc, char* argv[]) {
   cout << "Final value of counter: " << final_count << endl;
  
   rdx.disconnect();
-  rdx.wait();
   return 0;
 }
@@ -71,6 +71,5 @@ int main(int argc, char* argv[]) {
   cout << "Final value of counter: " << final_count << endl;
  
   rdx.disconnect();
-  rdx.wait();
   return 0;
 }
@@ -53,5 +53,6 @@ int main(int argc, char* argv[]) {
  
   cout << "Final value of counter: " << final_count << endl;
  
+  rdx.disconnect();
   return 0;
 }
@@ -56,7 +56,7 @@ void Redox::disconnectedCallback(const redisAsyncContext* ctx, int status) {
     rdx->connect_state_ = DISCONNECTED;
   }
  
-  rdx->disconnect();
+  rdx->stop();
   rdx->connect_waiter_.notify_all();
   if(rdx->user_connection_callback_) rdx->user_connection_callback_(rdx->connect_state_);
 }
@@ -208,8 +208,13 @@ bool Redox::connect() {
 }
  
 void Redox::disconnect() {
+  stop();
+  wait();
+}
+
+void Redox::stop() {
   to_exit_ = true;
-  logger_.debug() << "disconnect() called, breaking event loop";
+  logger_.debug() << "stop() called, breaking event loop";
   ev_async_send(evloop_, &watcher_stop_);
 }
  
@@ -221,7 +226,7 @@ void Redox::wait() {
 Redox::~Redox() {
  
   // Bring down the event loop
-  disconnect();
+  stop();
  
   if(event_loop_thread_.joinable()) event_loop_thread_.join();
   ev_loop_destroy(evloop_);
@@ -98,11 +98,17 @@ public:
   bool connect();
  
   /**
-  * Signal the event loop thread to disconnect from Redis and shut down.
+  * Disconnect from Redis, shut down the event loop, then return. A simple
+  * combination of .stop() and .wait().
   */
   void disconnect();
  
   /**
+  * Signal the event loop thread to disconnect from Redis and shut down.
+  */
+  void stop();
+
+  /**
   * Blocks until the event loop exits and disconnection is complete, then returns.
   * Usually no need to call manually as it is handled in the destructor.
   */
@@ -43,6 +43,15 @@ Subscriber::~Subscriber() {
  
 }
  
+void Subscriber::disconnect() {
+  stop();
+  wait();
+}
+
+void Subscriber::wait() {
+  rdx_.wait();
+}
+
 // This is a fairly awkward way of shutting down, where
 // we pause to wait for subscriptions to happen, and then
 // unsubscribe from everything and wait for that to finish.
@@ -50,7 +59,7 @@ Subscriber::~Subscriber() {
 // a segfault in freeReplyObject() under redisAsyncDisconnect()
 // if we don't do this first.
 // TODO look at hiredis, ask them what causes the error
-void Subscriber::disconnect() {
+void Subscriber::stop() {
  
   this_thread::sleep_for(chrono::milliseconds(1000));
  
@@ -63,11 +72,8 @@ void Subscriber::disconnect() {
   unique_lock<mutex> ul(cv_unsub_guard_);
   cv_unsub_.wait(ul, [this] {
     std::lock_guard<std::mutex> lg(subscribed_topics_guard_);
-    cout << "sub topic count : " << subscribed_topics_.size() << endl;
-    for(const string& topic : subscribed_topics_)
-      cout << "topic remaining: " << topic << endl;
-
-    return (subscribed_topics_.size() == 0); });
+    return (subscribed_topics_.size() == 0);
+  });
  
   unique_lock<mutex> ul2(cv_punsub_guard_);
   cv_punsub_.wait(ul, [this] {
@@ -78,7 +84,7 @@ void Subscriber::disconnect() {
   for(Command<redisReply*>* c : commands_)
     c->free();
  
-  rdx_.disconnect();
+  rdx_.stop();
 }
  
 // For debugging only
@@ -132,12 +138,10 @@ void Subscriber::subscribeBase(const string cmd_name, const string topic,
             num_pending_subs_--;
             if (sub_callback) sub_callback(topic);
           } else if (!strncmp(reply->element[0]->str, "uns", 3)) {
-            cout << "unsub from topic " << topic << endl;
             subscribed_topics_.erase(topic);
             if (unsub_callback) unsub_callback(topic);
             cv_unsub_.notify_all();
           } else if (!strncmp(reply->element[0]->str, "puns", 4)) {
-            cout << "punsub from topic " << topic << endl;
             psubscribed_topics_.erase(topic);
             if (unsub_callback) unsub_callback(topic);
             cv_punsub_.notify_all();
@@ -199,15 +203,12 @@ void Subscriber::psubscribe(const string topic,
 void Subscriber::unsubscribeBase(const string cmd_name, const string topic,
     function<void(const string&, int)> err_callback
 ) {
-  cout << "running " << cmd_name << " for " << topic << endl;
   rdx_.command<redisReply*>(cmd_name + " " + topic,
       [topic, err_callback](Command<redisReply*>& c) {
         if(!c.ok()) {
           if (err_callback) err_callback(topic, c.status());
           return;
         }
-
-        cout << "got unsub reply - " << c.cmd() << ": " << c.reply() << endl;
       }
   );
 }
@@ -60,6 +60,11 @@ public:
   bool connect() { return rdx_.connect(); }
  
   /**
+  * Same as .stop() on a Redox instance.
+  */
+  void stop();
+
+  /**
   * Same as .disconnect() on a Redox instance.
   */
   void disconnect();
@@ -67,7 +72,7 @@ public:
   /**
   * Same as .wait() on a Redox instance.
   */
-  void wait() { rdx_.wait(); }
+  void wait();
  
   /**
   * Subscribe to a topic.
@@ -100,7 +100,6 @@ protected:
     unique_lock<mutex> ul(cmd_waiter_lock);
     cmd_waiter.wait(ul, [this] { return (cmd_count == 0); });
     rdx.disconnect();
-    rdx.wait();
   };
  
   template<class ReplyT>
@@ -178,12 +177,14 @@ TEST_F(RedoxTest, Loop) {
 TEST_F(RedoxTest, GetSetSync) {
   print_and_check_sync<string>(rdx.commandSync<string>("SET redox_test:a apple"), "OK");
   print_and_check_sync<string>(rdx.commandSync<string>("GET redox_test:a"), "apple");
+  rdx.disconnect();
 }
  
 TEST_F(RedoxTest, DeleteSync) {
   print_and_check_sync<string>(rdx.commandSync<string>("SET redox_test:a apple"), "OK");
   print_and_check_sync(rdx.commandSync<int>("DEL redox_test:a"), 1);
   check_sync(rdx.commandSync<nullptr_t>("GET redox_test:a"), nullptr);
+  rdx.disconnect();
 }
  
 TEST_F(RedoxTest, IncrSync) {
@@ -192,6 +193,7 @@ TEST_F(RedoxTest, IncrSync) {
     check_sync(rdx.commandSync<int>("INCR redox_test:a"), i+1);
   }
   print_and_check_sync(rdx.commandSync<string>("GET redox_test:a"), to_string(count));
+  rdx.disconnect();
 }
  
 // -------------------------------------------