started implementing consensus api and service

8 files changed, 850 insertions, 28 deletions

diff --git a/src/consensus/Makefile.am b/src/consensus/Makefile.am
index 5929a4d97..4eb06ceea 100644
--- a/src/consensus/Makefile.am
+++ b/src/consensus/Makefile.am
@@ -21,6 +21,9 @@ bin_PROGRAMS = \
 libexec_PROGRAMS = \
  gnunet-service-consensus
 
+lib_LTLIBRARIES = \
+  libgnunetconsensus.la
+
 gnunet_consensus_SOURCES = \
  gnunet-consensus.c         
 gnunet_consensus_LDADD = \
@@ -33,16 +36,28 @@ gnunet_service_consensus_LDADD = \
   $(top_builddir)/src/util/libgnunetutil.la \
   $(GN_LIBINTL) 
 
+libgnunetconsensus_la_SOURCES = \
+  consensus_api.c
+libgnunetconsensus_la_LIBADD = \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(LTLIBINTL)
+libgnunetconsensus_la_LDFLAGS = \
+  $(GN_LIB_LDFLAGS)
+
+check_PROGRAMS = \
+ test_consensus_api
 
-#check_PROGRAMS = \
-# test_consensus_api
+if ENABLE_TEST_RUN
+TESTS = $(check_PROGRAMS)
+endif
+
+test_consensus_api_SOURCES = \
+ test_consensus_api.c
+test_consensus_api_LDADD = \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/testing/libgnunettesting.la \
+  $(top_builddir)/src/consensus/libgnunetconsensus.la
 
-#if ENABLE_TEST_RUN
-#TESTS = $(check_PROGRAMS)
-#endif
-#
-#test_consensus_api_SOURCES = \
-# test_consensus_api.c
-#test_consensus_api_LDADD = \
-#  $(top_builddir)/src/util/libgnunetutil.la  
 
