psyc/social: get state from psycstore

3 files changed, 281 insertions, 45 deletions

diff --git a/src/social/gnunet-service-social.c b/src/social/gnunet-service-social.c
index 4211772f1..56612b488 100644
--- a/src/social/gnunet-service-social.c
+++ b/src/social/gnunet-service-social.c
@@ -350,7 +350,7 @@ cleanup_guest (struct Guest *gst)
   struct GNUNET_CONTAINER_MultiHashMap *
     plc_gst = GNUNET_CONTAINER_multihashmap_get (place_guests,
                                                 &plc->pub_key_hash);
-  GNUNET_assert (NULL != plc_gst);
+  GNUNET_assert (NULL != plc_gst); // FIXME
   GNUNET_CONTAINER_multihashmap_remove (plc_gst, &gst->pub_key_hash, gst);
 
   if (0 == GNUNET_CONTAINER_multihashmap_size (plc_gst))
@@ -1302,7 +1302,7 @@ psyc_transmit_queue_next_method (struct Place *plc,
   {
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                 "%p psyc_transmit_queue_next_method: unexpected message part of type %u.\n",
-                plc, ntohs (pmsg->type));
+                plc, NULL != pmsg ? ntohs (pmsg->type) : 0);
     GNUNET_break (0);
     return GNUNET_SYSERR;
   }
@@ -1536,12 +1536,10 @@ client_recv_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
 
 
 /**
- * A historic message result arrived from PSYC.
+ * A historic message arrived from PSYC.
  */
 static void
