psyc/social: get state from psycstore

5 files changed, 117 insertions, 43 deletions

diff --git a/src/psycstore/gnunet-service-psycstore.c b/src/psycstore/gnunet-service-psycstore.c
index 6e40e7849..1f9de54f8 100644
--- a/src/psycstore/gnunet-service-psycstore.c
+++ b/src/psycstore/gnunet-service-psycstore.c
@@ -217,7 +217,7 @@ send_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
 
 static int
 send_state_var (void *cls, const char *name,
-                const void *value, size_t value_size)
+                const void *value, uint32_t value_size)
 {
   struct SendClosure *sc = cls;
   struct StateResult *res;
@@ -496,14 +496,14 @@ handle_counters_get (void *cls,
 
 struct StateModifyClosure
 {
-  const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key;
+  const struct GNUNET_CRYPTO_EddsaPublicKey channel_key;
   struct GNUNET_PSYC_ReceiveHandle *recv;
   enum GNUNET_PSYC_MessageState msg_state;
   char mod_oper;
   char *mod_name;
   char *mod_value;
-  uint64_t mod_value_size;
-  uint64_t mod_value_remaining;
+  uint32_t mod_value_size;
+  uint32_t mod_value_remaining;
 };
 
 
@@ -513,6 +513,12 @@ recv_state_message_part (void *cls, uint64_t message_id, uint64_t data_offset,
 {
   struct StateModifyClosure *scls = cls;
   uint16_t psize;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "recv_state_message_part()  message_id: %" PRIu64
+              ", data_offset: %" PRIu64 ", flags: %u\n",
+              message_id, data_offset, flags);
+
   if (NULL == msg)
   {
     scls->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
@@ -533,7 +539,7 @@ recv_state_message_part (void *cls, uint64_t message_id, uint64_t data_offset,
       pmod = (struct GNUNET_PSYC_MessageModifier *) msg;
     psize = ntohs (pmod->header.size);
     uint16_t name_size = ntohs (pmod->name_size);
-    uint16_t value_size = ntohs (pmod->value_size);
+    uint32_t value_size = ntohl (pmod->value_size);
 
     const char *name = (const char *) &pmod[1];
     const void *value = name + name_size;
@@ -542,7 +548,7 @@ recv_state_message_part (void *cls, uint64_t message_id, uint64_t data_offset,
     { // Apply non-transient operation.
       if (psize == sizeof (*pmod) + name_size + value_size)
       {
-        db->state_modify_op (db->cls, scls->channel_key,
+        db->state_modify_op (db->cls, &scls->channel_key,
                              pmod->oper, name, value, value_size);
       }
       else
@@ -576,7 +582,7 @@ recv_state_message_part (void *cls, uint64_t message_id, uint64_t data_offset,
       scls->mod_value_remaining -= psize - sizeof (*msg);
       if (0 == scls->mod_value_remaining)
       {
-        db->state_modify_op (db->cls, scls->channel_key,
+        db->state_modify_op (db->cls, &scls->channel_key,
                              scls->mod_oper, scls->mod_name,
                              scls->mod_value, scls->mod_value_size);
         GNUNET_free (scls->mod_name);
@@ -616,9 +622,13 @@ recv_state_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
                                              scls);
   }
 
-  const struct GNUNET_PSYC_MessageHeader *
-    pmsg = (const struct GNUNET_PSYC_MessageHeader *) &msg[1];
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "recv_state_fragment: %" PRIu64 "\n", GNUNET_ntohll (msg->fragment_id));
+
+  struct GNUNET_PSYC_MessageHeader *
+    pmsg = GNUNET_PSYC_message_header_create (msg, flags);
   GNUNET_PSYC_receive_message (scls->recv, pmsg);
+  GNUNET_free (pmsg);
 
   return GNUNET_YES;
 }
@@ -635,31 +645,41 @@ handle_state_modify (void *cls,
   uint64_t message_id = GNUNET_ntohll (req->message_id);
   uint64_t state_delta = GNUNET_ntohll (req->state_delta);
   uint64_t ret_frags = 0;
+  struct StateModifyClosure
+    scls = { .channel_key = req->channel_key };
 
-  struct StateModifyClosure scls = { 0 };
-
-  if (GNUNET_OK != db->state_modify_begin (db->cls, &req->channel_key,
-                                           message_id, state_delta))
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                _("Failed to begin modifying state!\n"));
-    GNUNET_break (0);
-  }
-
-  int ret = db->message_get (db->cls, &req->channel_key,
-                             message_id, message_id,
-                             &ret_frags, &recv_state_fragment, &scls);
+  int ret = db->state_modify_begin (db->cls, &req->channel_key,
+                                    message_id, state_delta);
 
-  if (GNUNET_OK != db->state_modify_end (db->cls, &req->channel_key, message_id))
+  if (GNUNET_OK != ret)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                _("Failed to end modifying state!\n"));
-    GNUNET_break (0);
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                _("Failed to begin modifying state: %d\n"), ret);
   }
-
-  if (NULL != scls.recv)
+  else
   {
-    GNUNET_PSYC_receive_destroy (scls.recv);
+    ret = db->message_get (db->cls, &req->channel_key,
+                           message_id, message_id,
+                           &ret_frags, &recv_state_fragment, &scls);
+    if (GNUNET_OK != ret)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  _("Failed to modify state: %d\n"), ret);
+      GNUNET_break (0);
+    }
+    else
+    {
+      if (GNUNET_OK != db->state_modify_end (db->cls, &req->channel_key, message_id))
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                    _("Failed to end modifying state!\n"));
+        GNUNET_break (0);
+      }
+    }
+    if (NULL != scls.recv)
+    {
+      GNUNET_PSYC_receive_destroy (scls.recv);
+    }
   }
 
   send_result_code (client, req->op_id, ret, NULL);