+EXTRA_DIST = \
+  test_consensus.conf
diff --git a/src/consensus/consensus.h b/src/consensus/consensus.h
new file mode 100644
index 000000000..3f1efe340
--- /dev/null
+++ b/src/consensus/consensus.h
@@ -0,0 +1,95 @@
+/*
+     This file is part of GNUnet.
+     (C) 2012 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 2, 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.
+*/
+
+/**
+ * @author Florian Dold
+ * @file consensus/consensus.h
+ * @brief
+ */
+#ifndef NSE_H
+#define NSE_H
+
+#include "gnunet_common.h"
+
+GNUNET_NETWORK_STRUCT_BEGIN
+
+struct GNUNET_CONSENSUS_JoinMessage
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN
+   */
+  struct GNUNET_MessageHeader header;
+
+  struct GNUNET_HashCode session_id;
+
+  uint16_t num_peers;
+
+  /* GNUNET_PeerIdentity[num_peers] */
+};
+
+
+struct GNUNET_CONSENSUS_ConcludeMessage
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE
+   */
+  struct GNUNET_MessageHeader header;
+
+  struct GNUNET_TIME_RelativeNBO timeout;
+};
+
+
+struct GNUNET_CONSENSUS_ConcludeDoneMessage
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint16_t num_peers;
+
+  /** PeerIdentity[num_peers] */
+};
+
+
+/**
+ * Message with an element
+ */
+struct GNUNET_CONSENSUS_ElementMessage
+{
+
+  /**
+   * Type:
+   * Either GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT
+   * or GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT_ELEMENT
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_NEW_ELEMENT
+   */
+  uint16_t element_type;
+
+  /* rest: element data */
+};
+
+GNUNET_NETWORK_STRUCT_END
+
+#endif
diff --git a/src/consensus/consensus_api.c b/src/consensus/consensus_api.c
new file mode 100644
index 000000000..0d7e6c8e4
--- /dev/null
+++ b/src/consensus/consensus_api.c
@@ -0,0 +1,502 @@
+/*
+     This file is part of GNUnet.
+     (C) 2012 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 consensus/consensus_api.c
+ * @brief 
+ * @author Florian Dold
+ */
+#include "platform.h"
+#include "gnunet_protocols.h"
+#include "gnunet_client_lib.h"
+#include "gnunet_consensus_service.h"
+#include "consensus.h"
+
+
+#define LOG(kind,...) GNUNET_log_from (kind, "consensus-api",__VA_ARGS__)
+
+
+/**
+ * Handle for the service.
+ */
+struct GNUNET_CONSENSUS_Handle
+{
+  /**
+   * Configuration to use.
+   */
+  const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+  /**
+   * Socket (if available).
+   */
+  struct GNUNET_CLIENT_Connection *client;
+
+  /**
+   * Callback for new elements. Not called for elements added locally.
+   */
+  GNUNET_CONSENSUS_NewElementCallback new_element_cb;
+
+  /**
+   * Closure for new_element_cb
+   */
+  void *new_element_cls;
+
+  /**
+   * Session identifier for the consensus session.
+   */
+  struct GNUNET_HashCode session_id;
+
+  /**
+   * Number of peers in the consensus. Optionally includes the local peer.
+   */
+  int num_peers;
+
+  /**
+   * Peer identities of peers in the consensus. Optionally includes the local peer.
+   */
+  struct GNUNET_PeerIdentity *peers;
+
+  /**
+   * Currently active transmit request.
+   */
+  struct GNUNET_CLIENT_TransmitHandle *th;
+
+  /**
+   * GNUNES_YES iff the join message has been sent to the service.
+   */
+  int joined;
+
+  /**
+   * Called when the current insertion operation finishes.
+   * NULL if there is no insert operation active.
+   */
+  GNUNET_CONSENSUS_InsertDoneCallback idc;
+
+  /**
+   * Closure for the insert done callback.
+   */
+  void *idc_cls;
+
+  /**
+   * An element that was requested to be inserted.
+   */
+  struct GNUNET_CONSENSUS_Element *insert_element;
+
+  /**
+   * Called when the conclude operation finishes or fails.
+   */
+  GNUNET_CONSENSUS_ConcludeCallback conclude_cb;
+
+  /**
+   * Closure for the conclude callback.
+   */
+  void *conclude_cls;
+
+  /**
+   * Deadline for the conclude operation.
+   */
+  struct GNUNET_TIME_Absolute conclude_deadline;
+};
+
+
+static void
+handle_new_element(struct GNUNET_CONSENSUS_Handle *consensus,
+                   struct GNUNET_CONSENSUS_ElementMessage *msg)
+{
+  struct GNUNET_CONSENSUS_Element element;
+  element.type = msg->element_type;
+  element.size = msg->header.size - sizeof (struct GNUNET_CONSENSUS_ElementMessage);
+  element.data = &msg[1];
+  consensus->new_element_cb(consensus->new_element_cls, &element);
+}
+
+static void
+handle_conclude_done(struct GNUNET_CONSENSUS_Handle *consensus,
+                     struct GNUNET_CONSENSUS_ConcludeDoneMessage *msg)
+{
+  GNUNET_assert (NULL != consensus->conclude_cb);
+  consensus->conclude_cb(consensus->conclude_cls,
+                         msg->num_peers,
+                         (struct GNUNET_PeerIdentity *) &msg[1]);
+  consensus->conclude_cb = NULL;
+}
+
+
+
+/**
+ * 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
+message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_CONSENSUS_Handle *consensus = cls;
+  GNUNET_CONSENSUS_InsertDoneCallback idc;
+  void *idc_cls;
+
+  if (msg == NULL)
+  {
+    /* Error, timeout, death */
+    GNUNET_CLIENT_disconnect (consensus->client);
+    consensus->client = NULL;
+    consensus->new_element_cb(NULL, NULL);
+    if (NULL != consensus->idc)
+    {
+      consensus->idc(consensus->idc_cls, GNUNET_NO);
+      consensus->idc = NULL;
+      consensus->idc_cls = NULL;
+    }
+    return;
+  }
+
+  switch (ntohs(msg->type))
+  {
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT_ACK:
+      idc = consensus->idc;
+      consensus->idc = NULL;
+      idc_cls = consensus->idc_cls;
+      consensus->idc_cls = NULL;
+      idc(idc_cls, GNUNET_YES);
+      break;
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT:
+      handle_new_element(consensus, (struct GNUNET_CONSENSUS_ElementMessage *) msg);
+      break;
+    case GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE:
+      handle_conclude_done(consensus, (struct GNUNET_CONSENSUS_ConcludeDoneMessage *) msg);
+      break;
+    default:
+      LOG(GNUNET_ERROR_TYPE_WARNING, "did not understand message type sent by service, ignoring");
+  }
+  GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
+                         GNUNET_TIME_UNIT_FOREVER_REL);
+}
+
+
+
+
+/**
+ * Function called to notify a client about the connection
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the connection was closed for
+ * writing in the meantime.
+ *
+ * @param cls closure
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+transmit_insert (void *cls, size_t size, void *buf)
+{
+  struct GNUNET_CONSENSUS_ElementMessage *msg;
+  struct GNUNET_CONSENSUS_Handle *consensus;
+  int msize;
+
+  GNUNET_assert (NULL != buf);
+
+  consensus = cls;
+
+  GNUNET_assert (NULL != consensus->insert_element);
+
+  consensus->th = NULL;
+
+
+  msg = buf;
+
+  msize = sizeof (struct GNUNET_CONSENSUS_ElementMessage) +
+      consensus->insert_element->size;
+
+  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT);
+  msg->header.size = htons (msize);
+  memcpy(&msg[1],
+         consensus->insert_element->data,
+         consensus->insert_element->size);
+
+  return msize;
+}
+
+
+/**
+ * Function called to notify a client about the connection
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the connection was closed for
+ * writing in the meantime.
+ *
+ * @param cls closure
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+transmit_join (void *cls, size_t size, void *buf)
+{
+  struct GNUNET_CONSENSUS_JoinMessage *msg;
+  struct GNUNET_CONSENSUS_Handle *consensus;
+  int msize;
+
+  LOG(GNUNET_ERROR_TYPE_DEBUG, "transmitting CLIENT_JOIN to service\n");
+
+  GNUNET_assert (NULL != buf);
+
+  consensus = cls;
+  consensus->th = NULL;
+  consensus->joined = 1;
+
+  msg = buf;
+
+  msize = sizeof (struct GNUNET_CONSENSUS_JoinMessage) +
+      consensus->num_peers * sizeof (struct GNUNET_PeerIdentity);
+
+  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN);
+  msg->header.size = htons (msize);
+  msg->session_id = consensus->session_id;
+  msg->num_peers = htons (consensus->num_peers);
+  memcpy(&msg[1],
+         consensus->peers,
+         consensus->num_peers * sizeof (struct GNUNET_PeerIdentity));
+
+  if (consensus->insert_element != NULL)
+  {
+    consensus->th =
+        GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                             msize,
+                                             GNUNET_TIME_UNIT_FOREVER_REL,
+                                             GNUNET_NO, &transmit_insert, consensus);
+  }
+
+
+  GNUNET_CLIENT_receive (consensus->client, &message_handler, consensus,
+                         GNUNET_TIME_UNIT_FOREVER_REL);
+  
+  return msize;
+}
+
+
+/**
+ * Function called to notify a client about the connection
+ * begin ready to queue more data.  "buf" will be
+ * NULL and "size" zero if the connection was closed for
+ * writing in the meantime.
+ *
+ * @param cls closure
+ * @param size number of bytes available in buf
+ * @param buf where the callee should write the message
+ * @return number of bytes written to buf
+ */
+static size_t
+transmit_conclude (void *cls, size_t size, void *buf)
+{
+  struct GNUNET_CONSENSUS_ConcludeMessage *msg;
+  struct GNUNET_CONSENSUS_Handle *consensus;
+  int msize;
+
+  GNUNET_assert (NULL != buf);
+
+  consensus = cls;
+  consensus->th = NULL;
+
+  msg = buf;
+
+  msize = sizeof (struct GNUNET_CONSENSUS_ConcludeMessage);
+
+  msg->header.type = htons (GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE);
+  msg->header.size = htons (msize);
+  msg->timeout =
+      GNUNET_TIME_relative_hton(GNUNET_TIME_absolute_get_remaining(consensus->conclude_deadline));
+
+  return msize;
+}
+
+
+
+
+/**
+ * Create a consensus session.
+ *
+ * @param cfg
+ * @param num_peers
+ * @param peers array of peers participating in this consensus session
+ *              Inclusion of the local peer is optional.
+ * @param session_id session identifier
+ *                   Allows a group of peers to have more than consensus session.
+ * @param num_initial_elements number of entries in the 'initial_elements' array
+ * @param initial_elements our elements for the consensus (each of 'element_size'
+ * @param new_element callback, called when a new element is added to the set by
+ *                    another peer
+ * @param new_element_cls closure for new_element
+ * @return handle to use, NULL on error
+ */
+struct GNUNET_CONSENSUS_Handle *
+GNUNET_CONSENSUS_create (const struct GNUNET_CONFIGURATION_Handle *cfg,
+                         unsigned int num_peers,
+                         const struct GNUNET_PeerIdentity *peers,
+                         const struct GNUNET_HashCode *session_id,
+                         /*
+                         unsigned int num_initial_elements,
+                         const struct GNUNET_CONSENSUS_Element **initial_elements,
+                         */
+                         GNUNET_CONSENSUS_NewElementCallback new_element,
+                         void *new_element_cls)
+{
+  struct GNUNET_CONSENSUS_Handle *consensus;
+  size_t join_message_size;
+
+
+  consensus = GNUNET_malloc (sizeof (struct GNUNET_CONSENSUS_Handle));
+  consensus->cfg = cfg;
+  consensus->new_element_cb = new_element;
+  consensus->new_element_cls = new_element_cls;
+  consensus->num_peers = num_peers;
+  consensus->session_id = *session_id;
+
+
+
+  if (0 == num_peers)
+  {
+    consensus->peers = NULL;
+  }
+  else if (num_peers > 0)
+  {
+
+    consensus->peers = GNUNET_memdup (peers, num_peers * sizeof (struct GNUNET_PeerIdentity));
+  }
+  else
+  {
+    GNUNET_break (0);
+  }
+
+
+  consensus->client = GNUNET_CLIENT_connect ("consensus", cfg);
+
+  GNUNET_assert (consensus->client != NULL);
+
+  join_message_size = (sizeof (struct GNUNET_CONSENSUS_JoinMessage)) +
+      (num_peers * sizeof (struct GNUNET_PeerIdentity));
+
+  consensus->th =
+      GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                           join_message_size,
+                                           GNUNET_TIME_UNIT_FOREVER_REL,
+                                           GNUNET_NO, &transmit_join, consensus);
+
+
+  GNUNET_assert (consensus->th != NULL);
+
+  return consensus;
+}
+
+
+
+/**
+ * Insert an element in the set being reconsiled.  Must not be called after
+ * "GNUNET_CONSENSUS_conclude".
+ *
+ * @param consensus handle for the consensus session
+ * @param element the element to be inserted
+ * @param idc function called when we are done with this element and it 
+ *            is thus allowed to call GNUNET_CONSENSUS_insert again
+ * @param idc_cls closure for 'idc'
+ */
+void
+GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
+                         const struct GNUNET_CONSENSUS_Element *element,
+                         GNUNET_CONSENSUS_InsertDoneCallback idc,
+                         void *idc_cls)
+{
+
+  GNUNET_assert (NULL == consensus->idc);
+  GNUNET_assert (NULL == consensus->insert_element);
+
+  consensus->idc = idc;
+  consensus->idc_cls = idc_cls;
+  consensus->insert_element = GNUNET_memdup(element, sizeof (struct GNUNET_CONSENSUS_Element) + element->size);
+
+  if (consensus->joined == 0)
+  {
+    GNUNET_assert (NULL != consensus->th);
+    return;
+  }
+
+  GNUNET_assert (NULL == consensus->th);
+
+  consensus->th =
+      GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                           element->size + sizeof (struct GNUNET_CONSENSUS_ElementMessage),
+                                           GNUNET_TIME_UNIT_FOREVER_REL,
+                                           GNUNET_NO, &transmit_insert, consensus);
+}
+
+
+/**
+ * We are finished inserting new elements into the consensus;
+ * try to conclude the consensus within a given time window.
+ *
+ * @param consensus consensus session
+ * @param timeout timeout after which the conculde callback
+ *                must be called
+ * @param conclude called when the conclusion was successful
+ * @param conclude_cls closure for the conclude callback
+ */
+void
+GNUNET_CONSENSUS_conclude (struct GNUNET_CONSENSUS_Handle *consensus,
+                           struct GNUNET_TIME_Relative timeout,
+                           GNUNET_CONSENSUS_ConcludeCallback conclude,
+                           void *conclude_cls)
+{
+  GNUNET_assert (NULL == consensus->th);
+  GNUNET_assert (NULL == consensus->conclude_cb);
+
+  consensus->conclude_cls = conclude_cls;
+  consensus->conclude_cb = conclude;
+  consensus->conclude_deadline = GNUNET_TIME_relative_to_absolute(timeout);
+
+  consensus->th =
+      GNUNET_CLIENT_notify_transmit_ready (consensus->client,
+                                           sizeof (struct GNUNET_CONSENSUS_ConcludeMessage),
+                                           timeout,
+                                           GNUNET_NO, &transmit_conclude, consensus);
+  if (NULL == consensus->th)
+  {
+    conclude(conclude_cls, 0, NULL);
+  }
+}
+
+
+/**
+ * Destroy a consensus handle (free all state associated with
+ * it, no longer call any of the callbacks).
+ *
+ * @param consensus handle to destroy
+ */
+void
+GNUNET_CONSENSUS_destroy (struct GNUNET_CONSENSUS_Handle *consensus)
+{
+  if (consensus->client != NULL)
+  {
+    GNUNET_CLIENT_disconnect (consensus->client);
+    consensus->client = NULL;
+  }
+  GNUNET_free (consensus->peers);
+  GNUNET_free (consensus);
+}
+
diff --git a/src/consensus/gnunet-service-consensus.cc b/src/consensus/gnunet-service-consensus.cc
index 82ea5ef79..9e8aba6bb 100644
--- a/src/consensus/gnunet-service-consensus.cc
+++ b/src/consensus/gnunet-service-consensus.cc
@@ -24,23 +24,87 @@
 #include <stdint.h>
 
 #include "platform.h"