-psyc_recv_history_message (void *cls,
-                           uint64_t message_id,
-                           uint32_t flags,
+psyc_recv_history_message (void *cls, uint64_t message_id, uint32_t flags,
                            const struct GNUNET_PSYC_MessageHeader *msg)
 {
   struct OperationClosure *opcls = cls;
@@ -1567,6 +1565,9 @@ psyc_recv_history_message (void *cls,
 }
 
 
+/**
+ * Result of message history replay from PSYC.
+ */
 static void
 psyc_recv_history_result (void *cls, int64_t result,
                           const void *err_msg, uint16_t err_msg_size)
@@ -1574,7 +1575,7 @@ psyc_recv_history_result (void *cls, int64_t result,
   struct OperationClosure *opcls = cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "%p History replay #%" PRIu64 ": "
-              "PSYCSTORE returned %" PRId64 " (%.*s)\n",
+              "PSYCstore returned %" PRId64 " (%.*s)\n",
               opcls->plc, GNUNET_ntohll (opcls->op_id), result, err_msg_size, err_msg);
 
   // FIXME: place might have been destroyed
@@ -1635,6 +1636,118 @@ client_recv_history_replay (void *cls, struct GNUNET_SERVER_Client *client,
 }
 
 
+/**
+ * A state variable part arrived from PSYC.
+ */
+void
+psyc_recv_state_var (void *cls,
+                     const struct GNUNET_MessageHeader *mod,
+                     const char *name,
+                     const void *value,
+                     uint32_t value_size,
+                     uint32_t full_value_size)
+{
+  struct OperationClosure *opcls = cls;
+  struct Place *plc = opcls->plc;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received state variable %s from PSYC\n",
+              plc, name);
+
+  uint16_t size = ntohs (mod->size);
+
+  struct GNUNET_OperationResultMessage *
+    res = GNUNET_malloc (sizeof (*res) + size);
+  res->header.size = htons (sizeof (*res) + size);
+  res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT);
+  res->op_id = opcls->op_id;
+  res->result_code = GNUNET_htonll (GNUNET_OK);
+
+  memcpy (&res[1], mod, size);
+
+  /** @todo FIXME: send only to requesting client */
+  client_send_msg (plc, &res->header);
+}
+
+
+/**
+ * Result of retrieving state variable from PSYC.
+ */
+static void
+psyc_recv_state_result (void *cls, int64_t result,
+                        const void *err_msg, uint16_t err_msg_size)
+{
+  struct OperationClosure *opcls = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p State get #%" PRIu64 ": "
+              "PSYCstore returned %" PRId64 " (%.*s)\n",
+              opcls->plc, GNUNET_ntohll (opcls->op_id), result, err_msg_size, err_msg);
+
+  // FIXME: place might have been destroyed
+  client_send_result (opcls->client, opcls->op_id, result, err_msg, err_msg_size);
+}
+
+
+/**
+ * Client requests channel history.
+ */
+static void
+client_recv_state_get (void *cls, struct GNUNET_SERVER_Client *client,
+                       const struct GNUNET_MessageHeader *msg)
+{
+  struct Client *
+    ctx = GNUNET_SERVER_client_get_user_context (client, struct Client);
+  GNUNET_assert (NULL != ctx);
+  struct Place *plc = ctx->plc;
+
+  const struct GNUNET_PSYC_StateRequestMessage *
+    req = (const struct GNUNET_PSYC_StateRequestMessage *) msg;
+  uint16_t size = ntohs (msg->size);
+  const char *name = (const char *) &req[1];
+
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "%p State get #%" PRIu64 ": %s\n",
+              plc, GNUNET_ntohll (req->op_id), name);
+
+  if (size < sizeof (*req) + 1
+      || '\0' != name[size - sizeof (*req) - 1])
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p State get #%" PRIu64 ": "
+                "invalid name. size: %u < %u?\n",
+                plc, GNUNET_ntohll (req->op_id), size, sizeof (*req) + 1);
+    GNUNET_break (0);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  struct OperationClosure *opcls = GNUNET_malloc (sizeof (*opcls));
+  opcls->client = client;
+  opcls->plc = plc;
+  opcls->op_id = req->op_id;
+
+  switch (ntohs (msg->type))
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_STATE_GET:
+      GNUNET_PSYC_channel_state_get (plc->channel, name,
+                                     psyc_recv_state_var,
+                                     psyc_recv_state_result, opcls);
+      break;
+
+  case GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX:
+      GNUNET_PSYC_channel_state_get_prefix (plc->channel, name,
+                                            psyc_recv_state_var,
+                                            psyc_recv_state_result, opcls);
+      break;
+
+  default:
+      GNUNET_assert (0);
+  }
+
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
 static const struct GNUNET_SERVER_MessageHandler handlers[] = {
   { &client_recv_host_enter, NULL,
     GNUNET_MESSAGE_TYPE_SOCIAL_HOST_ENTER, 0 },
@@ -1650,13 +1763,13 @@ static const struct GNUNET_SERVER_MessageHandler handlers[] = {
 
   { &client_recv_history_replay, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY, 0 },
-#if FIXME
+
   { &client_recv_state_get, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_STATE_GET, 0 },
 
-  { &client_recv_state_get_prefix, NULL,
+  { &client_recv_state_get, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX, 0 },
-#endif
+
   { NULL, NULL, 0, 0 }
 };
 
diff --git a/src/social/social_api.c b/src/social/social_api.c
index 74a13cc35..20d0dc8b8 100644
--- a/src/social/social_api.c
+++ b/src/social/social_api.c
@@ -310,6 +310,21 @@ struct GNUNET_SOCIAL_LookHandle
   GNUNET_ResultCallback result_cb;
 
   /**
+   * Name of current modifier being received.
+   */
+  char *mod_name;
+
+  /**
+   * Size of current modifier value being received.
+   */
+  size_t mod_value_size;
+
+  /**
+   * Remaining size of current modifier value still to be received.
+   */
+  size_t mod_value_remaining;
+
+  /**
    * Closure for @a result_cb.
    */
   void *cls;
@@ -753,49 +768,66 @@ place_recv_state_result (void *cls,
   const struct GNUNET_OperationResultMessage *
     res = (const struct GNUNET_OperationResultMessage *) msg;
 
-#if FIXME
   GNUNET_ResultCallback result_cb = NULL;
-  struct GNUNET_PSYC_StateRequest *sr = NULL;
+  struct GNUNET_SOCIAL_LookHandle *look = NULL;
 
   if (GNUNET_YES != GNUNET_CLIENT_MANAGER_op_find (plc->client,
                                                    GNUNET_ntohll (res->op_id),
-                                                   &result_cb, (void *) &sr))
+                                                   &result_cb, (void *) &look))
   { /* Operation not found. */
     return;
   }
 
   const struct GNUNET_MessageHeader *
-    modc = (struct GNUNET_MessageHeader *) &res[1];
-  uint16_t modc_size = ntohs (modc->size);
-  if (ntohs (msg->size) - sizeof (*msg) != modc_size)
+    mod = (struct GNUNET_MessageHeader *) &res[1];
+  uint16_t mod_size = ntohs (mod->size);
+  if (ntohs (msg->size) - sizeof (*res) != mod_size)
   {
-    GNUNET_break (0);
+    GNUNET_break_op (0);
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "Invalid modifier size in state result: %u - %u != %u\n",
+         ntohs (msg->size), sizeof (*res), mod_size);
     return;
   }
