towards new peerinfo API

1 files changed, 372 insertions, 3 deletions

diff --git a/src/peerinfo/peerinfo_api.c b/src/peerinfo/peerinfo_api.c
index 89422bcb6..aac83f71b 100644
--- a/src/peerinfo/peerinfo_api.c
+++ b/src/peerinfo/peerinfo_api.c
@@ -22,14 +22,63 @@
  * @file peerinfo/peerinfo_api.c
  * @brief API to access peerinfo service
  * @author Christian Grothoff
+ *
+ * TODO:
+ * - document NEW API implementation
+ * - add timeout for iteration
+ * - implement cancellation of iteration
  */
 #include "platform.h"
 #include "gnunet_client_lib.h"
+#include "gnunet_container_lib.h"
 #include "gnunet_peerinfo_service.h"
 #include "gnunet_protocols.h"
 #include "gnunet_time_lib.h"
 #include "peerinfo.h"
 
+/**
+ *
+ */
+typedef void (*TransmissionContinuation)(void *cls,
+                                         int success);
+
+
+/**
+ *
+ */
+struct TransmissionQueueEntry
+{
+  /**
+   *
+   */
+  struct TransmissionQueueEntry *next;
+  
+  /**
+   *
+   */
+  struct TransmissionQueueEntry *prev;
+
+  /**
+   *
+   */
+  TransmissionContinuation cont;
+  
+  /**
+   *
+   */
+  void *cont_cls;
+
+  /**
+   *
+   */
+  struct GNUNET_TIME_Absolute timeout;
+
+  /**
+   *
+   */
+  size_t size;
+
+};
 
 
 /**
@@ -37,6 +86,35 @@
  */
 struct GNUNET_PEERINFO_Handle
 {
+  /**
+   * Our configuration.
+   */
+  const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+  /**
+   * Our scheduler.
+   */
+  struct GNUNET_SCHEDULER_Handle *sched;
+
+  /**
+   * Connection to the service.
+   */
+  struct GNUNET_CLIENT_Connection *client;
+
+  /**
+   *
+   */
+  struct TransmissionQueueEntry *tq_head;
+
+  /**
+   *
+   */
+  struct TransmissionQueueEntry *tq_tail;
+
+  /**
+   *
+   */
+  struct GNUNET_CLIENT_TransmitHandle *th;
 };
 
 
@@ -52,7 +130,17 @@ struct GNUNET_PEERINFO_Handle *
 GNUNET_PEERINFO_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
                          struct GNUNET_SCHEDULER_Handle *sched)
 {
-  return NULL;
+  struct GNUNET_CLIENT_Connection *client;
+  struct GNUNET_PEERINFO_Handle *ret;
+
+  client = GNUNET_CLIENT_connect (sched, "peerinfo", cfg);
+  if (client == NULL)
+    return NULL;
+  ret = GNUNET_malloc (sizeof (struct GNUNET_PEERINFO_Handle));
+  ret->client = client;
+  ret->cfg = cfg;
+  ret->sched = sched;
+  return ret;
 }
 
 
@@ -68,11 +156,100 @@ GNUNET_PEERINFO_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 void
 GNUNET_PEERINFO_disconnect (struct GNUNET_PEERINFO_Handle *h)
 {
+  struct TransmissionQueueEntry *tqe;
+
+  while (NULL != (tqe = h->tq_head))
+    {     
+      GNUNET_CONTAINER_DLL_remove (h->tq_head,
+                                   h->tq_tail,
+                                   tqe);
+      if (tqe->cont != NULL)
+        tqe->cont (tqe->cont_cls, GNUNET_SYSERR);
+      GNUNET_free (tqe);
+    }
+  GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
+  GNUNET_free (h);
 }
 
 
