Improved connect/disconnect management

Consolidated user callbacks into one, improved some logic.

Improved connect/disconnect management
Consolidated user callbacks into one, improved some logic.
Hayk Martirosyan
1 parent b5d5ea9a
Showing 11 changed files with 39 additions and 45 deletions
CMakeLists.txt
examples/basic.cpp
examples/basic_threaded.cpp
examples/data_types.cpp
examples/lpush_benchmark.cpp
examples/speed_test_async.cpp
examples/speed_test_async_multi.cpp
examples/speed_test_sync.cpp
examples/string_vs_charp.cpp
src/redox.cpp
src/redox.hpp
@@ -32,11 +32,11 @@ set(LIB_ALL ${LIB_REDIS})
 add_executable(basic examples/basic.cpp ${SRC_ALL})
 target_link_libraries(basic ${LIB_REDIS})
-#add_executable(basic_threaded examples/basic_threaded.cpp ${SRC_ALL})
-#target_link_libraries(basic_threaded ${LIB_REDIS})
+add_executable(basic_threaded examples/basic_threaded.cpp ${SRC_ALL})
+target_link_libraries(basic_threaded ${LIB_REDIS})
-#add_executable(lpush_benchmark examples/lpush_benchmark.cpp ${SRC_ALL})
-#target_link_libraries(lpush_benchmark ${LIB_REDIS})
+add_executable(lpush_benchmark examples/lpush_benchmark.cpp ${SRC_ALL})
+target_link_libraries(lpush_benchmark ${LIB_REDIS})
 add_executable(speed_test_async examples/speed_test_async.cpp ${SRC_ALL})
 target_link_libraries(speed_test_async ${LIB_REDIS})
@@ -44,14 +44,14 @@ target_link_libraries(speed_test_async ${LIB_REDIS})
 add_executable(speed_test_sync examples/speed_test_sync.cpp ${SRC_ALL})
 target_link_libraries(speed_test_sync ${LIB_REDIS})
-#add_executable(speed_test_async_multi examples/speed_test_async_multi.cpp ${SRC_ALL})
-#target_link_libraries(speed_test_async_multi ${LIB_REDIS})
+add_executable(speed_test_async_multi examples/speed_test_async_multi.cpp ${SRC_ALL})
+target_link_libraries(speed_test_async_multi ${LIB_REDIS})
-#add_executable(data_types examples/data_types.cpp ${SRC_ALL})
-#target_link_libraries(data_types ${LIB_REDIS})
+add_executable(data_types examples/data_types.cpp ${SRC_ALL})
+target_link_libraries(data_types ${LIB_REDIS})
-#add_executable(string_v_char examples/string_vs_charp.cpp ${SRC_ALL})
-#target_link_libraries(string_v_char ${LIB_REDIS})
+add_executable(string_v_char examples/string_vs_charp.cpp ${SRC_ALL})
+target_link_libraries(string_v_char ${LIB_REDIS})
 add_executable(multi_client examples/multi-client.cpp ${SRC_ALL})
 target_link_libraries(multi_client ${LIB_REDIS})