+#include "gnunet_protocols.h"
 #include "gnunet_common.h"
 #include "gnunet_service_lib.h"
+#include "gnunet_consensus_service.h"
+#include "consensus.h"
 
 using namespace std;
 
+
+
+struct ConsensusSession
+{
+
+};
+
+
+
+struct ConsensusClient
+{
+
+};
+
+
+
+/**
+ * Called when a client wants to join a consensus session.
+ */
+void
+client_join (void *cls,
+             struct GNUNET_SERVER_Client *client,
+             const struct GNUNET_MessageHeader *message)
+{
+  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "client joined");
+}
+
+
 /**
- * Process statistics requests.
+ * Called when a client performs an insert operation.
+ */
+void
+client_insert (void *cls,
+             struct GNUNET_SERVER_Client *client,
+             const struct GNUNET_MessageHeader *message)
+{
+
+}
+
+
+
+/**
+ * Called when a client performs the conclude operation.
+ */
+void
+client_conclude (void *cls,
+             struct GNUNET_SERVER_Client *client,
+             const struct GNUNET_MessageHeader *message)
+{
+
+}
+
+
+
+/**
+ * Process consensus requests.
  *
  * @param cls closure
  * @param server the initialized server
  * @param c configuration to use
  */
 static void
-run (void *cls, struct GNUNET_SERVER_Handle *server,
-     const struct GNUNET_CONFIGURATION_Handle *c)
+run (void *cls, struct GNUNET_SERVER_Handle *server, const struct GNUNET_CONFIGURATION_Handle *c)
 {
-  /* TODO */
+  static const struct GNUNET_SERVER_MessageHandler handlers[] = {
+    {&client_join, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN, 0},
+    {&client_insert, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT, 0},
+    {&client_conclude, NULL, GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE,
+        sizeof (struct GNUNET_CONSENSUS_ConcludeMessage)},
+    {NULL, NULL, 0, 0}
+  };
+
+  GNUNET_SERVER_add_handlers(server, handlers);
 }
 
 