-  switch (ntohs (modc->type))
+  switch (ntohs (mod->type))
   {
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
   {
     const struct GNUNET_PSYC_MessageModifier *
-      mod = (const struct GNUNET_PSYC_MessageModifier *) modc;
+      pmod = (const struct GNUNET_PSYC_MessageModifier *) mod;
 
-    const char *name = (const char *) &mod[1];
-    uint16_t name_size = ntohs (mod->name_size);
+    const char *name = (const char *) &pmod[1];
+    uint16_t name_size = ntohs (pmod->name_size);
     if ('\0' != name[name_size - 1])
     {
-      GNUNET_break (0);
+      GNUNET_break_op (0);
+      LOG (GNUNET_ERROR_TYPE_WARNING,
+           "Invalid modifier name in state result\n");
       return;
     }
-    sr->var_cb (sr->cls, name, name + name_size, ntohs (mod->value_size));
+    look->mod_value_size = ntohs (pmod->value_size);
+    look->var_cb (look->cls, mod, name, name + name_size,
+                  mod_size - sizeof (*mod) - name_size,
+                  look->mod_value_size);
+    if (look->mod_value_size > mod_size - sizeof (*mod) - name_size)
+    {
+        look->mod_value_remaining = look->mod_value_size;
+        look->mod_name = GNUNET_malloc (name_size);
+        memcpy (look->mod_name, name, name_size);
+    }
     break;
   }
 
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
-    sr->var_cb (sr->cls, NULL, (const char *) &modc[1],
-                modc_size - sizeof (*modc));
+    look->var_cb (look->cls, mod, look->mod_name, (const char *) &mod[1],
+                  mod_size - sizeof (*mod), look->mod_value_size);
+    look->mod_value_remaining -= mod_size - sizeof (*mod);
+    if (0 == look->mod_value_remaining)
+    {
+        GNUNET_free (look->mod_name);
+    }
     break;
   }
-#endif
 }
 
 
@@ -1980,7 +2012,7 @@ place_state_get (struct GNUNET_SOCIAL_Place *plc,
  * what was requested).
  *
  * @param place
- *        The place to look the object at.
+ *        The place where to look.
  * @param full_name
  *        Full name of the object.
  * @param value_size
@@ -2004,7 +2036,7 @@ GNUNET_SOCIAL_place_look_at (struct GNUNET_SOCIAL_Place *plc,
  * Look for objects in the place with a matching name prefix.
  *
  * @param place
- *        The place to look its objects at.
+ *        The place where to look.
  * @param name_prefix
  *        Look at objects with names beginning with this value.
  * @param var_cb
diff --git a/src/social/test_social.c b/src/social/test_social.c
index 19a81f43d..dbcf822f8 100644
--- a/src/social/test_social.c
+++ b/src/social/test_social.c
@@ -36,7 +36,7 @@
 #include "gnunet_core_service.h"
 #include "gnunet_identity_service.h"
 
-#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
+#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 300)
 
 #define DATA2ARG(data) data, sizeof (data)
 
@@ -99,6 +99,10 @@ struct TransmitClosure
   uint8_t n;
 } tmit;
 
+struct ResultClosure {
+  uint32_t n;
+};
+
 uint8_t join_req_count;
 struct GNUNET_PSYC_Message *join_resp;
 
@@ -107,19 +111,21 @@ uint32_t counter;
 enum
 {
   TEST_NONE = 0,
-  TEST_HOST_ANSWER_DOOR_REFUSE      = 1,
-  TEST_GUEST_RECV_ENTRY_DCSN_REFUSE = 2,
-  TEST_HOST_ANSWER_DOOR_ADMIT       = 3,
-  TEST_GUEST_RECV_ENTRY_DCSN_ADMIT  = 4,
-  TEST_HOST_ANNOUNCE                = 5,
-  TEST_HOST_ANNOUNCE_END            = 6,
-  TEST_HOST_ANNOUNCE2               = 7,
-  TEST_HOST_ANNOUNCE2_END           = 8,
-  TEST_GUEST_TALK                   = 9,
-  TEST_GUEST_HISTORY_REPLAY        = 10,
-  TEST_GUEST_HISTORY_REPLAY_LATEST = 11,
-  TEST_GUEST_LEAVE                 = 12,
-  TEST_HOST_LEAVE                  = 13,
+  TEST_HOST_ANSWER_DOOR_REFUSE      =  1,
+  TEST_GUEST_RECV_ENTRY_DCSN_REFUSE =  2,
+  TEST_HOST_ANSWER_DOOR_ADMIT       =  3,
+  TEST_GUEST_RECV_ENTRY_DCSN_ADMIT  =  4,
+  TEST_HOST_ANNOUNCE                =  5,
+  TEST_HOST_ANNOUNCE_END            =  6,
+  TEST_HOST_ANNOUNCE2               =  7,
+  TEST_HOST_ANNOUNCE2_END           =  8,
+  TEST_GUEST_TALK                   =  9,
+  TEST_GUEST_HISTORY_REPLAY         = 10,
+  TEST_GUEST_HISTORY_REPLAY_LATEST  = 11,
+  TEST_GUEST_LOOK_AT                = 12,
+  TEST_GUEST_LOOK_FOR               = 13,
+  TEST_GUEST_LEAVE                  = 14,
+  TEST_HOST_LEAVE                   = 15,
 } test;
 
 