diff --git a/src/psycstore/plugin_psycstore_sqlite.c b/src/psycstore/plugin_psycstore_sqlite.c
index 1abc479d2..1bf14644b 100644
--- a/src/psycstore/plugin_psycstore_sqlite.c
+++ b/src/psycstore/plugin_psycstore_sqlite.c
@@ -64,7 +64,8 @@
 
 enum Transactions {
   TRANSACTION_NONE = 0,
-  TRANSACTION_STATE_MODIFY
+  TRANSACTION_STATE_MODIFY,
+  TRANSACTION_STATE_SYNC,
 };
 
 /**
@@ -1522,18 +1523,27 @@ state_modify_begin (void *cls,
 
     uint64_t max_state_message_id = 0;
     int ret = counters_state_get (plugin, channel_key, &max_state_message_id);
-    if (GNUNET_OK != ret)
+    switch (ret)
+    {
+    case GNUNET_OK:
+    case GNUNET_NO: // no state yet
+      ret = GNUNET_OK;
+      break;
+    default:
       return ret;
+    }
 
-    if (message_id - state_delta != max_state_message_id)
-      return GNUNET_NO;
+    if (max_state_message_id < message_id - state_delta)
+      return GNUNET_NO; /* some stateful messages not yet applied */
+    else if (message_id - state_delta < max_state_message_id)
+      return GNUNET_NO; /* changes already applied */
   }
 
-  // Make sure no other transaction is going on.
   if (TRANSACTION_NONE != plugin->transaction)
-      if (GNUNET_OK != transaction_rollback (plugin))
-          return GNUNET_SYSERR;
-
+  {
+    /** @todo FIXME: wait for other transaction to finish  */
+    return GNUNET_SYSERR;
+  }
   return transaction_begin (plugin, TRANSACTION_STATE_MODIFY);
 }
 
@@ -1560,8 +1570,8 @@ state_modify_op (void *cls,
     return state_assign (plugin, plugin->insert_state_current, channel_key,
                          name, value, value_size);
 
-  /// @todo implement more state operations
-  default:
+  default: /** @todo implement more state operations */
+    GNUNET_break (0);
     return GNUNET_SYSERR;
   }
 }