@@ -55,6 +119,6 @@ int
 main (int argc, char *const *argv)
 {
   return (GNUNET_OK ==
-          GNUNET_SERVICE_run (argc, argv, "statistics", GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;
+          GNUNET_SERVICE_run (argc, argv, "consensus", GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;
 }
 
diff --git a/src/consensus/test_consensus.conf b/src/consensus/test_consensus.conf
new file mode 100644
index 000000000..143313ff2
--- /dev/null
+++ b/src/consensus/test_consensus.conf
@@ -0,0 +1,15 @@
+[consensus]
+AUTOSTART = YES
+# PORT = 2103
+HOSTNAME = localhost
+HOME = $SERVICEHOME
+BINARY = gnunet-service-consensus
+ACCEPT_FROM = 127.0.0.1;
+ACCEPT_FROM6 = ::1;
+UNIXPATH = /tmp/gnunet-service-consensus.sock
+UNIX_MATCH_UID = YES
+UNIX_MATCH_GID = YES
+OPTIONS = -LDEBUG
+
+[transport]
+OPTIONS = -LERROR
diff --git a/src/consensus/test_consensus_api.c b/src/consensus/test_consensus_api.c
new file mode 100644
index 000000000..a6bd60119
--- /dev/null
+++ b/src/consensus/test_consensus_api.c
@@ -0,0 +1,73 @@
+/*
+     This file is part of GNUnet.
+     (C) 2012 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 consensus/test_consensus_api.c
+ * @brief testcase for consensus_api.c
+ */
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_consensus_service.h"
+#include "gnunet_testing_lib-new.h"
+
+
+static struct GNUNET_CONSENSUS_Handle *consensus;
+
+static struct GNUNET_HashCode session_id;
+
+
+void
+on_new_element (void *cls,
+                struct GNUNET_CONSENSUS_Element *element)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "received new element\n");
+}
+
+
+static void
+run (void *cls, 
+     const struct GNUNET_CONFIGURATION_Handle *cfg,
+     struct GNUNET_TESTING_Peer *peer)
+{
+  char *str = "foo";
+
+  GNUNET_CRYPTO_hash (str, strlen (str), &session_id);
+  consensus = GNUNET_CONSENSUS_create (cfg, 0, NULL, &session_id, on_new_element, cls);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connecting to consensus service.\n");
+  GNUNET_assert (consensus != NULL);
+}
+
+
+int
+main (int argc, char **argv)
+{
+  int ret;
+
+  GNUNET_log_setup ("test_consensus_api",
+                    "DEBUG",
+                    NULL);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "testing consensus api\n");
+
+  ret = GNUNET_TESTING_peer_run ("test_consensus_api",
+                                 "test_consensus.conf",
+                                 &run, NULL);
+  return ret;
+}
+
diff --git a/src/include/gnunet_consensus_service.h b/src/include/gnunet_consensus_service.h
index a2bf69021..8e2d8f572 100644
--- a/src/include/gnunet_consensus_service.h
+++ b/src/include/gnunet_consensus_service.h
@@ -38,20 +38,42 @@ extern "C"
 #include "platform.h"
 #include "gnunet_common.h"
 #include "gnunet_time_lib.h"