+/**
+ *
+ */
+static void
+trigger_transmit (struct GNUNET_PEERINFO_Handle *h);
+
+
+/**
+ *
+ */
+static void
+reconnect (struct GNUNET_PEERINFO_Handle *h)
+{
+  GNUNET_CLIENT_disconnect (h->client, GNUNET_SYSERR);
+  h->client = GNUNET_CLIENT_connect (h->sched, "client", h->cfg);
+  GNUNET_assert (h->client != NULL);
+}
+
+
+/**
+ *
+ */
+static size_t
+do_transmit (void *cls, size_t size, void *buf)
+{
+  struct GNUNET_PEERINFO_Handle *h = cls;
+  struct TransmissionQueueEntry *tqe = h->tq_head;
+  size_t ret;
+
+  h->th = NULL;
+  if (buf == NULL)
+    {
+#if DEBUG_PEERINFO
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  _
+                  ("Failed to transmit message of type %u to `%s' service.\n"),
+                  ntohs (msg->type), "peerinfo");
+#endif
+      GNUNET_CONTAINER_DLL_remove (h->tq_head,
+                                   h->tq_tail,
+                                   tqe);
+      reconnect (h);
+      trigger_transmit (h);
+      if (tqe->cont != NULL)
+        tqe->cont (tqe->cont_cls, GNUNET_SYSERR);
+      GNUNET_free (tqe);
+      return 0;
+    }
+  ret = tqe->size;
+  GNUNET_assert (size >= ret);
+  memcpy (buf, &tqe[1], ret);
+  GNUNET_CONTAINER_DLL_remove (h->tq_head,
+                               h->tq_tail,
+                               tqe);
+  if (tqe->cont != NULL)
+    tqe->cont (tqe->cont_cls, GNUNET_OK);
+  else
+    trigger_transmit (h);
+  GNUNET_free (tqe);
+  return ret;
+}
 
 
+static void
+trigger_transmit (struct GNUNET_PEERINFO_Handle *h)
+{
+  struct TransmissionQueueEntry *tqe;
+
+  if (NULL == (tqe = h->tq_head))
+    return;
+  if (h->th != NULL)
+    return;
+  h->th = GNUNET_CLIENT_notify_transmit_ready (h->client,
+                                               tqe->size,
+                                               GNUNET_TIME_absolute_get_remaining (tqe->timeout),
+                                               GNUNET_YES,
+                                               &do_transmit, h);  
+}
 
 
 /**
@@ -89,17 +266,163 @@ GNUNET_PEERINFO_disconnect (struct GNUNET_PEERINFO_Handle *h)
  */
 void
 GNUNET_PEERINFO_add_peer_new (struct GNUNET_PEERINFO_Handle *h,
-                              const struct GNUNET_PeerIdentity *peer,
                               const struct GNUNET_HELLO_Message *hello)
 {
+  uint16_t hs = GNUNET_HELLO_size (hello);
+  struct TransmissionQueueEntry *tqe;
+  
+#if DEBUG_PEERINFO
+  struct GNUNET_PeerIdentity peer;
+  GNUNET_assert (GNUNET_OK == GNUNET_HELLO_get_id (hello, &peer));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Adding peer `%s' to PEERINFO database (%u bytes of `%s')\n",
+              GNUNET_i2s(&peer),
+              hs,
+              "HELLO");
+#endif
+  tqe = GNUNET_malloc (sizeof (struct TransmissionQueueEntry) + hs);
+  tqe->size = hs;
+  tqe->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
+  memcpy (&tqe[1], hello, hs);
+  GNUNET_CONTAINER_DLL_insert_after (h->tq_head,
+                                     h->tq_tail,
+                                     h->tq_tail,
+                                     tqe);
+  trigger_transmit (h);
 }
 
 
+
+/**
+ *
+ */
 struct GNUNET_PEERINFO_NewIteratorContext
 {
+  /**
+   *
+   */
+  struct GNUNET_PEERINFO_Handle *h;
+
+  /**
+   *
+   */
+  GNUNET_PEERINFO_Processor callback;
+
+  /**
+   *
+   */
+  void *callback_cls;
+
+  /**
+   *
+   */
+  struct GNUNET_TIME_Absolute timeout;
 };
 
 
