From b0ba54e4c0590e659e9a9024ba6f19ae5ca77d1a Mon Sep 17 00:00:00 2001
From: Stephanie Wang <swang93@mit.edu>
Date: Tue, 13 Dec 2016 17:48:09 -0800
Subject: [PATCH] Fix psubscribe bug in object_table_subscribe (#126)

* Fix psubscribe

* Add TODO about subscription callbacks
---
 src/common/state/redis.c             | 12 ++++++++----
 src/common/test/object_table_tests.c |  2 +-
 2 files changed, 9 insertions(+), 5 deletions(-)

diff --git a/src/common/state/redis.c b/src/common/state/redis.c
index 941530e60..cd59223ee 100644
--- a/src/common/state/redis.c
+++ b/src/common/state/redis.c
@@ -604,6 +604,7 @@ void redis_object_table_get_entry(redisAsyncContext *c,
   REDIS_CALLBACK_HEADER(db, callback_data, r);
   redisReply *reply = r;
 
+  LOG_DEBUG("Object table get entry callback");
   db_client_id *managers = malloc(reply->elements * sizeof(db_client_id));
   int64_t manager_count = reply->elements;
 
@@ -630,6 +631,8 @@ void redis_object_table_get_entry(redisAsyncContext *c,
       object_table_object_available_callback sub_callback =
           sub_data->object_available_callback;
       if (manager_count > 0) {
+        /* TODO(swang): For global scheduler subscriptions, we should be
+         * calling this even when the manager_count is 0. */
         if (sub_callback) {
           sub_callback(id, manager_count, manager_vector,
                        sub_data->subscribe_context);
@@ -651,6 +654,7 @@ void redis_object_table_get_entry(redisAsyncContext *c,
     LOG_FATAL("expected integer or string, received type %d", reply->type);
   }
   free(managers);
+  LOG_DEBUG("Object table get entry finishing");
 }
 
 void object_table_redis_subscribe_callback(redisAsyncContext *c,
@@ -668,6 +672,7 @@ void object_table_redis_subscribe_callback(redisAsyncContext *c,
    * message is always the payload. */
   object_id id = NIL_ID;
   redisReply *message_type = reply->element[0];
+  LOG_DEBUG("Object table subscribe callback, message %s", message_type->str);
   if (strcmp(message_type->str, "message") == 0) {
     /* A SUBSCRIBE notification. */
     DCHECK(!IS_NIL_ID(callback_data->id));
@@ -682,10 +687,9 @@ void object_table_redis_subscribe_callback(redisAsyncContext *c,
 
     /* Parse the object ID from the keyspace. */
     redisReply *keyspace = reply->element[2];
-    char format[32];
-    snprintf(format, 32, "__keyspace@0__:obj:%%%ldc", sizeof(object_id));
-    int scanned = sscanf(keyspace->str, format, &id);
-    DCHECK(scanned == 1);
+    size_t prefix_length = strlen("__keyspace@0__:obj:");
+    DCHECK(keyspace->len == prefix_length + sizeof(object_id));
+    memcpy(&id, keyspace->str + prefix_length, sizeof(object_id));
   } else if (strcmp(message_type->str, "subscribe") == 0) {
     /* The reply for the initial SUBSCRIBE. */
     DCHECK(reply->elements == 3);
diff --git a/src/common/test/object_table_tests.c b/src/common/test/object_table_tests.c
index 0d896df3e..358713019 100644
--- a/src/common/test/object_table_tests.c
+++ b/src/common/test/object_table_tests.c
@@ -989,7 +989,7 @@ SUITE(object_table_tests) {
   RUN_REDIS_TEST(add_late_test);
   RUN_REDIS_TEST(subscribe_late_test);
   RUN_REDIS_TEST(subscribe_success_test);
-  // RUN_REDIS_TEST(psubscribe_success_test);
+  RUN_REDIS_TEST(psubscribe_success_test);
   RUN_REDIS_TEST(subscribe_object_present_test);
   RUN_REDIS_TEST(subscribe_object_not_present_test);
   RUN_REDIS_TEST(subscribe_object_available_later_test);