+#include "gnunet_configuration_lib.h"
 
 
 /**
- * Called when a new element was received from another peer; elements
- * given to a consensus operation by the local peer are NOT given
+ * An element of the consensus set.
+ */
+struct GNUNET_CONSENSUS_Element
+{
+  /**
+   * The actual data of the element.
+   */
+   const void *data;
+
+   /**
+    * Size of the element's data.
+    */
+   uint16_t size;
+
+   /**
+    * Application specific element type
+    */
+   uint16_t type;
+};
+
+
+/**
+ * Called when a new element was received from another peer, or an error occured.
+ *
+ * Elements given to a consensus operation by the local peer are NOT given
  * to this callback.
  *
  * @param cls closure
- * @param element_size will match the size given to GNUNET_CONSENSUS_create
- * @param element
+ * @param element new element, NULL on error
  */
 typedef void (*GNUNET_CONSENSUS_NewElementCallback) (void *cls,
-                                                     size_t element_size,
-                                                     const void *element);
+                                                     struct GNUNET_CONSENSUS_Element *element);
 
 
 
@@ -67,9 +89,9 @@ struct GNUNET_CONSENSUS_Handle;
  * @param cfg
  * @param num_peers
  * @param peers array of peers participating in this consensus session
+ *              Inclusion of the local peer is optional.
  * @param session_id session identifier
  *                   Allows a group of peers to have more than consensus session.
