peerstore: added 'replace' option and other fixes

7 files changed, 102 insertions, 35 deletions

diff --git a/src/peerstore/gnunet-service-peerstore.c b/src/peerstore/gnunet-service-peerstore.c
index 5d3ea6bb10..4182048507 100644
--- a/src/peerstore/gnunet-service-peerstore.c
+++ b/src/peerstore/gnunet-service-peerstore.c
@@ -30,24 +30,6 @@
 #include "peerstore_common.h"
 
 /**
- * Context of a PEERSTORE watch
- */
-struct WatchContext
-{
-
-  /**
-   * Hash of key of watched record
-   */
-  struct GNUNET_HashCode keyhash;
-
-  /**
-   * Client requested the watch
-   */
-  struct GNUNET_SERVER_Client *client;
-
-};
-
-/**
  * Interval for expired records cleanup (in seconds)
  */
 #define CLEANUP_INTERVAL 300 /* 5mins */
@@ -95,6 +77,7 @@ shutdown_task (void *cls,
   }
   GNUNET_SERVER_notification_context_destroy(nc);
   GNUNET_CONTAINER_multihashmap_destroy(watchers);
+  watchers = NULL;
   GNUNET_SCHEDULER_shutdown();
 }
 
@@ -123,7 +106,7 @@ cleanup_expired_records(void *cls,
  * @param cls closuer, a 'struct GNUNET_PEERSTORE_Record *'
  * @param key hash of record key
  * @param value the watcher client, a 'struct GNUNET_SERVER_Client *'
- * @return #GNUNET_YES to continue iterating
+ * @return #GNUNET_OK to continue iterating
  */
 int client_disconnect_it(void *cls,
     const struct GNUNET_HashCode *key,
@@ -131,7 +114,7 @@ int client_disconnect_it(void *cls,
 {
   if(cls == value)
     GNUNET_CONTAINER_multihashmap_remove(watchers, key, value);
-  return GNUNET_YES;
+  return GNUNET_OK;
 }
 
 /**
@@ -146,8 +129,9 @@ handle_client_disconnect (void *cls,
 			  * client)
 {
   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "A client was disconnected, cleaning up.\n");
-  GNUNET_CONTAINER_multihashmap_iterate(watchers,
-      &client_disconnect_it, client);
+  if(NULL != watchers)
+    GNUNET_CONTAINER_multihashmap_iterate(watchers,
+        &client_disconnect_it, client);
 }
 
 /**
@@ -196,12 +180,6 @@ int watch_notifier_it(void *cls,
   struct StoreRecordMessage *srm;
 
   GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Found a watcher to update.\n");
-  if(NULL == client)
-  {
-    GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Removing a dead client.\n");
-    GNUNET_CONTAINER_multihashmap_remove(watchers, key, client);
-    return GNUNET_YES;
-  }
   srm = PEERSTORE_create_record_message(record->sub_system,
       record->peer,
       record->key,
@@ -335,6 +313,7 @@ void handle_store (void *cls,
     const struct GNUNET_MessageHeader *message)
 {
   struct GNUNET_PEERSTORE_Record *record;
+  struct StoreRecordMessage *srm;
 
   record = PEERSTORE_parse_record_message(message);
   if(NULL == record)
@@ -343,6 +322,7 @@ void handle_store (void *cls,
     GNUNET_SERVER_receive_done(client, GNUNET_SYSERR);
     return;
   }
+  srm = (struct StoreRecordMessage *)message;
   if(NULL == record->sub_system
       || NULL == record->peer
       || NULL == record->key)
@@ -363,7 +343,8 @@ void handle_store (void *cls,
       record->key,
       record->value,
       record->value_size,
-      *record->expiry))
+      *record->expiry,
+      srm->options))
   {
     GNUNET_log(GNUNET_ERROR_TYPE_ERROR, "Failed to store requested value, sqlite database error.");
     PEERSTORE_destroy_record(record);
diff --git a/src/peerstore/peerstore.h b/src/peerstore/peerstore.h
index 5adf9f3631..5757f784ec 100644
--- a/src/peerstore/peerstore.h
+++ b/src/peerstore/peerstore.h
@@ -75,6 +75,12 @@ struct StoreRecordMessage
    */
   struct GNUNET_TIME_Absolute expiry;
 
+  /**
+   * Options, needed only in case of a
+   * store operation
+   */
+  enum GNUNET_PEERSTORE_StoreOption options;
+
 };
 
 /**
diff --git a/src/peerstore/peerstore_api.c b/src/peerstore/peerstore_api.c
index 238c7be194..8797d78180 100644
--- a/src/peerstore/peerstore_api.c
+++ b/src/peerstore/peerstore_api.c
@@ -419,6 +419,7 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h,
     const void *value,
     size_t size,
     struct GNUNET_TIME_Absolute expiry,
+    enum GNUNET_PEERSTORE_StoreOption options,
     GNUNET_PEERSTORE_Continuation cont,
     void *cont_cls)
 {
@@ -434,6 +435,7 @@ GNUNET_PEERSTORE_store (struct GNUNET_PEERSTORE_Handle *h,
       value,
       size,
       &expiry,
+      options,
       GNUNET_MESSAGE_TYPE_PEERSTORE_STORE);
   sc = GNUNET_new(struct GNUNET_PEERSTORE_StoreContext);
   sc->ev = ev;
@@ -595,6 +597,7 @@ GNUNET_PEERSTORE_iterate (struct GNUNET_PEERSTORE_Handle *h,
       NULL,
       0,
       NULL,
+      0,
       GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE);
   ic = GNUNET_new(struct GNUNET_PEERSTORE_IterateContext);
   ic->callback = callback;
diff --git a/src/peerstore/peerstore_common.c b/src/peerstore/peerstore_common.c
index eeaa3144c6..48c04010a8 100644
--- a/src/peerstore/peerstore_common.c
+++ b/src/peerstore/peerstore_common.c
@@ -137,6 +137,7 @@ PEERSTORE_create_record_mq_envelope(const char *sub_system,
     const void *value,
     size_t value_size,
     struct GNUNET_TIME_Absolute *expiry,
+    enum GNUNET_PEERSTORE_StoreOption options,
     uint16_t msg_type)
 {
   struct StoreRecordMessage *srm;
@@ -168,6 +169,7 @@ PEERSTORE_create_record_mq_envelope(const char *sub_system,
   }
   srm->sub_system_size = htons(ss_size);
   srm->value_size = htons(value_size);
+  srm->options = options;
   dummy = &srm[1];
   memcpy(dummy, sub_system, ss_size);
   dummy += ss_size;
diff --git a/src/peerstore/peerstore_common.h b/src/peerstore/peerstore_common.h
index dae2b437bf..297eb9fc09 100644
--- a/src/peerstore/peerstore_common.h
+++ b/src/peerstore/peerstore_common.h
@@ -76,6 +76,7 @@ PEERSTORE_create_record_mq_envelope(const char *sub_system,
     const void *value,
     size_t value_size,
     struct GNUNET_TIME_Absolute *expiry,
+    enum GNUNET_PEERSTORE_StoreOption options,
     uint16_t msg_type);
 
 /**
diff --git a/src/peerstore/plugin_peerstore_sqlite.c b/src/peerstore/plugin_peerstore_sqlite.c
index 8705ce1880..38e72df511 100644
--- a/src/peerstore/plugin_peerstore_sqlite.c
+++ b/src/peerstore/plugin_peerstore_sqlite.c
@@ -100,13 +100,58 @@ struct Plugin
   sqlite3_stmt *select_peerstoredata_by_all;
 
   /**
-   * Precompiled SQL for deleting expired records from peerstoredata
+   * Precompiled SQL for deleting expired
+   * records from peerstoredata
    */
   sqlite3_stmt *expire_peerstoredata;
 
+  /**
+   * Precompiled SQL for deleting records
+   * with given key
+   */
+  sqlite3_stmt *delete_peerstoredata;
+
 };
 
 /**
+ * Delete records with the given key
+ *
+ * @param cls closure (internal context for the plugin)
+ * @param sub_system name of sub system
+ * @param peer Peer identity (can be NULL)
+ * @param key entry key string (can be NULL)
+ * @return number of deleted records
+ */
+static int
+peerstore_sqlite_delete_records(void *cls,
+    const char *sub_system,
+    const struct GNUNET_PeerIdentity *peer,
+    const char *key)
+{
+  struct Plugin *plugin = cls;
+  sqlite3_stmt *stmt = plugin->delete_peerstoredata;
+
+  if((SQLITE_OK != sqlite3_bind_text(stmt, 1, sub_system, strlen(sub_system) + 1, SQLITE_STATIC))
+      || (SQLITE_OK != sqlite3_bind_blob(stmt, 2, peer, sizeof(struct GNUNET_PeerIdentity), SQLITE_STATIC))
+      || (SQLITE_OK != sqlite3_bind_text(stmt, 3, key, strlen(key) + 1, SQLITE_STATIC)))
+  {
+    LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_bind");
+  }
+  else if (SQLITE_DONE != sqlite3_step (stmt))
+  {
+    LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                "sqlite3_step");
+  }
+  if (SQLITE_OK != sqlite3_reset (stmt))
+  {
+    LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
+                "sqlite3_reset");
+    return 0;
+  }
+  return sqlite3_changes(plugin->dbh);
+}
+
+/**
  * Delete expired records (expiry < now)
  *
  * @param cls closure (internal context for the plugin)
@@ -120,7 +165,7 @@ peerstore_sqlite_expire_records(void *cls,
   struct Plugin *plugin = cls;
   sqlite3_stmt *stmt = plugin->expire_peerstoredata;
 
-  if(SQLITE_OK != sqlite3_bind_int64(stmt, 1, (sqlite3_int64)now.abs_value_us))
+  if(SQLITE_OK != sqlite3_bind_int64(stmt, 1, (sqlite3_uint64)now.abs_value_us))
   {
     LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK, "sqlite3_bind");
   }
@@ -254,16 +299,21 @@ peerstore_sqlite_store_record(void *cls,
     const char *key,
     const void *value,
     size_t size,
-    struct GNUNET_TIME_Absolute expiry)
+    struct GNUNET_TIME_Absolute expiry,
+    enum GNUNET_PEERSTORE_StoreOption options)
 {
   struct Plugin *plugin = cls;
   sqlite3_stmt *stmt = plugin->insert_peerstoredata;
 
+  if(GNUNET_PEERSTORE_STOREOPTION_REPLACE == options)
+  {
+    peerstore_sqlite_delete_records(cls, sub_system, peer, key);
+  }
   if(SQLITE_OK != sqlite3_bind_text(stmt, 1, sub_system, strlen(sub_system) + 1, SQLITE_STATIC)
       || SQLITE_OK != sqlite3_bind_blob(stmt, 2, peer, sizeof(struct GNUNET_PeerIdentity), SQLITE_STATIC)
       || SQLITE_OK != sqlite3_bind_text(stmt, 3, key, strlen(key) + 1, SQLITE_STATIC)
       || SQLITE_OK != sqlite3_bind_blob(stmt, 4, value, size, SQLITE_STATIC)
-      || SQLITE_OK != sqlite3_bind_int64(stmt, 5, (sqlite3_int64)expiry.abs_value_us))
+      || SQLITE_OK != sqlite3_bind_int64(stmt, 5, (sqlite3_uint64)expiry.abs_value_us))
     LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                     "sqlite3_bind");
   else if (SQLITE_DONE != sqlite3_step (stmt))
@@ -330,6 +380,21 @@ sql_prepare (sqlite3 *dbh, const char *sql, sqlite3_stmt **stmt)
 }
 
 /**
+ * sqlite3 custom function for comparison of uint64_t values
+ * since it is not supported by default
+ */
+void sqlite3_lessthan(sqlite3_context* ctx, int dummy,
+    sqlite3_value** values)
+{
+  uint64_t v1;
+  uint64_t v2;
+
+  v1 = (uint64_t)sqlite3_value_int64(values[0]);
+  v2 = (uint64_t)sqlite3_value_int64(values[1]);
+  sqlite3_result_int(ctx, v1 < v2);
+}
+
+/**
  * Initialize the database connections and associated
  * data structures (create tables and indices
  * as needed as well).
@@ -389,9 +454,11 @@ database_setup (struct Plugin *plugin)
       "  peer_id BLOB NOT NULL,\n"
       "  key TEXT NOT NULL,\n"
       "  value BLOB NULL,\n"
-      "  expiry INTEGER NOT NULL"
+      "  expiry sqlite3_uint64 NOT NULL"
       ");");
 
+  sqlite3_create_function(plugin->dbh, "UINT64_LT", 2, SQLITE_UTF8, NULL, &sqlite3_lessthan, NULL, NULL);
+
   /* Prepare statements */
 
   sql_prepare (plugin->dbh,
@@ -419,8 +486,14 @@ database_setup (struct Plugin *plugin)
       &plugin->select_peerstoredata_by_all);
   sql_prepare(plugin->dbh,
       "DELETE FROM peerstoredata"
-      " WHERE expiry < ?",
+      " WHERE UINT64_LT(expiry, ?)",
       &plugin->expire_peerstoredata);
+  sql_prepare(plugin->dbh,
+      "DELETE FROM peerstoredata"
+      " WHERE sub_system = ?"
+      " AND peer_id = ?"
+      " AND key = ?",
+      &plugin->delete_peerstoredata);
 
   return GNUNET_OK;
 }
diff --git a/src/peerstore/test_peerstore_api.c b/src/peerstore/test_peerstore_api.c
index 968467231c..e4f6225c6d 100644
--- a/src/peerstore/test_peerstore_api.c
+++ b/src/peerstore/test_peerstore_api.c
@@ -125,6 +125,7 @@ run (void *cls,
       val,
       val_size,
       expiry,
+      GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
       &store_cont,
       NULL);
 }