@@ -351,6 +357,86 @@ schedule_guest_leave (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 
 
 static void
+guest_look_for_result (void *cls, int64_t result_code,
+                      const void *data, uint16_t data_size)
+{
+  struct ResultClosure *rcls = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "guest_look_for_result: %d\n", result_code);
+  GNUNET_assert (GNUNET_OK == result_code);
+  GNUNET_assert (3 == rcls->n);
+  GNUNET_free (rcls);
+  GNUNET_SCHEDULER_add_now (&schedule_guest_leave, NULL);
+}
+
+
+static void
+guest_look_for_var (void *cls,
+                   const struct GNUNET_MessageHeader *mod,
+                   const char *name,
+                   const void *value,
+                   uint32_t value_size,
+                   uint32_t full_value_size)
+{
+  struct ResultClosure *rcls = cls;
+  rcls->n++;
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "guest_look_for_var: %s\n%.*s\n",
+              name, value_size, value);
+}
+
+
+static void
+guest_look_for ()
+{
+  test = TEST_GUEST_LOOK_FOR;
+  struct ResultClosure *rcls = GNUNET_malloc (sizeof (*rcls));
+  GNUNET_SOCIAL_place_look_for (gst_plc, "_foo", guest_look_for_var, guest_look_for_result, rcls);
+}
+
+
+static void
+guest_look_at_result (void *cls, int64_t result_code,
+                      const void *data, uint16_t data_size)
+{
+  struct ResultClosure *rcls = cls;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "guest_look_at_result: %d\n", result_code);
+  GNUNET_assert (GNUNET_OK == result_code);
+  GNUNET_assert (1 == rcls->n);
+  GNUNET_free (rcls);
+  guest_look_for ();
+}
+
+
+static void
+guest_look_at_var (void *cls,
+                   const struct GNUNET_MessageHeader *mod,
+                   const char *name,
+                   const void *value,
+                   uint32_t value_size,
+                   uint32_t full_value_size)
+{
+  struct ResultClosure *rcls = cls;
+  rcls->n++;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "guest_look_at_var: %s\n%.*s\n",
+              name, value_size, value);
+}
+
+
+static void
+guest_look_at ()
+{
+  test = TEST_GUEST_LOOK_AT;
+  struct ResultClosure *rcls = GNUNET_malloc (sizeof (*rcls));
+  GNUNET_SOCIAL_place_look_at (gst_plc, "_foo_bar", guest_look_at_var, guest_look_at_result, rcls);
+}
+
+
+static void
 guest_recv_history_replay_latest_result (void *cls, int64_t result,
                                          const void *data, uint16_t data_size)
 {
@@ -361,7 +447,7 @@ guest_recv_history_replay_latest_result (void *cls, int64_t result,
   GNUNET_assert (2 == counter); /* message count */
   GNUNET_assert (7 == result); /* fragment count */
 
-  GNUNET_SCHEDULER_add_now (&schedule_guest_leave, NULL);
+  guest_look_at ();
 }
 
 
@@ -488,6 +574,7 @@ guest_recv_eom (void *cls,
     break;
 
   default:
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "invalid test: %d\n", test);
     GNUNET_assert (0);
   }
 }
@@ -570,10 +657,11 @@ host_recv_eom (void *cls,
     break;
 
   case TEST_GUEST_TALK:
-    guest_history_replay ();
+      guest_history_replay ();
     break;
 
   default:
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "invalid test: %d\n", test);
     GNUNET_assert (0);
   }
 }
@@ -624,7 +712,8 @@ host_announce ()
   tmit.host_ann
     = GNUNET_SOCIAL_host_announce (hst, "_message_host", tmit.env,
                                    &notify_data, &tmit,
-                                   GNUNET_SOCIAL_ANNOUNCE_NONE);
+                                   GNUNET_SOCIAL_ANNOUNCE_NONE
+                                   | GNUNET_PSYC_MASTER_TRANSMIT_STATE_MODIFY);
 }
 
 
@@ -689,6 +778,7 @@ guest_recv_entry_decision (void *cls,
     break;
 
   default:
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "invalid test: %d\n", test);
     GNUNET_assert (0);
   }
 }
@@ -728,6 +818,7 @@ host_answer_door (void *cls,
     break;
 
   default:
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "invalid test: %d\n", test);
     GNUNET_assert (0);
   }
 }