@@ -9,7 +9,7 @@ using namespace std;
 int main(int argc, char* argv[]) {
-  redox::Redox rdx = {"localhost", 6379}; // Initialize Redox
+  redox::Redox rdx = {"localhost", 6379, [](int state) { cout << "Connection state: " << state << endl; }}; // Initialize Redox
   if(!rdx.start()) return 1; // Start the event loop
   rdx.del("occupation");
@@ -13,7 +13,7 @@ redox::Redox rdx = {&quot;localhost&quot;, 6379};
 int main(int argc, char* argv[]) {
-  rdx.start();
+  if(!rdx.start()) return 1;
   thread setter([]() {
     for(int i = 0; i < 5000; i++) {
@@ -10,7 +10,7 @@ using namespace std;
 int main(int argc, char* argv[]) {
   redox::Redox rdx; // Initialize Redox (default host/port)
-  rdx.start(); // Start the event loop
+  if(!rdx.start()) return 1; // Start the event loop
   rdx.del("mylist");
@@ -15,7 +15,7 @@ double time_s() {
 int main(int argc, char* argv[]) {
   redox::Redox rdx;
-  rdx.start();
+  if(!rdx.start()) return 1;
   rdx.del("test");
@@ -18,7 +18,7 @@ double time_s() {
 int main(int argc, char* argv[]) {
   Redox rdx = {"localhost", 6379};
-  rdx.start();
+  if(!rdx.start()) return 1;
   if(rdx.command_blocking("SET simple_loop:count 0")) {
     cout << "Reset the counter to zero." << endl;
@@ -19,7 +19,7 @@ double time_s() {
 int main(int argc, char* argv[]) {
   Redox rdx = {"localhost", 6379};
-  rdx.start();
+  if(!rdx.start()) return 1;
   if(rdx.set("simple_loop:count", "0")) {
     cout << "Reset the counter to zero." << endl;
@@ -18,7 +18,7 @@ double time_s() {
 int main(int argc, char* argv[]) {
   Redox rdx = {"localhost", 6379};
-  rdx.start();
+  if(!rdx.start()) return 1;
   if(rdx.command_blocking("SET simple_loop:count 0")) {
     cout << "Reset the counter to zero." << endl;
@@ -18,7 +18,7 @@ double time_s() {
 int main(int argc, char* argv[]) {
   Redox rdx;
-  rdx.start();
+  if(!rdx.start()) return 1;
   rdx.del("stringtest");
   rdx.set("stringtest", "value");
@@ -16,19 +16,16 @@ void Redox::connected_callback(const redisAsyncContext *ctx, int status) {
   if (status != REDIS_OK) {
     cerr << "[ERROR] Connecting to Redis: " << ctx->errstr << endl;
     rdx->connect_state = REDOX_CONNECT_ERROR;
-    rdx->connect_waiter.notify_all();
-    return;
-  }
-  // Disable hiredis automatically freeing reply objects
-  ctx->c.reader->fn->freeObject = [](void* reply) {};
+  } else {
+    // Disable hiredis automatically freeing reply objects
+    ctx->c.reader->fn->freeObject = [](void *reply) {};
+    rdx->connect_state = REDOX_CONNECTED;
+    cout << "[INFO] Connected to Redis." << endl;
+  }
-  rdx->connect_state = REDOX_CONNECTED;
   rdx->connect_waiter.notify_all();
-
-  cout << "[INFO] Connected to Redis." << endl;
-
-  if(rdx->user_connect_callback) rdx->user_connect_callback();
+  if(rdx->user_connection_callback) rdx->user_connection_callback(rdx->connect_state);
 }
 void Redox::disconnected_callback(const redisAsyncContext *ctx, int status) {
@@ -45,37 +42,36 @@ void Redox::disconnected_callback(const redisAsyncContext *ctx, int status) {
   rdx->stop_signal();
   rdx->connect_waiter.notify_all();
-
-  if(rdx->user_disconnect_callback) rdx->user_disconnect_callback();
+  if(rdx->user_connection_callback) rdx->user_connection_callback(rdx->connect_state);
 }
 Redox::Redox(
   const string& host, const int port,
-  std::function<void(void)> connected,
-  std::function<void(void)> disconnected
-) : host(host), port(port), user_connect_callback(connected), user_disconnect_callback(disconnected) {
+  std::function<void(int)> connection_callback
+) : host(host), port(port), user_connection_callback(connection_callback) {
-  // Required by libev
+  // libev setup
   signal(SIGPIPE, SIG_IGN);
+  evloop = ev_loop_new(EVFLAG_AUTO);
+  ev_set_userdata(evloop, (void*)this); // Back-reference
   // Create a redisAsyncContext
   ctx = redisAsyncConnect(host.c_str(), port);
+  ctx->data = (void*)this; // Back-reference
+
   if (ctx->err) {
-    printf("Error: %s\n", ctx->errstr);
+    cout << "[ERROR] Could not create a hiredis context: " << ctx->errstr << endl;
+    connect_state = REDOX_CONNECT_ERROR;
+    connect_waiter.notify_all();
     return;
   }
-  // Create a new event loop and attach it to hiredis
-  evloop = ev_loop_new(EVFLAG_AUTO);
+  // Attach event loop to hiredis
   redisLibevAttach(evloop, ctx);
   // Set the callbacks to be invoked on server connection/disconnection
   redisAsyncSetConnectCallback(ctx, Redox::connected_callback);
   redisAsyncSetDisconnectCallback(ctx, Redox::disconnected_callback);
-
-  // Set back references to this Redox object (for use in callbacks)
-  ev_set_userdata(evloop, (void*)this);
-  ctx->data = (void*)this;
 }
 void Redox::run_event_loop() {
@@ -46,8 +46,7 @@ public:
   Redox(
     const std::string& host = REDIS_DEFAULT_HOST,
     const int port = REDIS_DEFAULT_PORT,
-    std::function<void(void)> connected = nullptr,
-    std::function<void(void)> disconnected = nullptr
+    std::function<void(int)> connection_callback = nullptr
   );
   ~Redox();
@@ -184,8 +183,7 @@ private:
   std::condition_variable connect_waiter;
   // User connect/disconnect callbacks
-  std::function<void(void)> user_connect_callback;
-  std::function<void(void)> user_disconnect_callback;
+  std::function<void(int)> user_connection_callback;
   // Dynamically allocated libev event loop
   struct ev_loop* evloop;