client manager: higher level client API with a transmission queue and message handler registration

2 files changed, 497 insertions, 0 deletions

diff --git a/src/util/Makefile.am b/src/util/Makefile.am
index a33d6f63d..c8c760757 100644
--- a/src/util/Makefile.am
+++ b/src/util/Makefile.am
@@ -73,6 +73,7 @@ libgnunetutil_la_SOURCES = \
   bandwidth.c \
   bio.c \
   client.c \
+  client_manager.c \
   common_allocation.c \
   common_endian.c \
   common_logging.c \
diff --git a/src/util/client_manager.c b/src/util/client_manager.c
new file mode 100644
index 000000000..c415ce845
--- /dev/null
+++ b/src/util/client_manager.c
@@ -0,0 +1,496 @@
+/*
+     This file is part of GNUnet.
+     (C) 2013 Christian Grothoff (and other contributing authors)
+
+     GNUnet is free software; you can redistribute it and/or modify
+     it under the terms of the GNU General Public License as published
+     by the Free Software Foundation; either version 3, or (at your
+     option) any later version.
+
+     GNUnet is distributed in the hope that it will be useful, but
+     WITHOUT ANY WARRANTY; without even the implied warranty of
+     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+     General Public License for more details.
+
+     You should have received a copy of the GNU General Public License
+     along with GNUnet; see the file COPYING.  If not, write to the
+     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+     Boston, MA 02111-1307, USA.
+*/
+
+/**
+ * @file util/client_manager.c
+ * @brief Client manager; higher level client API with transmission queue
+ * and message handler registration.
+ * @author Gabor X Toth
+ */
+
+#include <inttypes.h>
+
+#include "platform.h"
+#include "gnunet_util_lib.h"
+
+#define LOG(kind,...) GNUNET_log_from (kind, "util",__VA_ARGS__)
+
+
+/**
+ * List of arrays of message handlers.
+ */
+struct HandlersListItem
+{
+  struct HandlersListItem *prev;
+  struct HandlersListItem *next;
+
+  /**
+   * NULL-terminated array of handlers.
+   */
+  const struct GNUNET_CLIENT_MANAGER_MessageHandler *handlers;
+};
+
+
+struct MessageQueueItem
+{
+  struct MessageQueueItem *prev;
+  struct MessageQueueItem *next;
+  struct GNUNET_MessageHeader *msg;
+};
+
+
+struct GNUNET_CLIENT_MANAGER_Connection
+{
+  /**
+   * Configuration to use.
+   */
+  const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+  /**
+   * Client connection to service.
+   */
+  struct GNUNET_CLIENT_Connection *client;
+
+  /**
+   * Currently pending transmission request, or NULL for none.
+   */
+  struct GNUNET_CLIENT_TransmitHandle *client_tmit;
+
+  /**
+   * Service name to connect to.
+   */
+  const char *service_name;
+
+  /**
+   * Head of messages to transmit to the service.
+   */
+  struct MessageQueueItem *tmit_head;
+
+  /**
+   * Tail of messages to transmit to the service.
+   */
+  struct MessageQueueItem *tmit_tail;
+
+  /**
+   * Message handlers.
+   */
+  const struct GNUNET_CLIENT_MANAGER_MessageHandler *handlers;
+
+  /**
+   * User context value.
+   * @see GNUNET_CLIENT_MANAGER_set_user_context()
+   * @see GNUNET_CLIENT_MANAGER_get_user_context()
+   */
+  void *user_ctx;
+
+  /**
+   * Last size given when user context was initialized.
+   * Used for sanity check.
+   */
+  size_t user_ctx_size;
+
+  /**
+   * Task doing exponential back-off trying to reconnect.
+   */
+  GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
+
+  /**
+   * Time for next connect retry.
+   */
+  struct GNUNET_TIME_Relative reconnect_delay;
+
+  /**
+   * Are we currently polling for incoming messages?
+   */
+  uint8_t in_receive;
+
+  /**
+   * #GNUNET_YES if GNUNET_CLIENT_MANAGER_disconnect() was called
+   * and we're transmitting the last messages from the queue.
+   */
+  uint8_t disconnecting;
+};
+
+
+/**
+ * Handle received messages from the service.
+ */
+static void
+recv_message (void *cls, const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_CLIENT_MANAGER_Connection *mgr = cls;
+  uint16_t type = 0, size = 0;
+
+  if (NULL != msg)
+  {
+    type = ntohs (msg->type);
+    size = ntohs (msg->size);
+    /* FIXME: decrease reconnect_delay gradually after a successful reconnection */
+  }
+
+  size_t i = 0;
+  while (NULL != mgr->handlers[i].callback)
+  {
+    const struct GNUNET_CLIENT_MANAGER_MessageHandler *mh = &mgr->handlers[i];
+    if ((mh->type == type) || (mh->type == GNUNET_MESSAGE_TYPE_ALL))
+    {
+      if (0 != mh->expected_size
+          && ((GNUNET_NO == mh->is_variable_size && size != mh->expected_size)
+              || (GNUNET_YES == mh->is_variable_size && size < mh->expected_size)))
+      {
+        LOG (GNUNET_ERROR_TYPE_ERROR,
+             "Expected %u bytes for message of type %u, got %u.\n",
+             mh->expected_size, type, size);
+        GNUNET_break_op (0);
+        GNUNET_CLIENT_disconnect (mgr->client);
+        mgr->client = NULL;
+        recv_message (mgr, NULL);
+        break;
+      }
+      mh->callback (mh->callback_cls, mgr, msg);
+    }
+    i++;
+  }
+  if (NULL != mgr->client)
+  {
+    GNUNET_CLIENT_receive (mgr->client, &recv_message, mgr,
+                           GNUNET_TIME_UNIT_FOREVER_REL);
+  }
+}
+
+
+/**
+ * Schedule transmission of the next message from our queue.
+ *
+ * @param mgr  Client manager connection.
+ */
+static void
+transmit_next (struct GNUNET_CLIENT_MANAGER_Connection *mgr);
+
+
+/**
+ * Transmit next message to service.
+ *
+ * @param cls   The struct GNUNET_PSYC_Channel.
+ * @param size  Number of bytes available in @a buf.
+ * @param buf   Where to copy the message.
+ *
+ * @return Number of bytes copied to @a buf.
+ */
+static size_t
+send_next_message (void *cls, size_t buf_size, void *buf)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "send_next_message()\n");
+  struct GNUNET_CLIENT_MANAGER_Connection *mgr = cls;
+
+  if (NULL == buf)
+  {
+    /* disconnected */
+    recv_message (mgr, NULL);
+    return 0;
+  }
+
+  struct MessageQueueItem *mqi = mgr->tmit_head;
+  if (NULL == mqi)
+    return 0;
+
+  uint16_t size = ntohs (mqi->msg->size);
+  mgr->client_tmit = NULL;
+  GNUNET_assert (size <= buf_size);
+  memcpy (buf, mqi->msg, size);
+
+  GNUNET_CONTAINER_DLL_remove (mgr->tmit_head, mgr->tmit_tail, mqi);
+  GNUNET_free (mqi->msg);
+  GNUNET_free (mqi);
+
+  if (NULL != mgr->tmit_head)
+    transmit_next (mgr);
+
+  if (GNUNET_NO == mgr->in_receive)
+  {
+    mgr->in_receive = GNUNET_YES;
+    GNUNET_CLIENT_receive (mgr->client, &recv_message, mgr,
+                           GNUNET_TIME_UNIT_FOREVER_REL);
+  }
+  return size;
+}
+
+
+/**
+ * Schedule transmission of the next message from our queue.
+ *
+ * @param mgr  Client manager connection.
+ */
+static void
+transmit_next (struct GNUNET_CLIENT_MANAGER_Connection *mgr)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "transmit_next()\n");
+  if (NULL != mgr->client_tmit || NULL == mgr->client)
+    return;
+
+  if (NULL == mgr->tmit_head)
+  {
+    if (GNUNET_YES == mgr->disconnecting)
+      GNUNET_CLIENT_MANAGER_disconnect (mgr, GNUNET_NO);
+    return;
+  }
+
+  mgr->client_tmit
+    = GNUNET_CLIENT_notify_transmit_ready (mgr->client,
+                                           ntohs (mgr->tmit_head->msg->size),
+                                           GNUNET_TIME_UNIT_FOREVER_REL,
+                                           GNUNET_NO,
+                                           &send_next_message,
+                                           mgr);
+}
+
+
+/**
+ * Try again to connect to the service.
+ *
+ * @param cls Channel handle.
+ * @param tc Scheduler context.
+ */
+static void
+reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct GNUNET_CLIENT_MANAGER_Connection *mgr = cls;
+  mgr->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Connecting to %s service.\n", mgr->service_name);
+  GNUNET_assert (NULL == mgr->client);
+  mgr->client = GNUNET_CLIENT_connect (mgr->service_name, mgr->cfg);
+  GNUNET_assert (NULL != mgr->client);
+
+  transmit_next (mgr);
+}
+
+
+/**
+ * Connect to service.
+ *
+ * @param cfg           Configuration to use.
+ * @param service_name  Service name to connect to.
+ * @param handlers      Message handlers.
+ *
+ * @return Client manager connection handle.
+ */
+struct GNUNET_CLIENT_MANAGER_Connection *
+GNUNET_CLIENT_MANAGER_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
+                               const char *service_name,
+                               const struct
+                               GNUNET_CLIENT_MANAGER_MessageHandler *handlers)
+{
+  struct GNUNET_CLIENT_MANAGER_Connection *
+    mgr = GNUNET_malloc (sizeof (*mgr));
+  mgr->cfg = cfg;
+  mgr->service_name = service_name;
+  mgr->handlers = handlers;
+  mgr->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
+  mgr->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, mgr);
+  return mgr;
+}
+
+
+/**
+ * Disconnect from the service.
+ *
+ * @param mgr             Client manager connection.
+ * @param transmit_queue  Transmit pending messages in queue before disconnecting.
+ */
+void
+GNUNET_CLIENT_MANAGER_disconnect (struct GNUNET_CLIENT_MANAGER_Connection *mgr,
+                                  int transmit_queue)
+{
+  if (NULL != mgr->tmit_head)
+  {
+    if (GNUNET_YES == transmit_queue)
+    {
+      mgr->disconnecting = GNUNET_YES;
+      transmit_next (mgr);
+    }
+    else
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "Disconnecting while there are still messages "
+           "in the transmission queue.\n");
+      GNUNET_CLIENT_MANAGER_drop_queue (mgr);
+    }
+  }
+  if (mgr->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
+  {
+    GNUNET_SCHEDULER_cancel (mgr->reconnect_task);
+    mgr->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+  }
+  if (NULL != mgr->client_tmit)
+  {
+    GNUNET_CLIENT_notify_transmit_ready_cancel (mgr->client_tmit);
+    mgr->client_tmit = NULL;
+  }
+  if (NULL != mgr->client)
+  {
+    GNUNET_CLIENT_disconnect (mgr->client);
+    mgr->client = NULL;
+  }
+  GNUNET_free (mgr);
+}
+
+
+/**
+ * Reschedule connect to the service using exponential back-off.
+ *
+ * @param mgr  Client manager connection.
+ */
+void
+GNUNET_CLIENT_MANAGER_reconnect (struct GNUNET_CLIENT_MANAGER_Connection *mgr)
+{
+  if (GNUNET_SCHEDULER_NO_TASK != mgr->reconnect_task)
+    return;
+
+  if (NULL != mgr->client_tmit)
+  {
+    GNUNET_CLIENT_notify_transmit_ready_cancel (mgr->client_tmit);
+    mgr->client_tmit = NULL;
+  }
+  if (NULL != mgr->client)
+  {
+    GNUNET_CLIENT_disconnect (mgr->client);
+    mgr->client = NULL;
+  }
+  mgr->in_receive = GNUNET_NO;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Scheduling task to reconnect to service in %s.\n",
+       GNUNET_STRINGS_relative_time_to_string (mgr->reconnect_delay, GNUNET_YES));
+  mgr->reconnect_task =
+      GNUNET_SCHEDULER_add_delayed (mgr->reconnect_delay, &reconnect, mgr);
+  mgr->reconnect_delay = GNUNET_TIME_STD_BACKOFF (mgr->reconnect_delay);
+}
+
+
+/**
+ * Add a message to the end of the transmission queue.
+ *
+ * @param mgr  Client manager connection.
+ * @param msg  Message to transmit.  It is free()'d after transmission.
+ */
+void
+GNUNET_CLIENT_MANAGER_transmit (struct GNUNET_CLIENT_MANAGER_Connection *mgr,
+                                struct GNUNET_MessageHeader *msg)
+{
+  struct MessageQueueItem *mqi = GNUNET_malloc (sizeof (*mqi));
+  mqi->msg = msg;
+  GNUNET_CONTAINER_DLL_insert_tail (mgr->tmit_head, mgr->tmit_tail, mqi);
+  transmit_next (mgr);
+}
+
+
+/**
+ * Add a message to the beginning of the transmission queue.
+ *
+ * @param mgr  Client manager connection.
+ * @param msg  Message to transmit.  It is free()'d after transmission.
+ */
+void
+GNUNET_CLIENT_MANAGER_transmit_now (struct GNUNET_CLIENT_MANAGER_Connection *mgr,
+                                    struct GNUNET_MessageHeader *msg)
+{
+  struct MessageQueueItem *mqi = GNUNET_malloc (sizeof (*mqi));
+  mqi->msg = msg;
+  GNUNET_CONTAINER_DLL_insert (mgr->tmit_head, mgr->tmit_tail, mqi);
+  transmit_next (mgr);
+}
+
+
+/**
+ * Drop all queued messages.
+ *
+ * @param mgr  Client manager connection.
+ */
+void
+GNUNET_CLIENT_MANAGER_drop_queue (struct GNUNET_CLIENT_MANAGER_Connection *mgr)
+{
+  struct MessageQueueItem *cur, *next = mgr->tmit_head;
+  while (NULL != next)
+  {
+    cur = next;
+    next = cur->next;
+    GNUNET_free (cur->msg);
+    GNUNET_free (cur);
+  }
+}
+
+
+/**
+ * Obtain client connection handle.
+ *
+ * @param mgr  Client manager connection handle.
+ *
+ * @return Client connection handle.
+ */
+struct GNUNET_CLIENT_Connection *
+GNUNET_CLIENT_MANAGER_get_client (struct GNUNET_CLIENT_MANAGER_Connection *mgr)
+{
+  return mgr->client;
+}
+
+
+/**
+ * Return user context associated with the given client.
+ * Note: you should probably use the macro (call without the underscore).
+ *
+ * @param mgr  Client manager connection.
+ * @param size Number of bytes in user context struct (for verification only).
+ * @return User context.
+ */
+void *
+GNUNET_CLIENT_MANAGER_get_user_context_ (struct GNUNET_CLIENT_MANAGER_Connection *mgr,
+                                         size_t size)
+{
+  if ((0 == mgr->user_ctx_size) &&
+      (NULL == mgr->user_ctx))
+    return NULL; /* never set */
+  GNUNET_assert (size == mgr->user_ctx_size);
+  return mgr->user_ctx;
+}
+
+
+/**
+ * Set user context to be associated with the given client.
+ * Note: you should probably use the macro (call without the underscore).
+ *
+ * @param mgr  Client manager connection.
+ * @param ctx  User context.
+ * @param size Number of bytes in user context struct (for verification only).
+ */
+void
+GNUNET_CLIENT_MANAGER_set_user_context_ (struct GNUNET_CLIENT_MANAGER_Connection *mgr,
+                                         void *ctx,
+                                         size_t size)
+{
+  if (NULL == ctx)
+  {
+    mgr->user_ctx_size = 0;
+    mgr->user_ctx = ctx;
+    return;
+  }
+  mgr->user_ctx_size = size;
+  mgr->user_ctx = ctx;
+}