2 files changed, 151 insertions, 42 deletions

diff --git a/TODO b/TODO
index acac5e4f3..6c0de7df2 100644
--- a/TODO
+++ b/TODO
@@ -6,9 +6,6 @@
 * UTIL:
   - only connect() sockets that are ready (select()) [Nils]
     [On W32, we need to select after calling socket before doing connect etc.]
-* DATASTORE [CG]:
-  - check for duplicates on insertion (currently, same content is frequently
-    stored again [seen with KBLOCKS and SBLOCKS]!)
 * GNUNET-GTK: [CG]
   - directory support:
     + download start: recursive/directory (!) & from-URI only
diff --git a/src/datastore/gnunet-service-datastore.c b/src/datastore/gnunet-service-datastore.c
index faae1ae99..073c74dd9 100644
--- a/src/datastore/gnunet-service-datastore.c
+++ b/src/datastore/gnunet-service-datastore.c
@@ -870,6 +870,142 @@ check_data (const struct GNUNET_MessageHeader *message)
 
 
 /**
+ * Context for a put request used to see if the content is
+ * already present.
+ */
+struct PutContext
+{
+  /**
+   * Client to notify on completion.
+   */
+  struct GNUNET_SERVER_Client *client;
+
+  /**
+   * Did we find the data already in the database?
+   */
+  int is_present;
+  
+  /* followed by the 'struct DataMessage' */
+};
+
+
+/**
+ * Actually put the data message.
+ */
+static void
+execute_put (struct GNUNET_SERVER_Client *client,
+             const struct DataMessage *dm)
+{
+  uint32_t size;
+  char *msg;
+  int ret;
+
+  size = ntohl(dm->size);
+  msg = NULL;
+  ret = plugin->api->put (plugin->api->cls,
+                          &dm->key,
+                          size,
+                          &dm[1],
+                          ntohl(dm->type),
+                          ntohl(dm->priority),
+                          ntohl(dm->anonymity),
+                          GNUNET_TIME_absolute_ntoh(dm->expiration),
+                          &msg);
+  if (GNUNET_OK == ret)
+    {
+      GNUNET_STATISTICS_update (stats,
+                                gettext_noop ("# bytes stored"),
+                                size,
+                                GNUNET_YES);
+      GNUNET_CONTAINER_bloomfilter_add (filter,
+                                        &dm->key);
+#if DEBUG_DATASTORE
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Successfully stored %u bytes of type %u under key `%s'\n",
+                  size,
+                  ntohl(dm->type),
+                  GNUNET_h2s (&dm->key));
+#endif
+    }
+  transmit_status (client, 
+                   (GNUNET_SYSERR == ret) ? GNUNET_SYSERR : GNUNET_OK, 
+                   msg);
+  GNUNET_free_non_null (msg);
+  if (quota - reserved - cache_size < plugin->api->get_size (plugin->api->cls))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  _("Need %llu bytes more space (%llu allowed, using %llu)\n"),
+                  (unsigned long long) size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
+                  (unsigned long long) (quota - reserved - cache_size),
+                  (unsigned long long) plugin->api->get_size (plugin->api->cls));
+      manage_space (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
+    }
+}
+
+
+
+/**
+ * Function that will check if the given datastore entry
+ * matches the put and if none match executes the put.
+ *
+ * @param cls closure, pointer to the client (of type 'struct PutContext').
+ * @param next_cls closure to use to ask for the next item
+ * @param key key for the content
+ * @param size number of bytes in data
+ * @param data content stored
+ * @param type type of the content
+ * @param priority priority of the content
+ * @param anonymity anonymity-level for the content
+ * @param expiration expiration time for the content
+ * @param uid unique identifier for the datum;
+ *        maybe 0 if no unique identifier is available
+ *
+ * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue,
+ *         GNUNET_NO to delete the item and continue (if supported)
+ */
+static int
+check_present (void *cls,
+               void *next_cls,
+               const GNUNET_HashCode * key,
+               uint32_t size,
+               const void *data,
+               enum GNUNET_BLOCK_Type type,
+               uint32_t priority,
+               uint32_t anonymity,
+               struct GNUNET_TIME_Absolute
+               expiration, uint64_t uid)
+{
+  struct PutContext *pc = cls;
+  const struct DataMessage *dm;
+
+  dm = (const struct DataMessage*) &pc[1];
+  if (key == NULL)
+    {
+      if (pc->is_present == GNUNET_YES) 
+        transmit_status (pc->client, GNUNET_OK, NULL);
+      else
+        execute_put (pc->client, dm);
+      GNUNET_SERVER_client_drop (pc->client);
+      GNUNET_free (pc);
+      return GNUNET_SYSERR;
+    }
+  if ( (size == ntohl(dm->size)) &&
+       (0 == memcmp (&dm[1],
+                     data,
+                     size)) )
+    {
+      pc->is_present = GNUNET_YES;
+      plugin->api->next_request (next_cls, GNUNET_YES);
+    }
+  else
+    {
+      plugin->api->next_request (next_cls, GNUNET_NO);
+    }
+  return GNUNET_OK;
+}
+
+
+/**
  * Handle PUT-message.
  *
  * @param cls closure
@@ -882,10 +1018,9 @@ handle_put (void *cls,
             const struct GNUNET_MessageHeader *message)
 {
   const struct DataMessage *dm = check_data (message);
-  char *msg;
-  int ret;
   int rid;
   struct ReservationList *pos;
+  struct PutContext *pc;
   uint32_t size;
 
   if ( (dm == NULL) ||
@@ -924,45 +1059,22 @@ handle_put (void *cls,
                                  GNUNET_NO);
         }
     }
-  msg = NULL;
-  ret = plugin->api->put (plugin->api->cls,
-                          &dm->key,
-                          size,
-                          &dm[1],
-                          ntohl(dm->type),
-                          ntohl(dm->priority),
-                          ntohl(dm->anonymity),
-                          GNUNET_TIME_absolute_ntoh(dm->expiration),
-                          &msg);
-  if (GNUNET_OK == ret)
-    {
-      GNUNET_STATISTICS_update (stats,
-                                gettext_noop ("# bytes stored"),
-                                size,
-                                GNUNET_YES);
-      GNUNET_CONTAINER_bloomfilter_add (filter,
-                                        &dm->key);
-#if DEBUG_DATASTORE
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "Successfully stored %u bytes of type %u under key `%s'\n",
-                  size,
-                  ntohl(dm->type),
-                  GNUNET_h2s (&dm->key));
-#endif
-    }
-  transmit_status (client, 
-                   (GNUNET_SYSERR == ret) ? GNUNET_SYSERR : GNUNET_OK, 
-                   msg);
-  GNUNET_free_non_null (msg);
-  if (quota - reserved - cache_size < plugin->api->get_size (plugin->api->cls))
+  if (GNUNET_YES == GNUNET_CONTAINER_bloomfilter_test (filter,
+                                                       &dm->key))
     {
-      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                  _("Need %llu bytes more space (%llu allowed, using %llu)\n"),
-                  (unsigned long long) size + GNUNET_DATASTORE_ENTRY_OVERHEAD,
-                  (unsigned long long) (quota - reserved - cache_size),
-                  (unsigned long long) plugin->api->get_size (plugin->api->cls));
-      manage_space (size + GNUNET_DATASTORE_ENTRY_OVERHEAD);
+      pc = GNUNET_malloc (sizeof (struct PutContext) + size + sizeof (struct DataMessage));
+      pc->client = client;
+      GNUNET_SERVER_client_keep (client);
+      memcpy (&pc[1], dm, size + sizeof (struct DataMessage));
+      plugin->api->get (plugin->api->cls,
+                        &dm->key,
+                        NULL,
+                        ntohl (dm->type),
+                        &check_present,
+                        pc);      
+      return;
     }
+  execute_put (client, dm);
 }