- * @param element_size size of the elements in the reconciled set in bytes
  * @param num_initial_elements number of entries in the 'initial_elements' array
  * @param initial_elements our elements for the consensus (each of 'element_size'
  * @param new_element callback, called when a new element is added to the set by
@@ -82,9 +104,10 @@ GNUNET_CONSENSUS_create (const struct GNUNET_CONFIGURATION_Handle *cfg,
                          unsigned int num_peers,
                          const struct GNUNET_PeerIdentity *peers,
                          const struct GNUNET_HashCode *session_id,
-                         size_t element_size,
+                         /*
                          unsigned int num_initial_elements,
-                         const void **initial_elements,
+                         const struct GNUNET_CONSENSUS_Element **initial_elements,
+                         */
                          GNUNET_CONSENSUS_NewElementCallback new_element,
                          void *new_element_cls);
 
@@ -107,7 +130,6 @@ typedef void (*GNUNET_CONSENSUS_InsertDoneCallback) (void *cls,
  * "GNUNET_CONSENSUS_conclude".
  *
  * @param consensus handle for the consensus session
- * @param element_size must match element size from GNUNET_CONSENSUS_create
  * @param element the element to be inserted
  * @param idc function called when we are done with this element and it 
  *            is thus allowed to call GNUNET_CONSENSUS_insert again
@@ -115,8 +137,7 @@ typedef void (*GNUNET_CONSENSUS_InsertDoneCallback) (void *cls,
  */
 void
 GNUNET_CONSENSUS_insert (struct GNUNET_CONSENSUS_Handle *consensus,
-                         size_t element_size,
-                         const void *element,
+                         const struct GNUNET_CONSENSUS_Element *element,
                          GNUNET_CONSENSUS_InsertDoneCallback idc,
                          void *idc_cls);
 
diff --git a/src/include/gnunet_protocols.h b/src/include/gnunet_protocols.h
index b9102f429..21ebbfdfe 100644
--- a/src/include/gnunet_protocols.h
+++ b/src/include/gnunet_protocols.h
@@ -1606,8 +1606,45 @@ extern "C"
 #define GNUNET_MESSAGE_TYPE_GNS_GET_AUTH_RESULT 505
 
 
+/*******************************************************************************
+ * CONSENSUS message types
+ ******************************************************************************/
+
+/**
+ * Join a consensus session. Sent by client to service as first message.
+ */
+#define GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_JOIN 520
+
+/**
+ * Insert an element. Sent by client to service.
+ */
+#define GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT 521
+
+/**
+ * The has accepted the inserted element and is ready to receive new elements.
+ * Sent by service to client.
+ */
+#define GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_INSERT_ACK 522
+
+/**
+ * Sent by service when a new element is added.
+ */
+#define GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_RECEIVED_ELEMENT 523
+
+/**
+ * Sent by client to service in order to start the consensus conclusion.
+ */
+#define GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE 524
+
+/**
+ * Sent by service to client in order to signal a completed consensus conclusion.
+ * Last message sent in a consensus session.
+ */
+#define GNUNET_MESSAGE_TYPE_CONSENSUS_CLIENT_CONCLUDE_DONE 524
+
+
 /**
- *  Next available: 520
+ *  Next available: 570
  */