@@ -1630,7 +1640,13 @@ state_sync_end (void *cls,
   struct Plugin *plugin = cls;
   int ret = GNUNET_SYSERR;
 
-  GNUNET_OK == transaction_begin (plugin, TRANSACTION_NONE)
+  if (TRANSACTION_NONE != plugin->transaction)
+  {
+    /** @todo FIXME: wait for other transaction to finish  */
+    return GNUNET_SYSERR;
+  }
+
+  GNUNET_OK == transaction_begin (plugin, TRANSACTION_STATE_SYNC)
     && GNUNET_OK == exec_channel (plugin, plugin->delete_state, channel_key)
     && GNUNET_OK == exec_channel (plugin, plugin->insert_state_from_sync,
                                   channel_key)
diff --git a/src/psycstore/psyc_util_lib.c b/src/psycstore/psyc_util_lib.c
index 75f97aad0..f6dd4e593 100644
--- a/src/psycstore/psyc_util_lib.c
+++ b/src/psycstore/psyc_util_lib.c
@@ -570,7 +570,7 @@ transmit_notify_env (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
 {
   struct GNUNET_PSYC_TransmitHandle *tmit = cls;
   uint16_t name_size = 0;
-  size_t value_size = 0;
+  uint32_t value_size = 0;
   const char *value = NULL;
 
   if (NULL != oper)
@@ -1231,3 +1231,41 @@ GNUNET_PSYC_message_parse (const struct GNUNET_PSYC_Message *msg,
     ? GNUNET_OK
     : GNUNET_SYSERR;
 }
+
+
+/**
+ * Initialize PSYC message header.
+ */
+void
+GNUNET_PSYC_message_header_init (struct GNUNET_PSYC_MessageHeader *pmsg,
+                                 const struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                                 uint32_t flags)
+{
+  uint16_t size = ntohs (mmsg->header.size);
+  uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+
+  pmsg->header.size = htons (psize);
+  pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+  pmsg->message_id = mmsg->message_id;
+  pmsg->fragment_offset = mmsg->fragment_offset;
+  pmsg->flags = htonl (flags);
+
+  memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
+}
+
+
+/**
+ * Create a new PSYC message header from a multicast message for sending it to clients.
+ */
+struct GNUNET_PSYC_MessageHeader *
+GNUNET_PSYC_message_header_create (const struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                                   uint32_t flags)
+{
+  struct GNUNET_PSYC_MessageHeader *pmsg;
+  uint16_t size = ntohs (mmsg->header.size);
+  uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+
+  pmsg = GNUNET_malloc (psize);
+  GNUNET_PSYC_message_header_init (pmsg, mmsg, flags);
+  return pmsg;
+}
diff --git a/src/psycstore/test_plugin_psycstore.c b/src/psycstore/test_plugin_psycstore.c
index 0a7824929..5a5f970b9 100644
--- a/src/psycstore/test_plugin_psycstore.c
+++ b/src/psycstore/test_plugin_psycstore.c
@@ -141,7 +141,7 @@ struct StateClosure {
 };
 
 static int
-state_cb (void *cls, const char *name, const void *value, size_t value_size)
+state_cb (void *cls, const char *name, const void *value, uint32_t value_size)
 {
   struct StateClosure *scls = cls;
   const void *val = scls->value[scls->n];
diff --git a/src/psycstore/test_psycstore.c b/src/psycstore/test_psycstore.c
index c869a862f..58e6243b7 100644
--- a/src/psycstore/test_psycstore.c
+++ b/src/psycstore/test_psycstore.c
@@ -170,7 +170,7 @@ state_reset_result (void *cls, int64_t result,
 
 
 static int
-state_result (void *cls, const char *name, const void *value, size_t value_size)
+state_result (void *cls, const char *name, const void *value, uint32_t value_size)
 {
   struct StateClosure *scls = cls;
   const char *nam = scls->name[scls->n];