+
+/**
+ * Type of a function to call when we receive a message
+ * from the service.
+ *
+ * @param cls closure
+ * @param msg message received, NULL on timeout or fatal error
+ */
+static void
+peerinfo_handler (void *cls, const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_PEERINFO_NewIteratorContext *ic = cls;
+  const struct InfoMessage *im;
+  const struct GNUNET_HELLO_Message *hello;
+  uint16_t ms;
+
+  if (msg == NULL)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  _("Failed to receive response from `%s' service.\n"),
+                  "peerinfo");
+      reconnect (ic->h);
+      trigger_transmit (ic->h);
+      ic->callback (ic->callback_cls, NULL, NULL, 1);
+      GNUNET_free (ic);
+      return;
+    }
+  if (ntohs (msg->type) == GNUNET_MESSAGE_TYPE_PEERINFO_INFO_END)
+    {
+#if DEBUG_PEERINFO
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Received end of list of peers from peerinfo database\n");
+#endif
+      trigger_transmit (ic->h);
+      ic->callback (ic->callback_cls, NULL, NULL, 0);
+      GNUNET_free (ic);
+      return;
+    }
+  ms = ntohs (msg->size);
+  if ((ms < sizeof (struct InfoMessage)) ||
+      (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_PEERINFO_INFO))
+    {
+      GNUNET_break (0);
+      reconnect (ic->h);
+      trigger_transmit (ic->h);
+      ic->callback (ic->callback_cls, NULL, NULL, 2);
+      GNUNET_free (ic);
+      return;
+    }
+  im = (const struct InfoMessage *) msg;
+  hello = NULL;
+  if (ms > sizeof (struct InfoMessage) + sizeof (struct GNUNET_MessageHeader))
+    {
+      hello = (const struct GNUNET_HELLO_Message *) &im[1];
+      if (ms != sizeof (struct InfoMessage) + GNUNET_HELLO_size (hello))
+        {
+          GNUNET_break (0);
+          reconnect (ic->h);
+          trigger_transmit (ic->h);
+          ic->callback (ic->callback_cls, NULL, NULL, 2);
+          GNUNET_free (ic);
+          return;
+        }
+    }
+#if DEBUG_PEERINFO
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Received %u bytes of `%s' information about peer `%s' from PEERINFO database\n",
+              (hello == NULL) ? 0 : (unsigned int) GNUNET_HELLO_size (hello),
+              "HELLO",
+              GNUNET_i2s (&im->peer));
+#endif
+  ic->callback (ic->callback_cls, &im->peer, hello, ntohl (im->trust));
+  GNUNET_CLIENT_receive (ic->h->client,
+                         &peerinfo_handler,
+                         ic,
+                         GNUNET_TIME_absolute_get_remaining (ic->timeout));
+}
+
+
+/**
+ *
+ */
+static void
+iterator_start_receive (void *cls,
+                        int transmit_success)
+{
+  struct GNUNET_PEERINFO_NewIteratorContext *ic = cls;
+
+  if (GNUNET_OK != transmit_success)
+    {
+      ic->callback (ic->callback_cls, NULL, NULL, 2);
+      reconnect (ic->h);
+      trigger_transmit (ic->h);
+      GNUNET_free (ic);
+      return;
+    }  
+  GNUNET_CLIENT_receive (ic->h->client,
+                         &peerinfo_handler,
+                         ic,
+                         GNUNET_TIME_absolute_get_remaining (ic->timeout));
+}
+
+
 /**
  * Call a method for each known matching host and change its trust
  * value.  The callback method will be invoked once for each matching
@@ -130,7 +453,51 @@ GNUNET_PEERINFO_iterate_new (struct GNUNET_PEERINFO_Handle *h,
                              GNUNET_PEERINFO_Processor callback,
                              void *callback_cls)
 {
-  return NULL;
+  struct ListAllPeersMessage *lapm;
+  struct ListPeerMessage *lpm;
+  struct GNUNET_PEERINFO_NewIteratorContext *ic;
+  struct TransmissionQueueEntry *tqe;
+
+#if DEBUG_PEERINFO
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Requesting list of peers from peerinfo database\n");
+#endif
+  if (peer == NULL)
+    {
+      tqe = GNUNET_malloc (sizeof (struct TransmissionQueueEntry) +
+                           sizeof (struct ListAllPeersMessage));
+      tqe->size = sizeof (struct ListAllPeersMessage);
+      lapm = (struct ListAllPeersMessage *) &tqe[1];
+      lapm->header.size = htons (sizeof (struct ListAllPeersMessage));
+      lapm->header.type = htons (GNUNET_MESSAGE_TYPE_PEERINFO_GET_ALL);
+      lapm->trust_change = htonl (trust_delta);
+    }
+  else
+    {
+      tqe = GNUNET_malloc (sizeof (struct TransmissionQueueEntry) +
+                           sizeof (struct ListPeerMessage));
+      tqe->size = sizeof (struct ListPeerMessage);
+      lpm = (struct ListPeerMessage *) &tqe[1];
+      lpm->header.size = htons (sizeof (struct ListPeerMessage));
+      lpm->header.type = htons (GNUNET_MESSAGE_TYPE_PEERINFO_GET);
+      lpm->trust_change = htonl (trust_delta);
+      memcpy (&lpm->peer, peer, sizeof (struct GNUNET_PeerIdentity));
+    }
+  ic = GNUNET_malloc (sizeof (struct GNUNET_PEERINFO_NewIteratorContext));
+  ic->callback = callback;
+  ic->callback_cls = callback_cls;
+  ic->timeout = GNUNET_TIME_relative_to_absolute (timeout);
+  tqe->timeout = ic->timeout;
+  tqe->cont = &iterator_start_receive;
+  tqe->cont_cls = ic;
+  /* FIXME: sort DLL by timeout? */
+  /* FIXME: add timeout task!? */
+  GNUNET_CONTAINER_DLL_insert_after (h->tq_head,
+                                     h->tq_tail,
+                                     h->tq_tail,
+                                     tqe);
+  trigger_transmit (h);
+  return ic;
 }
 
 
@@ -143,6 +510,8 @@ GNUNET_PEERINFO_iterate_new (struct GNUNET_PEERINFO_Handle *h,
 void
 GNUNET_PEERINFO_iterate_cancel_new (struct GNUNET_PEERINFO_NewIteratorContext *ic)
 {
+  GNUNET_assert (0); 
+  // FIXME: not implemented
 }