initial import from gnunet.git

10 files changed, 6364 insertions, 0 deletions

diff --git a/src/psyc/.gitignore b/src/psyc/.gitignore
new file mode 100644
index 0000000..14a1753
--- /dev/null
+++ b/src/psyc/.gitignore
@@ -0,0 +1,2 @@
+gnunet-service-psyc
+test_psyc
diff --git a/src/psyc/Makefile.am b/src/psyc/Makefile.am
new file mode 100644
index 0000000..511e3e3
--- /dev/null
+++ b/src/psyc/Makefile.am
@@ -0,0 +1,77 @@
+# This Makefile.am is in the public domain
+AM_CPPFLAGS = -I$(top_srcdir)/src/include
+
+pkgcfgdir= $(pkgdatadir)/config.d/
+
+libexecdir= $(pkglibdir)/libexec/
+
+pkgcfg_DATA = \
+  psyc.conf
+
+
+if MINGW
+ WINFLAGS = -Wl,--no-undefined -Wl,--export-all-symbols
+endif
+
+if USE_COVERAGE
+  AM_CFLAGS = --coverage -O0
+  XLIB = -lgcov
+endif
+
+lib_LTLIBRARIES = libgnunetpsyc.la
+
+libgnunetpsyc_la_SOURCES = \
+  psyc_api.c psyc.h
+libgnunetpsyc_la_LIBADD = \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/psycutil/libgnunetpsycutil.la \
+  $(GN_LIBINTL) $(XLIB)
+libgnunetpsyc_la_LDFLAGS = \
+  $(GN_LIB_LDFLAGS)  $(WINFLAGS) \
+  -version-info 0:0:0
+
+bin_PROGRAMS =
+
+libexec_PROGRAMS = \
+  gnunet-service-psyc
+
+gnunet_service_psyc_SOURCES = \
+  gnunet-service-psyc.c
+gnunet_service_psyc_LDADD = \
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/statistics/libgnunetstatistics.la \
+  $(top_builddir)/src/multicast/libgnunetmulticast.la \
+  $(top_builddir)/src/psycstore/libgnunetpsycstore.la \
+  $(top_builddir)/src/psycutil/libgnunetpsycutil.la \
+  $(GN_LIBINTL)
+gnunet_service_psyc_CFLAGS = $(AM_CFLAGS)
+
+
+if HAVE_TESTING
+check_PROGRAMS = \
+ test_psyc
+# test_psyc2
+endif
+
+if ENABLE_TEST_RUN
+AM_TESTS_ENVIRONMENT=export GNUNET_PREFIX=$${GNUNET_PREFIX:-@libdir@};export PATH=$${GNUNET_PREFIX:-@prefix@}/bin:$$PATH;unset XDG_DATA_HOME;unset XDG_CONFIG_HOME;
+TESTS = $(check_PROGRAMS)
+endif
+
+test_psyc_SOURCES = \
+ test_psyc.c
+test_psyc_LDADD = \
+  libgnunetpsyc.la \
+  $(top_builddir)/src/psycutil/libgnunetpsycutil.la \
+  $(top_builddir)/src/testing/libgnunettesting.la \
+  $(top_builddir)/src/util/libgnunetutil.la
+#test_psyc2_SOURCES = \
+#  test_psyc2.c
+#test_psyc2_LDADD = \
+#  libgnunetpsyc.la \
+#  $(top_builddir)/src/psycutil/libgnunetpsycutil.la \
+#  $(top_builddir)/src/testbed/libgnunettestbed.la \
+#  $(top_builddir)/src/util/libgnunetutil.la
+
+EXTRA_DIST = \
+  test_psyc.conf
diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
new file mode 100644
index 0000000..6f2f7a9
--- /dev/null
+++ b/src/psyc/gnunet-service-psyc.c
@@ -0,0 +1,2860 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/gnunet-service-psyc.c
+ * @brief PSYC service
+ * @author Gabor X Toth
+ */
+
+#include <inttypes.h>
+
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_constants.h"
+#include "gnunet_protocols.h"
+#include "gnunet_statistics_service.h"
+#include "gnunet_multicast_service.h"
+#include "gnunet_psycstore_service.h"
+#include "gnunet_psyc_service.h"
+#include "gnunet_psyc_util_lib.h"
+#include "psyc.h"
+
+
+/**
+ * Handle to our current configuration.
+ */
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+/**
+ * Service handle.
+ */
+static struct GNUNET_SERVICE_Handle *service;
+
+/**
+ * Handle to the statistics service.
+ */
+static struct GNUNET_STATISTICS_Handle *stats;
+
+/**
+ * Handle to the PSYCstore.
+ */
+static struct GNUNET_PSYCSTORE_Handle *store;
+
+/**
+ * All connected masters.
+ * Channel's pub_key_hash -> struct Master
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *masters;
+
+/**
+ * All connected slaves.
+ * Channel's pub_key_hash -> struct Slave
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *slaves;
+
+/**
+ * Connected slaves per channel.
+ * Channel's pub_key_hash -> Slave's pub_key -> struct Slave
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *channel_slaves;
+
+
+/**
+ * Message in the transmission queue.
+ */
+struct TransmitMessage
+{
+  struct TransmitMessage *prev;
+  struct TransmitMessage *next;
+
+  struct GNUNET_SERVICE_Client *client;
+
+  /**
+   * ID assigned to the message.
+   */
+  uint64_t id;
+
+  /**
+   * Size of message.
+   */
+  uint16_t size;
+
+  /**
+   * Type of first message part.
+   */
+  uint16_t first_ptype;
+
+  /**
+   * Type of last message part.
+   */
+  uint16_t last_ptype;
+
+  /* Followed by message */
+};
+
+
+/**
+ * Cache for received message fragments.
+ * Message fragments are only sent to clients after all modifiers arrived.
+ *
+ * chan_key -> MultiHashMap chan_msgs
+ */
+static struct GNUNET_CONTAINER_MultiHashMap *recv_cache;
+
+
+/**
+ * Entry in the chan_msgs hashmap of @a recv_cache:
+ * fragment_id -> RecvCacheEntry
+ */
+struct RecvCacheEntry
+{
+  struct GNUNET_MULTICAST_MessageHeader *mmsg;
+  uint16_t ref_count;
+};
+
+
+/**
+ * Entry in the @a recv_frags hash map of a @a Channel.
+ * message_id -> FragmentQueue
+ */
+struct FragmentQueue
+{
+  /**
+   * Fragment IDs stored in @a recv_cache.
+   */
+  struct GNUNET_CONTAINER_Heap *fragments;
+
+  /**
+   * Total size of received fragments.
+   */
+  uint64_t size;
+
+  /**
+   * Total size of received header fragments (METHOD & MODIFIERs)
+   */
+  uint64_t header_size;
+
+  /**
+   * The @a state_delta field from struct GNUNET_PSYC_MessageMethod.
+   */
+  uint64_t state_delta;
+
+  /**
+   * The @a flags field from struct GNUNET_PSYC_MessageMethod.
+   */
+  uint32_t flags;
+
+  /**
+   * Receive state of message.
+   *
+   * @see MessageFragmentState
+   */
+  uint8_t state;
+
+  /**
+   * Whether the state is already modified in PSYCstore.
+   */
+  uint8_t state_is_modified;
+
+  /**
+   * Is the message queued for delivery to the client?
+   * i.e. added to the recv_msgs queue
+   */
+  uint8_t is_queued;
+};
+
+
+/**
+ * List of connected clients.
+ */
+struct ClientList
+{
+  struct ClientList *prev;
+  struct ClientList *next;
+
+  struct GNUNET_SERVICE_Client *client;
+};
+
+
+struct Operation
+{
+  struct Operation *prev;
+  struct Operation *next;
+
+  struct GNUNET_SERVICE_Client *client;
+  struct Channel *channel;
+  uint64_t op_id;
+  uint32_t flags;
+};
+
+
+/**
+ * Common part of the client context for both a channel master and slave.
+ */
+struct Channel
+{
+  struct ClientList *clients_head;
+  struct ClientList *clients_tail;
+
+  struct Operation *op_head;
+  struct Operation *op_tail;
+
+  struct TransmitMessage *tmit_head;
+  struct TransmitMessage *tmit_tail;
+
+  /**
+   * Current PSYCstore operation.
+   */
+  struct GNUNET_PSYCSTORE_OperationHandle *store_op;
+
+  /**
+   * Received fragments not yet sent to the client.
+   * message_id -> FragmentQueue
+   */
+  struct GNUNET_CONTAINER_MultiHashMap *recv_frags;
+
+  /**
+   * Received message IDs not yet sent to the client.
+   */
+  struct GNUNET_CONTAINER_Heap *recv_msgs;
+
+  /**
+   * Public key of the channel.
+   */
+  struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
+
+  /**
+   * Hash of @a pub_key.
+   */
+  struct GNUNET_HashCode pub_key_hash;
+
+  /**
+   * Last message ID sent to the client.
+   * 0 if there is no such message.
+   */
+  uint64_t max_message_id;
+
+  /**
+   * ID of the last stateful message, where the state operations has been
+   * processed and saved to PSYCstore and which has been sent to the client.
+   * 0 if there is no such message.
+   */
+  uint64_t max_state_message_id;
+
+  /**
+   * Expected value size for the modifier being received from the PSYC service.
+   */
+  uint32_t tmit_mod_value_size_expected;
+
+  /**
+   * Actual value size for the modifier being received from the PSYC service.
+   */
+  uint32_t tmit_mod_value_size;
+
+  /**
+   * Is this channel ready to receive messages from client?
+   * #GNUNET_YES or #GNUNET_NO
+   */
+  uint8_t is_ready;
+
+  /**
+   * Is the client disconnected?
+   * #GNUNET_YES or #GNUNET_NO
+   */
+  uint8_t is_disconnecting;
+
+  /**
+   * Is this a channel master (#GNUNET_YES), or slave (#GNUNET_NO)?
+   */
+  uint8_t is_master;
+
+  union {
+    struct Master *master;
+    struct Slave *slave;
+  };
+};
+
+
+/**
+ * Client context for a channel master.
+ */
+struct Master
+{
+  /**
+   * Channel struct common for Master and Slave
+   */
+  struct Channel channel;
+
+  /**
+   * Private key of the channel.
+   */
+  struct GNUNET_CRYPTO_EddsaPrivateKey priv_key;
+
+  /**
+   * Handle for the multicast origin.
+   */
+  struct GNUNET_MULTICAST_Origin *origin;
+
+  /**
+   * Transmit handle for multicast.
+   */
+  struct GNUNET_MULTICAST_OriginTransmitHandle *tmit_handle;
+
+  /**
+   * Incoming join requests from multicast.
+   * member_pub_key -> struct GNUNET_MULTICAST_JoinHandle *
+   */
+  struct GNUNET_CONTAINER_MultiHashMap *join_reqs;
+
+  /**
+   * Last message ID transmitted to this channel.
+   *
+   * Incremented before sending a message, thus the message_id in messages sent
+   * starts from 1.
+   */
+  uint64_t max_message_id;
+
+  /**
+   * ID of the last message with state operations transmitted to the channel.
+   * 0 if there is no such message.
+   */
+  uint64_t max_state_message_id;
+
+  /**
+   * Maximum group generation transmitted to the channel.
+   */
+  uint64_t max_group_generation;
+
+  /**
+   * @see enum GNUNET_PSYC_Policy
+   */
+  enum GNUNET_PSYC_Policy policy;
+};
+
+
+/**
+ * Client context for a channel slave.
+ */
+struct Slave
+{
+  /**
+   * Channel struct common for Master and Slave
+   */
+  struct Channel channel;
+
+  /**
+   * Private key of the slave.
+   */
+  struct GNUNET_CRYPTO_EcdsaPrivateKey priv_key;
+
+  /**
+   * Public key of the slave.
+   */
+  struct GNUNET_CRYPTO_EcdsaPublicKey pub_key;
+
+  /**
+   * Hash of @a pub_key.
+   */
+  struct GNUNET_HashCode pub_key_hash;
+
+  /**
+   * Handle for the multicast member.
+   */
+  struct GNUNET_MULTICAST_Member *member;
+
+  /**
+   * Transmit handle for multicast.
+   */
+  struct GNUNET_MULTICAST_MemberTransmitHandle *tmit_handle;
+
+  /**
+   * Peer identity of the origin.
+   */
+  struct GNUNET_PeerIdentity origin;
+
+  /**
+   * Number of items in @a relays.
+   */
+  uint32_t relay_count;
+
+  /**
+   * Relays that multicast can use to connect.
+   */
+  struct GNUNET_PeerIdentity *relays;
+
+  /**
+   * Join request to be transmitted to the master on join.
+   */
+  struct GNUNET_PSYC_Message *join_msg;
+
+  /**
+   * Join decision received from multicast.
+   */
+  struct GNUNET_PSYC_JoinDecisionMessage *join_dcsn;
+
+  /**
+   * Maximum request ID for this channel.
+   */
+  uint64_t max_request_id;
+
+  /**
+   * Join flags.
+   */
+  enum GNUNET_PSYC_SlaveJoinFlags join_flags;
+};
+
+
+/**
+ * Client context.
+ */
+struct Client {
+  struct GNUNET_SERVICE_Client *client;
+  struct Channel *channel;
+};
+
+
+struct ReplayRequestKey
+{
+  uint64_t fragment_id;
+  uint64_t message_id;
+  uint64_t fragment_offset;
+  uint64_t flags;
+};
+
+
+static void
+transmit_message (struct Channel *chn);
+
+static uint64_t
+message_queue_run (struct Channel *chn);
+
+static uint64_t
+message_queue_drop (struct Channel *chn);
+
+
+static void
+schedule_transmit_message (void *cls)
+{
+  struct Channel *chn = cls;
+
+  transmit_message (chn);
+}
+
+
+/**
+ * Task run during shutdown.
+ *
+ * @param cls unused
+ */
+static void
+shutdown_task (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "shutting down...\n");
+  GNUNET_PSYCSTORE_disconnect (store);
+  if (NULL != stats)
+  {
+    GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
+    stats = NULL;
+  }
+}
+
+
+static struct Operation *
+op_add (struct Channel *chn, struct GNUNET_SERVICE_Client *client,
+        uint64_t op_id, uint32_t flags)
+{
+  struct Operation *op = GNUNET_malloc (sizeof (*op));
+  op->client = client;
+  op->channel = chn;
+  op->op_id = op_id;
+  op->flags = flags;
+  GNUNET_CONTAINER_DLL_insert (chn->op_head, chn->op_tail, op);
+  return op;
+}
+
+
+static void
+op_remove (struct Operation *op)
+{
+  GNUNET_CONTAINER_DLL_remove (op->channel->op_head, op->channel->op_tail, op);
+  GNUNET_free (op);
+}
+
+
+/**
+ * Clean up master data structures after a client disconnected.
+ */
+static void
+cleanup_master (struct Master *mst)
+{
+  struct Channel *chn = &mst->channel;
+
+  GNUNET_CONTAINER_multihashmap_destroy (mst->join_reqs);
+  GNUNET_CONTAINER_multihashmap_remove (masters, &chn->pub_key_hash, mst);
+}
+
+
+/**
+ * Clean up slave data structures after a client disconnected.
+ */
+static void
+cleanup_slave (struct Slave *slv)
+{
+  struct Channel *chn = &slv->channel;
+  struct GNUNET_CONTAINER_MultiHashMap *
+    chn_slv = GNUNET_CONTAINER_multihashmap_get (channel_slaves,
+                                                &chn->pub_key_hash);
+  GNUNET_assert (NULL != chn_slv);
+  GNUNET_CONTAINER_multihashmap_remove (chn_slv, &slv->pub_key_hash, slv);
+
+  if (0 == GNUNET_CONTAINER_multihashmap_size (chn_slv))
+  {
+    GNUNET_CONTAINER_multihashmap_remove (channel_slaves, &chn->pub_key_hash,
+                                          chn_slv);
+    GNUNET_CONTAINER_multihashmap_destroy (chn_slv);
+  }
+  GNUNET_CONTAINER_multihashmap_remove (slaves, &chn->pub_key_hash, slv);
+
+  if (NULL != slv->join_msg)
+  {
+    GNUNET_free (slv->join_msg);
+    slv->join_msg = NULL;
+  }
+  if (NULL != slv->relays)
+  {
+    GNUNET_free (slv->relays);
+    slv->relays = NULL;
+  }
+  GNUNET_CONTAINER_multihashmap_remove (slaves, &chn->pub_key_hash, slv);
+}
+
+
+/**
+ * Clean up channel data structures after a client disconnected.
+ */
+static void
+cleanup_channel (struct Channel *chn)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Cleaning up channel %s. master? %u\n",
+              chn,
+              GNUNET_h2s (&chn->pub_key_hash),
+              chn->is_master);
+  message_queue_drop (chn);
+  GNUNET_CONTAINER_multihashmap_destroy (chn->recv_frags);
+  chn->recv_frags = NULL;
+
+  if (NULL != chn->store_op)
+  {
+    GNUNET_PSYCSTORE_operation_cancel (chn->store_op);
+    chn->store_op = NULL;
+  }
+
+  (GNUNET_YES == chn->is_master)
+    ? cleanup_master (chn->master)
+    : cleanup_slave (chn->slave);
+  GNUNET_free (chn);
+}
+
+
+/**
+ * Called whenever a client is disconnected.
+ * Frees our resources associated with that client.
+ *
+ * @param cls closure
+ * @param client identification of the client
+ * @param app_ctx must match @a client
+ */
+static void
+client_notify_disconnect (void *cls,
+                          struct GNUNET_SERVICE_Client *client,
+                          void *app_ctx)
+{
+  struct Client *c = app_ctx;
+  struct Channel *chn = c->channel;
+  GNUNET_free (c);
+
+  if (NULL == chn)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p User context is NULL in client_notify_disconnect ()\n",
+                chn);
+    GNUNET_break (0);
+    return;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Client %p (%s) disconnected from channel %s\n",
+              chn,
+              client,
+              (GNUNET_YES == chn->is_master) ? "master" : "slave",
+              GNUNET_h2s (&chn->pub_key_hash));
+
+  struct ClientList *cli = chn->clients_head;
+  while (NULL != cli)
+  {
+    if (cli->client == client)
+    {
+      GNUNET_CONTAINER_DLL_remove (chn->clients_head, chn->clients_tail, cli);
+      GNUNET_free (cli);
+      break;
+    }
+    cli = cli->next;
+  }
+
+  struct Operation *op = chn->op_head;
+  while (NULL != op)
+  {
+    if (op->client == client)
+    {
+      op->client = NULL;
+      break;
+    }
+    op = op->next;
+  }
+
+  if (NULL == chn->clients_head)
+  { /* Last client disconnected. */
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Last client (%s) disconnected from channel %s\n",
+                chn,
+                (GNUNET_YES == chn->is_master) ? "master" : "slave",
+                GNUNET_h2s (&chn->pub_key_hash));
+    chn->is_disconnecting = GNUNET_YES;
+    cleanup_channel (chn);
+  }
+}
+
+
+/**
+ * A new client connected.
+ *
+ * @param cls NULL
+ * @param client client to add
+ * @param mq message queue for @a client
+ * @return @a client
+ */
+static void *
+client_notify_connect (void *cls,
+                       struct GNUNET_SERVICE_Client *client,
+                       struct GNUNET_MQ_Handle *mq)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client connected: %p\n", client);
+
+  struct Client *c = GNUNET_malloc (sizeof (*c));
+  c->client = client;
+
+  return c;
+}
+
+
+/**
+ * Send message to all clients connected to the channel.
+ */
+static void
+client_send_msg (const struct Channel *chn,
+                 const struct GNUNET_MessageHeader *msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Sending message to clients of channel %p.\n",
+              chn);
+
+  struct ClientList *cli = chn->clients_head;
+  while (NULL != cli)
+  {
+    struct GNUNET_MQ_Envelope *
+      env = GNUNET_MQ_msg_copy (msg);
+
+    GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (cli->client),
+                    env);
+    cli = cli->next;
+  }
+}
+
+
+/**
+ * Send a result code back to the client.
+ *
+ * @param client
+ *        Client that should receive the result code.
+ * @param result_code
+ *        Code to transmit.
+ * @param op_id
+ *        Operation ID in network byte order.
+ * @param data
+ *        Data payload or NULL.
+ * @param data_size
+ *        Size of @a data.
+ */
+static void
+client_send_result (struct GNUNET_SERVICE_Client *client, uint64_t op_id,
+                    int64_t result_code, const void *data, uint16_t data_size)
+{
+  struct GNUNET_OperationResultMessage *res;
+  struct GNUNET_MQ_Envelope *
+    env = GNUNET_MQ_msg_extra (res,
+                               data_size,
+                               GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE);
+  res->result_code = GNUNET_htonll (result_code);
+  res->op_id = op_id;
+  if (0 < data_size)
+    GNUNET_memcpy (&res[1], data, data_size);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Sending result to client for OP ID %" PRIu64 ": %" PRId64 " (size: %u)\n",
+              client,
+              GNUNET_ntohll (op_id),
+              result_code,
+              data_size);
+
+  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
+}
+
+
+/**
+ * Closure for join_mem_test_cb()
+ */
+struct JoinMemTestClosure
+{
+  struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
+  struct Channel *channel;
+  struct GNUNET_MULTICAST_JoinHandle *join_handle;
+  struct GNUNET_PSYC_JoinRequestMessage *join_msg;
+};
+
+
+/**
+ * Membership test result callback used for join requests.
+ */
+static void
+join_mem_test_cb (void *cls, int64_t result,
+                  const char *err_msg, uint16_t err_msg_size)
+{
+  struct JoinMemTestClosure *jcls = cls;
+
+  if (GNUNET_NO == result && GNUNET_YES == jcls->channel->is_master)
+  { /* Pass on join request to client if this is a master channel */
+    struct Master *mst = jcls->channel->master;
+    struct GNUNET_HashCode slave_pub_hash;
+    GNUNET_CRYPTO_hash (&jcls->slave_pub_key, sizeof (jcls->slave_pub_key),
+                        &slave_pub_hash);
+    GNUNET_CONTAINER_multihashmap_put (mst->join_reqs, &slave_pub_hash, jcls->join_handle,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+    client_send_msg (jcls->channel, &jcls->join_msg->header);
+  }
+  else
+  {
+    if (GNUNET_SYSERR == result)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Could not perform membership test (%.*s)\n",
+                  err_msg_size, err_msg);
+    }
+    // FIXME: add relays
+    GNUNET_MULTICAST_join_decision (jcls->join_handle, result, 0, NULL, NULL);
+  }
+  GNUNET_free (jcls->join_msg);
+  GNUNET_free (jcls);
+}
+
+
+/**
+ * Incoming join request from multicast.
+ */
+static void
+mcast_recv_join_request (void *cls,
+                         const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
+                         const struct GNUNET_MessageHeader *join_msg,
+                         struct GNUNET_MULTICAST_JoinHandle *jh)
+{
+  struct Channel *chn = cls;
+  uint16_t join_msg_size = 0;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Got join request.\n",
+              chn);
+  if (NULL != join_msg)
+  {
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE == ntohs (join_msg->type))
+    {
+      join_msg_size = ntohs (join_msg->size);
+    }
+    else
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                  "%p Got join message with invalid type %u.\n",
+                  chn,
+                  ntohs (join_msg->type));
+    }
+  }
+
+  struct GNUNET_PSYC_JoinRequestMessage *
+    req = GNUNET_malloc (sizeof (*req) + join_msg_size);
+  req->header.size = htons (sizeof (*req) + join_msg_size);
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST);
+  req->slave_pub_key = *slave_pub_key;
+  if (0 < join_msg_size)
+    GNUNET_memcpy (&req[1], join_msg, join_msg_size);
+
+  struct JoinMemTestClosure *jcls = GNUNET_malloc (sizeof (*jcls));
+  jcls->slave_pub_key = *slave_pub_key;
+  jcls->channel = chn;
+  jcls->join_handle = jh;
+  jcls->join_msg = req;
+
+  GNUNET_PSYCSTORE_membership_test (store, &chn->pub_key, slave_pub_key,
+                                    chn->max_message_id, 0,
+                                    &join_mem_test_cb, jcls);
+}
+
+
+/**
+ * Join decision received from multicast.
+ */
+static void
+mcast_recv_join_decision (void *cls, int is_admitted,
+                          const struct GNUNET_PeerIdentity *peer,
+                          uint16_t relay_count,
+                          const struct GNUNET_PeerIdentity *relays,
+                          const struct GNUNET_MessageHeader *join_resp)
+{
+  struct Slave *slv = cls;
+  struct Channel *chn = &slv->channel;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Got join decision: %d\n",
+              slv,
+              is_admitted);
+  if (GNUNET_YES == chn->is_ready)
+  {
+    /* Already admitted */
+    return;
+  }
+
+  uint16_t join_resp_size = (NULL != join_resp) ? ntohs (join_resp->size) : 0;
+  struct GNUNET_PSYC_JoinDecisionMessage *
+    dcsn = slv->join_dcsn = GNUNET_malloc (sizeof (*dcsn) + join_resp_size);
+  dcsn->header.size = htons (sizeof (*dcsn) + join_resp_size);
+  dcsn->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION);
+  dcsn->is_admitted = htonl (is_admitted);
+  if (0 < join_resp_size)
+    GNUNET_memcpy (&dcsn[1], join_resp, join_resp_size);
+
+  client_send_msg (chn, &dcsn->header);
+
+  if (GNUNET_YES == is_admitted
+      && ! (GNUNET_PSYC_SLAVE_JOIN_LOCAL & slv->join_flags))
+  {
+    chn->is_ready = GNUNET_YES;
+  }
+}
+
+
+static int
+store_recv_fragment_replay (void *cls,
+                            struct GNUNET_MULTICAST_MessageHeader *msg,
+                            enum GNUNET_PSYCSTORE_MessageFlags flags)
+{
+  struct GNUNET_MULTICAST_ReplayHandle *rh = cls;
+
+  GNUNET_MULTICAST_replay_response (rh, &msg->header, GNUNET_MULTICAST_REC_OK);
+  return GNUNET_YES;
+}
+
+
+/**
+ * Received result of GNUNET_PSYCSTORE_fragment_get() for multicast replay.
+ */
+static void
+store_recv_fragment_replay_result (void *cls,
+                                   int64_t result,
+                                   const char *err_msg,
+                                   uint16_t err_msg_size)
+{
+  struct GNUNET_MULTICAST_ReplayHandle *rh = cls;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Fragment replay: PSYCSTORE returned %" PRId64 " (%.*s)\n",
+              rh,
+              result,
+              err_msg_size,
+              err_msg);
+  switch (result)
+  {
+  case GNUNET_YES:
+    break;
+
+  case GNUNET_NO:
+    GNUNET_MULTICAST_replay_response (rh, NULL,
+                                      GNUNET_MULTICAST_REC_NOT_FOUND);
+    return;
+
+  case GNUNET_PSYCSTORE_MEMBERSHIP_TEST_FAILED:
+    GNUNET_MULTICAST_replay_response (rh, NULL,
+                                      GNUNET_MULTICAST_REC_ACCESS_DENIED);
+    return;
+
+  case GNUNET_SYSERR:
+    GNUNET_MULTICAST_replay_response (rh, NULL,
+                                      GNUNET_MULTICAST_REC_INTERNAL_ERROR);
+    return;
+  }
+  /* GNUNET_MULTICAST_replay_response frees 'rh' when passed
+   * an error code, so it must be ensured no further processing
+   * is attempted on 'rh'. Maybe this should be refactored as
+   * it doesn't look very intuitive.    --lynX
+   */
+  GNUNET_MULTICAST_replay_response_end (rh);
+}
+
+
+/**
+ * Incoming fragment replay request from multicast.
+ */
+static void
+mcast_recv_replay_fragment (void *cls,
+                            const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
+                            uint64_t fragment_id, uint64_t flags,
+                            struct GNUNET_MULTICAST_ReplayHandle *rh)
+
+{
+  struct Channel *chn = cls;
+  GNUNET_PSYCSTORE_fragment_get (store, &chn->pub_key, slave_pub_key,
+                                 fragment_id, fragment_id,
+                                 &store_recv_fragment_replay,
+                                 &store_recv_fragment_replay_result, rh);
+}
+
+
+/**
+ * Incoming message replay request from multicast.
+ */
+static void
+mcast_recv_replay_message (void *cls,
+                           const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
+                           uint64_t message_id,
+                           uint64_t fragment_offset,
+                           uint64_t flags,
+                           struct GNUNET_MULTICAST_ReplayHandle *rh)
+{
+  struct Channel *chn = cls;
+  GNUNET_PSYCSTORE_message_get (store, &chn->pub_key, slave_pub_key,
+                                message_id, message_id, 1, NULL,
+                                &store_recv_fragment_replay,
+                                &store_recv_fragment_replay_result, rh);
+}
+
+
+/**
+ * Convert an uint64_t in network byte order to a HashCode
+ * that can be used as key in a MultiHashMap
+ */
+static inline void
+hash_key_from_nll (struct GNUNET_HashCode *key, uint64_t n)
+{
+  /* use little-endian order, as idx_of MultiHashMap casts key to unsigned int */
+  /* TODO: use built-in byte swap functions if available */
+
+  n = ((n <<  8) & 0xFF00FF00FF00FF00ULL) | ((n >>  8) & 0x00FF00FF00FF00FFULL);
+  n = ((n << 16) & 0xFFFF0000FFFF0000ULL) | ((n >> 16) & 0x0000FFFF0000FFFFULL);
+
+  *key = (struct GNUNET_HashCode) {};
+  *((uint64_t *) key)
+    = (n << 32) | (n >> 32);
+}
+
+
+/**
+ * Convert an uint64_t in host byte order to a HashCode
+ * that can be used as key in a MultiHashMap
+ */
+static inline void
+hash_key_from_hll (struct GNUNET_HashCode *key, uint64_t n)
+{
+#if __BYTE_ORDER == __BIG_ENDIAN
+  hash_key_from_nll (key, n);
+#elif __BYTE_ORDER == __LITTLE_ENDIAN
+  *key = (struct GNUNET_HashCode) {};
+  *((uint64_t *) key) = n;
+#else
+  #error byteorder undefined
+#endif
+}
+
+
+/**
+ * Initialize PSYC message header.
+ */
+static inline void
+psyc_msg_init (struct GNUNET_PSYC_MessageHeader *pmsg,
+               const struct GNUNET_MULTICAST_MessageHeader *mmsg, uint32_t flags)
+{
+  uint16_t size = ntohs (mmsg->header.size);
+  uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+
+  pmsg->header.size = htons (psize);
+  pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+  pmsg->message_id = mmsg->message_id;
+  pmsg->fragment_offset = mmsg->fragment_offset;
+  pmsg->flags = htonl (flags);
+
+  GNUNET_memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
+}
+
+
+/**
+ * Create a new PSYC message from a multicast message for sending it to clients.
+ */
+static inline struct GNUNET_PSYC_MessageHeader *
+psyc_msg_new (const struct GNUNET_MULTICAST_MessageHeader *mmsg, uint32_t flags)
+{
+  struct GNUNET_PSYC_MessageHeader *pmsg;
+  uint16_t size = ntohs (mmsg->header.size);
+  uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+
+  pmsg = GNUNET_malloc (psize);
+  psyc_msg_init (pmsg, mmsg, flags);
+  return pmsg;
+}
+
+
+/**
+ * Send multicast message to all clients connected to the channel.
+ */
+static void
+client_send_mcast_msg (struct Channel *chn,
+                       const struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                       uint32_t flags)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Sending multicast message to client. fragment_id: %" PRIu64 ", message_id: %" PRIu64 "\n",
+              chn,
+              GNUNET_ntohll (mmsg->fragment_id),
+              GNUNET_ntohll (mmsg->message_id));
+
+  struct GNUNET_PSYC_MessageHeader *
+    pmsg = GNUNET_PSYC_message_header_create (mmsg, flags);
+  client_send_msg (chn, &pmsg->header);
+  GNUNET_free (pmsg);
+}
+
+
+/**
+ * Send multicast request to all clients connected to the channel.
+ */
+static void
+client_send_mcast_req (struct Master *mst,
+                       const struct GNUNET_MULTICAST_RequestHeader *req)
+{
+  struct Channel *chn = &mst->channel;
+
+  struct GNUNET_PSYC_MessageHeader *pmsg;
+  uint16_t size = ntohs (req->header.size);
+  uint16_t psize = sizeof (*pmsg) + size - sizeof (*req);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Sending multicast request to client. fragment_id: %" PRIu64 ", message_id: %" PRIu64 "\n",
+              chn,
+              GNUNET_ntohll (req->fragment_id),
+              GNUNET_ntohll (req->request_id));
+
+  pmsg = GNUNET_malloc (psize);
+  pmsg->header.size = htons (psize);
+  pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+  pmsg->message_id = req->request_id;
+  pmsg->fragment_offset = req->fragment_offset;
+  pmsg->flags = htonl (GNUNET_PSYC_MESSAGE_REQUEST);
+  pmsg->slave_pub_key = req->member_pub_key;
+  GNUNET_memcpy (&pmsg[1], &req[1], size - sizeof (*req));
+
+  client_send_msg (chn, &pmsg->header);
+
+  /* FIXME: save req to PSYCstore so that it can be resent later to clients */
+
+  GNUNET_free (pmsg);
+}
+
+
+/**
+ * Insert a multicast message fragment into the queue belonging to the message.
+ *
+ * @param chn          Channel.
+ * @param mmsg         Multicast message fragment.
+ * @param msg_id_hash  Message ID of @a mmsg in a struct GNUNET_HashCode.
+ * @param first_ptype  First PSYC message part type in @a mmsg.
+ * @param last_ptype   Last PSYC message part type in @a mmsg.
+ */
+static void
+fragment_queue_insert (struct Channel *chn,
+                       const struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                       uint16_t first_ptype, uint16_t last_ptype)
+{
+  const uint16_t size = ntohs (mmsg->header.size);
+  const uint64_t frag_offset = GNUNET_ntohll (mmsg->fragment_offset);
+  struct GNUNET_CONTAINER_MultiHashMap
+    *chan_msgs = GNUNET_CONTAINER_multihashmap_get (recv_cache,
+                                                    &chn->pub_key_hash);
+
+  struct GNUNET_HashCode msg_id_hash;
+  hash_key_from_nll (&msg_id_hash, mmsg->message_id);
+
+  struct FragmentQueue
+    *fragq = GNUNET_CONTAINER_multihashmap_get (chn->recv_frags, &msg_id_hash);
+
+  if (NULL == fragq)
+  {
+    fragq = GNUNET_malloc (sizeof (*fragq));
+    fragq->state = MSG_FRAG_STATE_HEADER;
+    fragq->fragments
+      = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
+
+    GNUNET_CONTAINER_multihashmap_put (chn->recv_frags, &msg_id_hash, fragq,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+
+    if (NULL == chan_msgs)
+    {
+      chan_msgs = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+      GNUNET_CONTAINER_multihashmap_put (recv_cache, &chn->pub_key_hash, chan_msgs,
+                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+    }
+  }
+
+  struct GNUNET_HashCode frag_id_hash;
+  hash_key_from_nll (&frag_id_hash, mmsg->fragment_id);
+  struct RecvCacheEntry
+    *cache_entry = GNUNET_CONTAINER_multihashmap_get (chan_msgs, &frag_id_hash);
+  if (NULL == cache_entry)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Adding message fragment to cache. message_id: %" PRIu64 ", fragment_id: %" PRIu64 "\n",
+                chn,
+                GNUNET_ntohll (mmsg->message_id),
+                GNUNET_ntohll (mmsg->fragment_id));
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p header_size: %" PRIu64 " + %u\n",
+                chn,
+                fragq->header_size,
+                size);
+    cache_entry = GNUNET_malloc (sizeof (*cache_entry));
+    cache_entry->ref_count = 1;
+    cache_entry->mmsg = GNUNET_malloc (size);
+    GNUNET_memcpy (cache_entry->mmsg, mmsg, size);
+    GNUNET_CONTAINER_multihashmap_put (chan_msgs, &frag_id_hash, cache_entry,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+  }
+  else
+  {
+    cache_entry->ref_count++;
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Message fragment is already in cache. message_id: %" PRIu64 ", fragment_id: %" PRIu64 ", ref_count: %u\n",
+                chn,
+                GNUNET_ntohll (mmsg->message_id),
+                GNUNET_ntohll (mmsg->fragment_id),
+                cache_entry->ref_count);
+  }
+
+  if (MSG_FRAG_STATE_HEADER == fragq->state)
+  {
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == first_ptype)
+    {
+      struct GNUNET_PSYC_MessageMethod *
+        pmeth = (struct GNUNET_PSYC_MessageMethod *) &mmsg[1];
+      fragq->state_delta = GNUNET_ntohll (pmeth->state_delta);
+      fragq->flags = ntohl (pmeth->flags);
+    }
+
+    if (last_ptype < GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA)
+    {
+      fragq->header_size += size;
+    }
+    else if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == first_ptype
+             || frag_offset == fragq->header_size)
+    { /* header is now complete */
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "%p Header of message %" PRIu64 " is complete.\n",
+                  chn,
+                  GNUNET_ntohll (mmsg->message_id));
+
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "%p Adding message %" PRIu64 " to queue.\n",
+                  chn,
+                  GNUNET_ntohll (mmsg->message_id));
+      fragq->state = MSG_FRAG_STATE_DATA;
+    }
+    else
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "%p Header of message %" PRIu64 " is NOT complete yet: %" PRIu64 " != %" PRIu64 "\n",
+                  chn,
+                  GNUNET_ntohll (mmsg->message_id),
+                  frag_offset,
+                  fragq->header_size);
+    }
+  }
+
+  switch (last_ptype)
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+    if (frag_offset == fragq->size)
+      fragq->state = MSG_FRAG_STATE_END;
+    else
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "%p Message %" PRIu64 " is NOT complete yet: %" PRIu64 " != %" PRIu64 "\n",
+                  chn,
+                  GNUNET_ntohll (mmsg->message_id),
+                  frag_offset,
+                  fragq->size);
+    break;
+
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+    /* Drop message without delivering to client if it's a single fragment */
+    fragq->state =
+      (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == first_ptype)
+      ? MSG_FRAG_STATE_DROP
+      : MSG_FRAG_STATE_CANCEL;
+  }
+
+  switch (fragq->state)
+  {
+  case MSG_FRAG_STATE_DATA:
+  case MSG_FRAG_STATE_END:
+  case MSG_FRAG_STATE_CANCEL:
+    if (GNUNET_NO == fragq->is_queued)
+    {
+      GNUNET_CONTAINER_heap_insert (chn->recv_msgs, NULL,
+                                    GNUNET_ntohll (mmsg->message_id));
+      fragq->is_queued = GNUNET_YES;
+    }
+  }
+
+  fragq->size += size;
+  GNUNET_CONTAINER_heap_insert (fragq->fragments, NULL,
+                                GNUNET_ntohll (mmsg->fragment_id));
+}
+
+
+/**
+ * Run fragment queue of a message.
+ *
+ * Send fragments of a message in order to client, after all modifiers arrived
+ * from multicast.
+ *
+ * @param chn
+ *        Channel.
+ * @param msg_id
+ *        ID of the message @a fragq belongs to.
+ * @param fragq
+ *        Fragment queue of the message.
+ * @param drop
+ *        Drop message without delivering to client?
+ *        #GNUNET_YES or #GNUNET_NO.
+ */
+static void
+fragment_queue_run (struct Channel *chn, uint64_t msg_id,
+                    struct FragmentQueue *fragq, uint8_t drop)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Running message fragment queue for message %" PRIu64 " (state: %u).\n",
+              chn,
+              msg_id,
+              fragq->state);
+
+  struct GNUNET_CONTAINER_MultiHashMap
+    *chan_msgs = GNUNET_CONTAINER_multihashmap_get (recv_cache,
+                                                    &chn->pub_key_hash);
+  GNUNET_assert (NULL != chan_msgs);
+  uint64_t frag_id;
+
+  while (GNUNET_YES == GNUNET_CONTAINER_heap_peek2 (fragq->fragments, NULL,
+                                                    &frag_id))
+  {
+    struct GNUNET_HashCode frag_id_hash;
+    hash_key_from_hll (&frag_id_hash, frag_id);
+    struct RecvCacheEntry *cache_entry
+      = GNUNET_CONTAINER_multihashmap_get (chan_msgs, &frag_id_hash);
+    if (cache_entry != NULL)
+    {
+      if (GNUNET_NO == drop)
+      {
+        client_send_mcast_msg (chn, cache_entry->mmsg, 0);
+      }
+      if (cache_entry->ref_count <= 1)
+      {
+        GNUNET_CONTAINER_multihashmap_remove (chan_msgs, &frag_id_hash,
+                                              cache_entry);
+        GNUNET_free (cache_entry->mmsg);
+        GNUNET_free (cache_entry);
+      }
+      else
+      {
+        cache_entry->ref_count--;
+      }
+    }
+#if CACHE_AGING_IMPLEMENTED
+    else if (GNUNET_NO == drop)
+    {
+      /* TODO: fragment not in cache anymore, retrieve it from PSYCstore */
+    }
+#endif
+
+    GNUNET_CONTAINER_heap_remove_root (fragq->fragments);
+  }
+
+  if (MSG_FRAG_STATE_END <= fragq->state)
+  {
+    struct GNUNET_HashCode msg_id_hash;
+    hash_key_from_hll (&msg_id_hash, msg_id);
+
+    GNUNET_CONTAINER_multihashmap_remove (chn->recv_frags, &msg_id_hash, fragq);
+    GNUNET_CONTAINER_heap_destroy (fragq->fragments);
+    GNUNET_free (fragq);
+  }
+  else
+  {
+    fragq->is_queued = GNUNET_NO;
+  }
+}
+
+
+struct StateModifyClosure
+{
+  struct Channel *channel;
+  uint64_t msg_id;
+  struct GNUNET_HashCode msg_id_hash;
+};
+
+
+void
+store_recv_state_modify_result (void *cls, int64_t result,
+                                const char *err_msg, uint16_t err_msg_size)
+{
+  struct StateModifyClosure *mcls = cls;
+  struct Channel *chn = mcls->channel;
+  uint64_t msg_id = mcls->msg_id;
+
+  struct FragmentQueue *
+    fragq = GNUNET_CONTAINER_multihashmap_get (chn->recv_frags, &mcls->msg_id_hash);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p GNUNET_PSYCSTORE_state_modify() returned %" PRId64 " (%.*s)\n",
+              chn, result, err_msg_size, err_msg);
+
+  switch (result)
+  {
+  case GNUNET_OK:
+  case GNUNET_NO:
+    if (NULL != fragq)
+      fragq->state_is_modified = GNUNET_YES;
+    if (chn->max_state_message_id < msg_id)
+      chn->max_state_message_id = msg_id;
+    if (chn->max_message_id < msg_id)
+      chn->max_message_id = msg_id;
+
+    if (NULL != fragq)
+      fragment_queue_run (chn, msg_id, fragq, MSG_FRAG_STATE_DROP == fragq->state);
+    GNUNET_CONTAINER_heap_remove_root (chn->recv_msgs);
+    message_queue_run (chn);
+    break;
+
+  default:
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p GNUNET_PSYCSTORE_state_modify() failed with error %" PRId64 " (%.*s)\n",
+                chn, result, err_msg_size, err_msg);
+    /** @todo FIXME: handle state_modify error */
+  }
+}
+
+
+/**
+ * Run message queue.
+ *
+ * Send messages in queue to client in order after a message has arrived from
+ * multicast, according to the following:
+ * - A message is only sent if all of its modifiers arrived.
+ * - A stateful message is only sent if the previous stateful message
+ *   has already been delivered to the client.
+ *
+ * @param chn  Channel.
+ *
+ * @return Number of messages removed from queue and sent to client.
+ */
+static uint64_t
+message_queue_run (struct Channel *chn)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Running message queue.\n", chn);
+  uint64_t n = 0;
+  uint64_t msg_id;
+
+  while (GNUNET_YES == GNUNET_CONTAINER_heap_peek2 (chn->recv_msgs, NULL,
+                                                    &msg_id))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Processing message %" PRIu64 " in queue.\n", chn, msg_id);
+    struct GNUNET_HashCode msg_id_hash;
+    hash_key_from_hll (&msg_id_hash, msg_id);
+
+    struct FragmentQueue *
+      fragq = GNUNET_CONTAINER_multihashmap_get (chn->recv_frags, &msg_id_hash);
+
+    if (NULL == fragq || fragq->state <= MSG_FRAG_STATE_HEADER)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "%p No fragq (%p) or header not complete.\n",
+                  chn, fragq);
+      break;
+    }
+
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Fragment queue entry:  state: %u, state delta: "
+                "%" PRIu64 " - %" PRIu64 " ?= %" PRIu64 "\n",
+                chn, fragq->state, msg_id, fragq->state_delta, chn->max_state_message_id);
+
+    if (MSG_FRAG_STATE_DATA <= fragq->state)
+    {
+      /* Check if there's a missing message before the current one */
+      if (GNUNET_PSYC_STATE_NOT_MODIFIED == fragq->state_delta)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p state NOT modified\n", chn);
+
+        if (!(fragq->flags & GNUNET_PSYC_MESSAGE_ORDER_ANY)
+            && (chn->max_message_id != msg_id - 1
+                && chn->max_message_id != msg_id))
+        {
+          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                      "%p Out of order message. "
+                      "(%" PRIu64 " != %" PRIu64 " - 1)\n",
+                      chn, chn->max_message_id, msg_id);
+          break;
+          // FIXME: keep track of messages processed in this queue run,
+          //        and only stop after reaching the end
+        }
+      }
+      else
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p state modified\n", chn);
+        if (GNUNET_YES != fragq->state_is_modified)
+        {
+          if (msg_id - fragq->state_delta != chn->max_state_message_id)
+          {
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                        "%p Out of order stateful message. "
+                        "(%" PRIu64 " - %" PRIu64 " != %" PRIu64 ")\n",
+                        chn, msg_id, fragq->state_delta, chn->max_state_message_id);
+            break;
+            // FIXME: keep track of messages processed in this queue run,
+            //        and only stop after reaching the end
+          }
+
+          struct StateModifyClosure *mcls = GNUNET_malloc (sizeof (*mcls));
+          mcls->channel = chn;
+          mcls->msg_id = msg_id;
+          mcls->msg_id_hash = msg_id_hash;
+
+          /* Apply modifiers to state in PSYCstore */
+          GNUNET_PSYCSTORE_state_modify (store, &chn->pub_key, msg_id,
+                                         fragq->state_delta,
+                                         store_recv_state_modify_result, mcls);
+          break; // continue after asynchronous state modify result
+        }
+      }
+      chn->max_message_id = msg_id;
+    }
+    fragment_queue_run (chn, msg_id, fragq, MSG_FRAG_STATE_DROP == fragq->state);
+    GNUNET_CONTAINER_heap_remove_root (chn->recv_msgs);
+    n++;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Removed %" PRIu64 " messages from queue.\n", chn, n);
+  return n;
+}
+
+
+/**
+ * Drop message queue of a channel.
+ *
+ * Remove all messages in queue without sending it to clients.
+ *
+ * @param chn  Channel.
+ *
+ * @return Number of messages removed from queue.
+ */
+static uint64_t
+message_queue_drop (struct Channel *chn)
+{
+  uint64_t n = 0;
+  uint64_t msg_id;
+  while (GNUNET_YES == GNUNET_CONTAINER_heap_peek2 (chn->recv_msgs, NULL,
+                                                    &msg_id))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p Dropping message %" PRIu64 " from queue.\n", chn, msg_id);
+    struct GNUNET_HashCode msg_id_hash;
+    hash_key_from_hll (&msg_id_hash, msg_id);
+
+    struct FragmentQueue *
+      fragq = GNUNET_CONTAINER_multihashmap_get (chn->recv_frags, &msg_id_hash);
+    GNUNET_assert (NULL != fragq);
+    fragment_queue_run (chn, msg_id, fragq, GNUNET_YES);
+    GNUNET_CONTAINER_heap_remove_root (chn->recv_msgs);
+    n++;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Removed %" PRIu64 " messages from queue.\n", chn, n);
+  return n;
+}
+
+
+/**
+ * Received result of GNUNET_PSYCSTORE_fragment_store().
+ */
+static void
+store_recv_fragment_store_result (void *cls, int64_t result,
+                                  const char *err_msg, uint16_t err_msg_size)
+{
+  struct Channel *chn = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p GNUNET_PSYCSTORE_fragment_store() returned %" PRId64 " (%.*s)\n",
+              chn, result, err_msg_size, err_msg);
+}
+
+
+/**
+ * Handle incoming message fragment from multicast.
+ *
+ * Store it using PSYCstore and send it to the clients of the channel in order.
+ */
+static void
+mcast_recv_message (void *cls, const struct GNUNET_MULTICAST_MessageHeader *mmsg)
+{
+  struct Channel *chn = cls;
+  uint16_t size = ntohs (mmsg->header.size);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received multicast message of size %u. "
+              "fragment_id=%" PRIu64 ", message_id=%" PRIu64
+              ", fragment_offset=%" PRIu64 ", flags=%" PRIu64 "\n",
+              chn, size,
+              GNUNET_ntohll (mmsg->fragment_id),
+              GNUNET_ntohll (mmsg->message_id),
+              GNUNET_ntohll (mmsg->fragment_offset),
+              GNUNET_ntohll (mmsg->flags));
+
+  GNUNET_PSYCSTORE_fragment_store (store, &chn->pub_key, mmsg, 0,
+                                   &store_recv_fragment_store_result, chn);
+
+  uint16_t first_ptype = 0, last_ptype = 0;
+  int check = GNUNET_PSYC_receive_check_parts (size - sizeof (*mmsg),
+                                               (const char *) &mmsg[1],
+                                               &first_ptype, &last_ptype);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Message check result %d, first part type %u, last part type %u\n",
+              chn, check, first_ptype, last_ptype);
+  if (GNUNET_SYSERR == check)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p Dropping incoming multicast message with invalid parts.\n",
+                chn);
+    GNUNET_break_op (0);
+    return;
+  }
+
+  fragment_queue_insert (chn, mmsg, first_ptype, last_ptype);
+  message_queue_run (chn);
+}
+
+
+/**
+ * Incoming request fragment from multicast for a master.
+ *
+ * @param cls   Master.
+ * @param req   The request.
+ */
+static void
+mcast_recv_request (void *cls,
+                    const struct GNUNET_MULTICAST_RequestHeader *req)
+{
+  struct Master *mst = cls;
+  uint16_t size = ntohs (req->header.size);
+
+  char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&req->member_pub_key);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received multicast request of size %u from %s.\n",
+              mst, size, str);
+  GNUNET_free (str);
+
+  uint16_t first_ptype = 0, last_ptype = 0;
+  if (GNUNET_SYSERR
+      == GNUNET_PSYC_receive_check_parts (size - sizeof (*req),
+                                          (const char *) &req[1],
+                                          &first_ptype, &last_ptype))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p Dropping incoming multicast request with invalid parts.\n",
+                mst);
+    GNUNET_break_op (0);
+    return;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Message parts: first: type %u, last: type %u\n",
+              first_ptype, last_ptype);
+
+  /* FIXME: in-order delivery */
+  client_send_mcast_req (mst, req);
+}
+
+
+/**
+ * Response from PSYCstore with the current counter values for a channel master.
+ */
+static void
+store_recv_master_counters (void *cls, int result, uint64_t max_fragment_id,
+                            uint64_t max_message_id, uint64_t max_group_generation,
+                            uint64_t max_state_message_id)
+{
+  struct Master *mst = cls;
+  struct Channel *chn = &mst->channel;
+  chn->store_op = NULL;
+
+  struct GNUNET_PSYC_CountersResultMessage res;
+  res.header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK);
+  res.header.size = htons (sizeof (res));
+  res.result_code = htonl (result);
+  res.max_message_id = GNUNET_htonll (max_message_id);
+
+  if (GNUNET_OK == result || GNUNET_NO == result)
+  {
+    mst->max_message_id = max_message_id;
+    chn->max_message_id = max_message_id;
+    chn->max_state_message_id = max_state_message_id;
+    mst->max_group_generation = max_group_generation;
+    mst->origin
+      = GNUNET_MULTICAST_origin_start (cfg, &mst->priv_key, max_fragment_id,
+                                       mcast_recv_join_request,
+                                       mcast_recv_replay_fragment,
+                                       mcast_recv_replay_message,
+                                       mcast_recv_request,
+                                       mcast_recv_message, chn);
+    chn->is_ready = GNUNET_YES;
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p GNUNET_PSYCSTORE_counters_get() "
+                "returned %d for channel %s.\n",
+                chn, result, GNUNET_h2s (&chn->pub_key_hash));
+  }
+
+  client_send_msg (chn, &res.header);
+}
+
+
+/**
+ * Response from PSYCstore with the current counter values for a channel slave.
+ */
+void
+store_recv_slave_counters (void *cls, int result, uint64_t max_fragment_id,
+                           uint64_t max_message_id, uint64_t max_group_generation,
+                           uint64_t max_state_message_id)
+{
+  struct Slave *slv = cls;
+  struct Channel *chn = &slv->channel;
+  chn->store_op = NULL;
+
+  struct GNUNET_PSYC_CountersResultMessage res;
+  res.header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK);
+  res.header.size = htons (sizeof (res));
+  res.result_code = htonl (result);
+  res.max_message_id = GNUNET_htonll (max_message_id);
+
+  if (GNUNET_YES == result || GNUNET_NO == result)
+  {
+    chn->max_message_id = max_message_id;
+    chn->max_state_message_id = max_state_message_id;
+    slv->member
+      = GNUNET_MULTICAST_member_join (cfg, &chn->pub_key, &slv->priv_key,
+                                      &slv->origin,
+                                      slv->relay_count, slv->relays,
+                                      &slv->join_msg->header,
+                                      mcast_recv_join_request,
+                                      mcast_recv_join_decision,
+                                      mcast_recv_replay_fragment,
+                                      mcast_recv_replay_message,
+                                      mcast_recv_message, chn);
+    if (NULL != slv->join_msg)
+    {
+      GNUNET_free (slv->join_msg);
+      slv->join_msg = NULL;
+    }
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p GNUNET_PSYCSTORE_counters_get() "
+                "returned %d for channel %s.\n",
+                chn, result, GNUNET_h2s (&chn->pub_key_hash));
+  }
+
+  client_send_msg (chn, &res.header);
+}
+
+
+static void
+channel_init (struct Channel *chn)
+{
+  chn->recv_msgs
+    = GNUNET_CONTAINER_heap_create (GNUNET_CONTAINER_HEAP_ORDER_MIN);
+  chn->recv_frags = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+}
+
+
+/**
+ * Handle a connecting client starting a channel master.
+ */
+static void
+handle_client_master_start (void *cls,
+                            const struct MasterStartRequest *req)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+
+  struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
+  struct GNUNET_HashCode pub_key_hash;
+
+  GNUNET_CRYPTO_eddsa_key_get_public (&req->channel_key, &pub_key);
+  GNUNET_CRYPTO_hash (&pub_key, sizeof (pub_key), &pub_key_hash);
+
+  struct Master *
+    mst = GNUNET_CONTAINER_multihashmap_get (masters, &pub_key_hash);
+  struct Channel *chn;
+
+  if (NULL == mst)
+  {
+    mst = GNUNET_malloc (sizeof (*mst));
+    mst->policy = ntohl (req->policy);
+    mst->priv_key = req->channel_key;
+    mst->join_reqs = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+
+    chn = c->channel = &mst->channel;
+    chn->master = mst;
+    chn->is_master = GNUNET_YES;
+    chn->pub_key = pub_key;
+    chn->pub_key_hash = pub_key_hash;
+    channel_init (chn);
+
+    GNUNET_CONTAINER_multihashmap_put (masters, &chn->pub_key_hash, chn,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+    chn->store_op = GNUNET_PSYCSTORE_counters_get (store, &chn->pub_key,
+                                                   store_recv_master_counters, mst);
+  }
+  else
+  {
+    chn = &mst->channel;
+
+    struct GNUNET_PSYC_CountersResultMessage *res;
+    struct GNUNET_MQ_Envelope *
+      env = GNUNET_MQ_msg (res, GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK);
+    res->result_code = htonl (GNUNET_OK);
+    res->max_message_id = GNUNET_htonll (mst->max_message_id);
+
+    GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Client connected as master to channel %s.\n",
+              mst, GNUNET_h2s (&chn->pub_key_hash));
+
+  struct ClientList *cli = GNUNET_malloc (sizeof (*cli));
+  cli->client = client;
+  GNUNET_CONTAINER_DLL_insert (chn->clients_head, chn->clients_tail, cli);
+
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+static int
+check_client_slave_join (void *cls,
+                         const struct SlaveJoinRequest *req)
+{
+  return GNUNET_OK;
+}
+
+
+/**
+ * Handle a connecting client joining as a channel slave.
+ */
+static void
+handle_client_slave_join (void *cls,
+                          const struct SlaveJoinRequest *req)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+
+  uint16_t req_size = ntohs (req->header.size);
+
+  struct GNUNET_CRYPTO_EcdsaPublicKey slv_pub_key;
+  struct GNUNET_HashCode pub_key_hash, slv_pub_hash;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "got join request from client %p\n",
+              client);
+  GNUNET_CRYPTO_ecdsa_key_get_public (&req->slave_key, &slv_pub_key);
+  GNUNET_CRYPTO_hash (&slv_pub_key, sizeof (slv_pub_key), &slv_pub_hash);
+  GNUNET_CRYPTO_hash (&req->channel_pub_key, sizeof (req->channel_pub_key), &pub_key_hash);
+
+  struct GNUNET_CONTAINER_MultiHashMap *
+    chn_slv = GNUNET_CONTAINER_multihashmap_get (channel_slaves, &pub_key_hash);
+  struct Slave *slv = NULL;
+  struct Channel *chn;
+
+  if (NULL != chn_slv)
+  {
+    slv = GNUNET_CONTAINER_multihashmap_get (chn_slv, &slv_pub_hash);
+  }
+  if (NULL == slv)
+  {
+    slv = GNUNET_malloc (sizeof (*slv));
+    slv->priv_key = req->slave_key;
+    slv->pub_key = slv_pub_key;
+    slv->pub_key_hash = slv_pub_hash;
+    slv->origin = req->origin;
+    slv->relay_count = ntohl (req->relay_count);
+    slv->join_flags = ntohl (req->flags);
+
+    const struct GNUNET_PeerIdentity *
+      relays = (const struct GNUNET_PeerIdentity *) &req[1];
+    uint16_t relay_size = slv->relay_count * sizeof (*relays);
+    uint16_t join_msg_size = 0;
+
+    if (sizeof (*req) + relay_size + sizeof (struct GNUNET_MessageHeader)
+        <= req_size)
+    {
+      struct GNUNET_PSYC_Message *
+        join_msg = (struct GNUNET_PSYC_Message *) (((char *) &req[1]) + relay_size);
+      join_msg_size = ntohs (join_msg->header.size);
+      slv->join_msg = GNUNET_malloc (join_msg_size);
+      GNUNET_memcpy (slv->join_msg, join_msg, join_msg_size);
+    }
+    if (sizeof (*req) + relay_size + join_msg_size != req_size)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "%u + %u + %u != %u\n",
+                  (unsigned int) sizeof (*req),
+                  relay_size,
+                  join_msg_size,
+                  req_size);
+      GNUNET_break (0);
+      GNUNET_SERVICE_client_drop (client);
+      GNUNET_free (slv);
+      return;
+    }
+    if (0 < slv->relay_count)
+    {
+      slv->relays = GNUNET_malloc (relay_size);
+      GNUNET_memcpy (slv->relays, &req[1], relay_size);
+    }
+
+    chn = c->channel = &slv->channel;
+    chn->slave = slv;
+    chn->is_master = GNUNET_NO;
+    chn->pub_key = req->channel_pub_key;
+    chn->pub_key_hash = pub_key_hash;
+    channel_init (chn);
+
+    if (NULL == chn_slv)
+    {
+      chn_slv = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+      GNUNET_CONTAINER_multihashmap_put (channel_slaves, &chn->pub_key_hash, chn_slv,
+                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+    }
+    GNUNET_CONTAINER_multihashmap_put (chn_slv, &slv->pub_key_hash, chn,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+    GNUNET_CONTAINER_multihashmap_put (slaves, &chn->pub_key_hash, chn,
+                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+    chn->store_op = GNUNET_PSYCSTORE_counters_get (store, &chn->pub_key,
+                                                   &store_recv_slave_counters, slv);
+  }
+  else
+  {
+    chn = &slv->channel;
+
+    struct GNUNET_PSYC_CountersResultMessage *res;
+
+    struct GNUNET_MQ_Envelope *
+      env = GNUNET_MQ_msg (res, GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK);
+    res->result_code = htonl (GNUNET_OK);
+    res->max_message_id = GNUNET_htonll (chn->max_message_id);
+
+    GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
+
+    if (GNUNET_PSYC_SLAVE_JOIN_LOCAL & slv->join_flags)
+    {
+      mcast_recv_join_decision (slv, GNUNET_YES,
+                                NULL, 0, NULL, NULL);
+    }
+    else if (NULL == slv->member)
+    {
+      slv->member
+        = GNUNET_MULTICAST_member_join (cfg, &chn->pub_key, &slv->priv_key,
+                                        &slv->origin,
+                                        slv->relay_count, slv->relays,
+                                        &slv->join_msg->header,
+                                        &mcast_recv_join_request,
+                                        &mcast_recv_join_decision,
+                                        &mcast_recv_replay_fragment,
+                                        &mcast_recv_replay_message,
+                                        &mcast_recv_message, chn);
+      if (NULL != slv->join_msg)
+      {
+        GNUNET_free (slv->join_msg);
+        slv->join_msg = NULL;
+      }
+    }
+    else if (NULL != slv->join_dcsn)
+    {
+      struct GNUNET_MQ_Envelope *
+        env = GNUNET_MQ_msg_copy (&slv->join_dcsn->header);
+      GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
+    }
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Client %p connected as slave to channel %s.\n",
+              client,
+              GNUNET_h2s (&chn->pub_key_hash));
+
+  struct ClientList *cli = GNUNET_malloc (sizeof (*cli));
+  cli->client = client;
+  GNUNET_CONTAINER_DLL_insert (chn->clients_head, chn->clients_tail, cli);
+
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+struct JoinDecisionClosure
+{
+  int32_t is_admitted;
+  struct GNUNET_MessageHeader *msg;
+};
+
+
+/**
+ * Iterator callback for sending join decisions to multicast.
+ */
+static int
+mcast_send_join_decision (void *cls, const struct GNUNET_HashCode *pub_key_hash,
+                          void *value)
+{
+  struct JoinDecisionClosure *jcls = cls;
+  struct GNUNET_MULTICAST_JoinHandle *jh = value;
+  // FIXME: add relays
+  GNUNET_MULTICAST_join_decision (jh, jcls->is_admitted, 0, NULL, jcls->msg);
+  return GNUNET_YES;
+}
+
+
+static int
+check_client_join_decision (void *cls,
+                            const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
+{
+  return GNUNET_OK;
+}
+
+
+/**
+ * Join decision from client.
+ */
+static void
+handle_client_join_decision (void *cls,
+                             const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+  struct Channel *chn = c->channel;
+  if (NULL == chn)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+  GNUNET_assert (GNUNET_YES == chn->is_master);
+  struct Master *mst = chn->master;
+
+  struct JoinDecisionClosure jcls;
+  jcls.is_admitted = ntohl (dcsn->is_admitted);
+  jcls.msg
+    = (sizeof (*dcsn) + sizeof (*jcls.msg) <= ntohs (dcsn->header.size))
+    ? (struct GNUNET_MessageHeader *) &dcsn[1]
+    : NULL;
+
+  struct GNUNET_HashCode slave_pub_hash;
+  GNUNET_CRYPTO_hash (&dcsn->slave_pub_key, sizeof (dcsn->slave_pub_key),
+                      &slave_pub_hash);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Got join decision (%d) from client for channel %s..\n",
+              mst, jcls.is_admitted, GNUNET_h2s (&chn->pub_key_hash));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p ..and slave %s.\n",
+              mst, GNUNET_h2s (&slave_pub_hash));
+
+  GNUNET_CONTAINER_multihashmap_get_multiple (mst->join_reqs, &slave_pub_hash,
+                                              &mcast_send_join_decision, &jcls);
+  GNUNET_CONTAINER_multihashmap_remove_all (mst->join_reqs, &slave_pub_hash);
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+static void
+channel_part_cb (void *cls)
+{
+  struct GNUNET_SERVICE_Client *client = cls;
+  struct GNUNET_MQ_Envelope *env;
+
+  env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_ACK);
+  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client),
+                  env);
+}
+
+
+static void
+handle_client_part_request (void *cls,
+                            const struct GNUNET_MessageHeader *msg)
+{
+  struct Client *c = cls;
+
+  c->channel->is_disconnecting = GNUNET_YES;
+  if (GNUNET_YES == c->channel->is_master)
+  {
+    struct Master *mst = (struct Master *) c->channel;
+   
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Got part request from master %p\n",
+                mst);
+    GNUNET_assert (NULL != mst->origin);
+    GNUNET_MULTICAST_origin_stop (mst->origin, channel_part_cb, c->client);
+  }
+  else
+  {
+    struct Slave *slv = (struct Slave *) c->channel;
+
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Got part request from slave %p\n",
+                slv);
+    GNUNET_assert (NULL != slv->member);
+    GNUNET_MULTICAST_member_part (slv->member, channel_part_cb, c->client);
+  }
+  GNUNET_SERVICE_client_continue (c->client);
+}
+
+
+/**
+ * Send acknowledgement to a client.
+ *
+ * Sent after a message fragment has been passed on to multicast.
+ *
+ * @param chn The channel struct for the client.
+ */
+static void
+send_message_ack (struct Channel *chn, struct GNUNET_SERVICE_Client *client)
+{
+  struct GNUNET_MessageHeader *res;
+  struct GNUNET_MQ_Envelope *
+      env = GNUNET_MQ_msg (res, GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK);
+
+  /* FIXME? */
+  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client), env);
+}
+
+
+/**
+ * Callback for the transmit functions of multicast.
+ */
+static int
+transmit_notify (void *cls, size_t *data_size, void *data)
+{
+  struct Channel *chn = cls;
+  struct TransmitMessage *tmit_msg = chn->tmit_head;
+
+  if (NULL == tmit_msg || *data_size < tmit_msg->size)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p transmit_notify: nothing to send.\n", chn);
+    if (NULL != tmit_msg && *data_size < tmit_msg->size)
+      GNUNET_break (0);
+    *data_size = 0;
+    return GNUNET_NO;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p transmit_notify: sending %u bytes.\n", chn, tmit_msg->size);
+
+  *data_size = tmit_msg->size;
+  GNUNET_memcpy (data, &tmit_msg[1], *data_size);
+
+  int ret
+    = (tmit_msg->last_ptype < GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END)
+    ? GNUNET_NO
+    : GNUNET_YES;
+
+  /* FIXME: handle disconnecting clients */
+  if (NULL != tmit_msg->client)
+    send_message_ack (chn, tmit_msg->client);
+
+  GNUNET_CONTAINER_DLL_remove (chn->tmit_head, chn->tmit_tail, tmit_msg);
+
+  if (NULL != chn->tmit_head)
+  {
+    GNUNET_SCHEDULER_add_now (&schedule_transmit_message, chn);
+  }
+  else if (GNUNET_YES == chn->is_disconnecting
+           && tmit_msg->last_ptype < GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END)
+  {
+    /* FIXME: handle partial message (when still in_transmit) */
+    GNUNET_free (tmit_msg);
+    return GNUNET_SYSERR;
+  }
+  GNUNET_free (tmit_msg);
+  return ret;
+}
+
+
+/**
+ * Callback for the transmit functions of multicast.
+ */
+static int
+master_transmit_notify (void *cls, size_t *data_size, void *data)
+{
+  int ret = transmit_notify (cls, data_size, data);
+
+  if (GNUNET_YES == ret)
+  {
+    struct Master *mst = cls;
+    mst->tmit_handle = NULL;
+  }
+  return ret;
+}
+
+
+/**
+ * Callback for the transmit functions of multicast.
+ */
+static int
+slave_transmit_notify (void *cls, size_t *data_size, void *data)
+{
+  int ret = transmit_notify (cls, data_size, data);
+
+  if (GNUNET_YES == ret)
+  {
+    struct Slave *slv = cls;
+    slv->tmit_handle = NULL;
+  }
+  return ret;
+}
+
+
+/**
+ * Transmit a message from a channel master to the multicast group.
+ */
+static void
+master_transmit_message (struct Master *mst)
+{
+  struct Channel *chn = &mst->channel;
+  struct TransmitMessage *tmit_msg = chn->tmit_head;
+  if (NULL == tmit_msg)
+    return;
+  if (NULL == mst->tmit_handle)
+  {
+    mst->tmit_handle = GNUNET_MULTICAST_origin_to_all (mst->origin,
+                                                       tmit_msg->id,
+                                                       mst->max_group_generation,
+                                                       &master_transmit_notify,
+                                                       mst);
+  }
+  else
+  {
+    GNUNET_MULTICAST_origin_to_all_resume (mst->tmit_handle);
+  }
+}
+
+
+/**
+ * Transmit a message from a channel slave to the multicast group.
+ */
+static void
+slave_transmit_message (struct Slave *slv)
+{
+  if (NULL == slv->channel.tmit_head)
+    return;
+  if (NULL == slv->tmit_handle)
+  {
+    slv->tmit_handle = GNUNET_MULTICAST_member_to_origin (slv->member,
+                                                          slv->channel.tmit_head->id,
+                                                          &slave_transmit_notify,
+                                                          slv);
+  }
+  else
+  {
+    GNUNET_MULTICAST_member_to_origin_resume (slv->tmit_handle);
+  }
+}
+
+
+static void
+transmit_message (struct Channel *chn)
+{
+  chn->is_master
+    ? master_transmit_message (chn->master)
+    : slave_transmit_message (chn->slave);
+}
+
+
+/**
+ * Queue a message from a channel master for sending to the multicast group.
+ */
+static void
+master_queue_message (struct Master *mst, struct TransmitMessage *tmit_msg)
+{
+  if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == tmit_msg->first_ptype)
+  {
+    tmit_msg->id = ++mst->max_message_id;
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p master_queue_message: message_id=%" PRIu64 "\n",
+                mst, tmit_msg->id);
+    struct GNUNET_PSYC_MessageMethod *pmeth
+      = (struct GNUNET_PSYC_MessageMethod *) &tmit_msg[1];
+
+    if (pmeth->flags & GNUNET_PSYC_MASTER_TRANSMIT_STATE_RESET)
+    {
+      pmeth->state_delta = GNUNET_htonll (GNUNET_PSYC_STATE_RESET);
+    }
+    else if (pmeth->flags & GNUNET_PSYC_MASTER_TRANSMIT_STATE_MODIFY)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "%p master_queue_message: state_delta=%" PRIu64 "\n",
+                  mst, tmit_msg->id - mst->max_state_message_id);
+      pmeth->state_delta = GNUNET_htonll (tmit_msg->id
+                                          - mst->max_state_message_id);
+      mst->max_state_message_id = tmit_msg->id;
+    }
+    else
+    {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                    "%p master_queue_message: state not modified\n", mst);
+      pmeth->state_delta = GNUNET_htonll (GNUNET_PSYC_STATE_NOT_MODIFIED);
+    }
+
+    if (pmeth->flags & GNUNET_PSYC_MASTER_TRANSMIT_STATE_HASH)
+    {
+      /// @todo add state_hash to PSYC header
+    }
+  }
+}
+
+
+/**
+ * Queue a message from a channel slave for sending to the multicast group.
+ */
+static void
+slave_queue_message (struct Slave *slv, struct TransmitMessage *tmit_msg)
+{
+  if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == tmit_msg->first_ptype)
+  {
+    struct GNUNET_PSYC_MessageMethod *pmeth
+      = (struct GNUNET_PSYC_MessageMethod *) &tmit_msg[1];
+    pmeth->state_delta = GNUNET_htonll (GNUNET_PSYC_STATE_NOT_MODIFIED);
+    tmit_msg->id = ++slv->max_request_id;
+  }
+}
+
+
+/**
+ * Queue PSYC message parts for sending to multicast.
+ *
+ * @param chn
+ *        Channel to send to.
+ * @param client
+ *        Client the message originates from.
+ * @param data_size
+ *        Size of @a data.
+ * @param data
+ *        Concatenated message parts.
+ * @param first_ptype
+ *        First message part type in @a data.
+ * @param last_ptype
+ *        Last message part type in @a data.
+ */
+static struct TransmitMessage *
+queue_message (struct Channel *chn,
+               struct GNUNET_SERVICE_Client *client,
+               size_t data_size,
+               const void *data,
+               uint16_t first_ptype, uint16_t last_ptype)
+{
+  struct TransmitMessage *
+    tmit_msg = GNUNET_malloc (sizeof (*tmit_msg) + data_size);
+  GNUNET_memcpy (&tmit_msg[1], data, data_size);
+  tmit_msg->client = client;
+  tmit_msg->size = data_size;
+  tmit_msg->first_ptype = first_ptype;
+  tmit_msg->last_ptype = last_ptype;
+
+  /* FIXME: separate queue per message ID */
+
+  GNUNET_CONTAINER_DLL_insert_tail (chn->tmit_head, chn->tmit_tail, tmit_msg);
+
+  chn->is_master
+    ? master_queue_message (chn->master, tmit_msg)
+    : slave_queue_message (chn->slave, tmit_msg);
+  return tmit_msg;
+}
+
+
+/**
+ * Cancel transmission of current message.
+ *
+ * @param chn     Channel to send to.
+ * @param client  Client the message originates from.
+ */
+static void
+transmit_cancel (struct Channel *chn, struct GNUNET_SERVICE_Client *client)
+{
+  uint16_t type = GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL;
+
+  struct GNUNET_MessageHeader msg;
+  msg.size = htons (sizeof (msg));
+  msg.type = htons (type);
+
+  queue_message (chn, client, sizeof (msg), &msg, type, type);
+  transmit_message (chn);
+
+  /* FIXME: cleanup */
+}
+
+
+static int
+check_client_psyc_message (void *cls,
+                           const struct GNUNET_MessageHeader *msg)
+{
+  return GNUNET_OK;
+}
+
+
+/**
+ * Incoming message from a master or slave client.
+ */
+static void
+handle_client_psyc_message (void *cls,
+                            const struct GNUNET_MessageHeader *msg)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+  struct Channel *chn = c->channel;
+  if (NULL == chn)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received message from client.\n", chn);
+  GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, msg);
+
+  if (GNUNET_YES != chn->is_ready)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p Channel is not ready yet, disconnecting client %p.\n",
+                chn,
+                client);
+    GNUNET_break (0);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+
+  uint16_t size = ntohs (msg->size);
+  if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < size - sizeof (*msg))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p Message payload too large: %u < %u.\n",
+                chn,
+                (unsigned int) GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD,
+                (unsigned int) (size - sizeof (*msg)));
+    GNUNET_break (0);
+    transmit_cancel (chn, client);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+
+  uint16_t first_ptype = 0, last_ptype = 0;
+  if (GNUNET_SYSERR
+      == GNUNET_PSYC_receive_check_parts (size - sizeof (*msg),
+                                          (const char *) &msg[1],
+                                          &first_ptype, &last_ptype))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p Received invalid message part from client.\n", chn);
+    GNUNET_break (0);
+    transmit_cancel (chn, client);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received message with first part type %u and last part type %u.\n",
+              chn, first_ptype, last_ptype);
+
+  queue_message (chn, client, size - sizeof (*msg), &msg[1],
+                 first_ptype, last_ptype);
+  transmit_message (chn);
+  /* FIXME: send a few ACKs even before transmit_notify is called */
+
+  GNUNET_SERVICE_client_continue (client);
+};
+
+
+/**
+ * Received result of GNUNET_PSYCSTORE_membership_store()
+ */
+static void
+store_recv_membership_store_result (void *cls,
+                                    int64_t result,
+                                    const char *err_msg,
+                                    uint16_t err_msg_size)
+{
+  struct Operation *op = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p GNUNET_PSYCSTORE_membership_store() returned %" PRId64 " (%.*s)\n",
+              op->channel,
+              result,
+              (int) err_msg_size,
+              err_msg);
+
+  if (NULL != op->client)
+    client_send_result (op->client, op->op_id, result, err_msg, err_msg_size);
+  op_remove (op);
+}
+
+
+/**
+ * Client requests to add/remove a slave in the membership database.
+ */
+static void
+handle_client_membership_store (void *cls,
+                                const struct ChannelMembershipStoreRequest *req)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+  struct Channel *chn = c->channel;
+  if (NULL == chn)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+
+  struct Operation *op = op_add (chn, client, req->op_id, 0);
+
+  uint64_t announced_at = GNUNET_ntohll (req->announced_at);
+  uint64_t effective_since = GNUNET_ntohll (req->effective_since);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received membership store request from client.\n", chn);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p did_join: %u, announced_at: %" PRIu64 ", effective_since: %" PRIu64 "\n",
+              chn, req->did_join, announced_at, effective_since);
+
+  GNUNET_PSYCSTORE_membership_store (store, &chn->pub_key, &req->slave_pub_key,
+                                     req->did_join, announced_at, effective_since,
+                                     0, /* FIXME: group_generation */
+                                     &store_recv_membership_store_result, op);
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+/**
+ * Received a fragment for GNUNET_PSYCSTORE_fragment_get(),
+ * in response to a history request from a client.
+ */
+static int
+store_recv_fragment_history (void *cls,
+                             struct GNUNET_MULTICAST_MessageHeader *mmsg,
+                             enum GNUNET_PSYCSTORE_MessageFlags flags)
+{
+  struct Operation *op = cls;
+  if (NULL == op->client)
+  { /* Requesting client already disconnected. */
+    return GNUNET_NO;
+  }
+  struct Channel *chn = op->channel;
+
+  struct GNUNET_PSYC_MessageHeader *pmsg;
+  uint16_t msize = ntohs (mmsg->header.size);
+  uint16_t psize = sizeof (*pmsg) + msize - sizeof (*mmsg);
+
+  struct GNUNET_OperationResultMessage *
+    res = GNUNET_malloc (sizeof (*res) + psize);
+  res->header.size = htons (sizeof (*res) + psize);
+  res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT);
+  res->op_id = op->op_id;
+  res->result_code = GNUNET_htonll (GNUNET_OK);
+
+  pmsg = (struct GNUNET_PSYC_MessageHeader *) &res[1];
+  GNUNET_PSYC_message_header_init (pmsg, mmsg, flags | GNUNET_PSYC_MESSAGE_HISTORIC);
+  GNUNET_memcpy (&res[1], pmsg, psize);
+
+  /** @todo FIXME: send only to requesting client */
+  client_send_msg (chn, &res->header);
+
+  GNUNET_free (res);
+  return GNUNET_YES;
+}
+
+
+/**
+ * Received the result of GNUNET_PSYCSTORE_fragment_get(),
+ * in response to a history request from a client.
+ */
+static void
+store_recv_fragment_history_result (void *cls, int64_t result,
+                                    const char *err_msg, uint16_t err_msg_size)
+{
+  struct Operation *op = cls;
+  if (NULL == op->client)
+  { /* Requesting client already disconnected. */
+    return;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p History replay #%" PRIu64 ": "
+              "PSYCSTORE returned %" PRId64 " (%.*s)\n",
+              op->channel, GNUNET_ntohll (op->op_id), result, err_msg_size, err_msg);
+
+  if (op->flags & GNUNET_PSYC_HISTORY_REPLAY_REMOTE)
+  {
+    /** @todo Multicast replay request for messages not found locally. */
+  }
+
+  client_send_result (op->client, op->op_id, result, err_msg, err_msg_size);
+  op_remove (op);
+}
+
+
+static int
+check_client_history_replay (void *cls,
+                             const struct GNUNET_PSYC_HistoryRequestMessage *req)
+{
+  return GNUNET_OK;
+}
+
+
+/**
+ * Client requests channel history.
+ */
+static void
+handle_client_history_replay (void *cls,
+                              const struct GNUNET_PSYC_HistoryRequestMessage *req)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+  struct Channel *chn = c->channel;
+  if (NULL == chn)
+  {
+    GNUNET_break (0);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+
+  uint16_t size = ntohs (req->header.size);
+  const char *method_prefix = (const char *) &req[1];
+
+  if (size < sizeof (*req) + 1
+      || '\0' != method_prefix[size - sizeof (*req) - 1])
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p History replay #%" PRIu64 ": "
+                "invalid method prefix. size: %u < %u?\n",
+                chn,
+                GNUNET_ntohll (req->op_id),
+                size,
+                (unsigned int) sizeof (*req) + 1);
+    GNUNET_break (0);
+    GNUNET_SERVICE_client_drop (client);
+    return;
+  }
+
+  struct Operation *op = op_add (chn, client, req->op_id, ntohl (req->flags));
+
+  if (0 == req->message_limit)
+  {
+    GNUNET_PSYCSTORE_message_get (store, &chn->pub_key, NULL,
+                                  GNUNET_ntohll (req->start_message_id),
+                                  GNUNET_ntohll (req->end_message_id),
+                                  0, method_prefix,
+                                  &store_recv_fragment_history,
+                                  &store_recv_fragment_history_result, op);
+  }
+  else
+  {
+    GNUNET_PSYCSTORE_message_get_latest (store, &chn->pub_key, NULL,
+                                         GNUNET_ntohll (req->message_limit),
+                                         method_prefix,
+                                         &store_recv_fragment_history,
+                                         &store_recv_fragment_history_result,
+                                         op);
+  }
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+/**
+ * Received state var from PSYCstore, send it to client.
+ */
+static int
+store_recv_state_var (void *cls, const char *name,
+                      const void *value, uint32_t value_size)
+{
+  struct Operation *op = cls;
+  struct GNUNET_OperationResultMessage *res;
+  struct GNUNET_MQ_Envelope *env;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p state_get #%" PRIu64 " - received var from PSYCstore: %s\n",
+              op->channel, GNUNET_ntohll (op->op_id), name);
+
+  if (NULL != name) /* First part */
+  {
+    uint16_t name_size = strnlen (name, GNUNET_PSYC_MODIFIER_MAX_PAYLOAD) + 1;
+    struct GNUNET_PSYC_MessageModifier *mod;
+    env = GNUNET_MQ_msg_extra (res,
+                               sizeof (*mod) + name_size + value_size,
+                               GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT);
+    res->op_id = op->op_id;
+
+    mod = (struct GNUNET_PSYC_MessageModifier *) &res[1];
+    mod->header.size = htons (sizeof (*mod) + name_size + value_size);
+    mod->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER);
+    mod->name_size = htons (name_size);
+    mod->value_size = htonl (value_size);
+    mod->oper = htons (GNUNET_PSYC_OP_ASSIGN);
+    GNUNET_memcpy (&mod[1], name, name_size);
+    GNUNET_memcpy (((char *) &mod[1]) + name_size, value, value_size);
+  }
+  else /* Continuation */
+  {
+    struct GNUNET_MessageHeader *mod;
+    env = GNUNET_MQ_msg_extra (res,
+                               sizeof (*mod) + value_size,
+                               GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT);
+    res->op_id = op->op_id;
+
+    mod = (struct GNUNET_MessageHeader *) &res[1];
+    mod->size = htons (sizeof (*mod) + value_size);
+    mod->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT);
+    GNUNET_memcpy (&mod[1], value, value_size);
+  }
+
+  // FIXME: client might have been disconnected
+  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (op->client), env);
+  return GNUNET_YES;
+}
+
+
+/**
+ * Received result of GNUNET_PSYCSTORE_state_get()
+ * or GNUNET_PSYCSTORE_state_get_prefix()
+ */
+static void
+store_recv_state_result (void *cls, int64_t result,
+                         const char *err_msg, uint16_t err_msg_size)
+{
+  struct Operation *op = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p state_get #%" PRIu64 ": "
+              "PSYCSTORE returned %" PRId64 " (%.*s)\n",
+              op->channel, GNUNET_ntohll (op->op_id), result, err_msg_size, err_msg);
+
+  // FIXME: client might have been disconnected
+  client_send_result (op->client, op->op_id, result, err_msg, err_msg_size);
+  op_remove (op);
+}
+
+
+static int
+check_client_state_get (void *cls,
+                         const struct StateRequest *req)
+{
+  struct Client *c = cls;
+  struct Channel *chn = c->channel;
+  if (NULL == chn)
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
+  const char *name = (const char *) &req[1];
+  if (0 == name_size || '\0' != name[name_size - 1])
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  return GNUNET_OK;
+}
+
+
+/**
+ * Client requests best matching state variable from PSYCstore.
+ */
+static void
+handle_client_state_get (void *cls,
+                         const struct StateRequest *req)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+  struct Channel *chn = c->channel;
+
+  const char *name = (const char *) &req[1];
+  struct Operation *op = op_add (chn, client, req->op_id, 0);
+  GNUNET_PSYCSTORE_state_get (store, &chn->pub_key, name,
+                              &store_recv_state_var,
+                              &store_recv_state_result, op);
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+static int
+check_client_state_get_prefix (void *cls,
+                               const struct StateRequest *req)
+{
+  struct Client *c = cls;
+  struct Channel *chn = c->channel;
+  if (NULL == chn)
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  uint16_t name_size = ntohs (req->header.size) - sizeof (*req);
+  const char *name = (const char *) &req[1];
+  if (0 == name_size || '\0' != name[name_size - 1])
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+
+  return GNUNET_OK;
+}
+
+
+/**
+ * Client requests state variables with a given prefix from PSYCstore.
+ */
+static void
+handle_client_state_get_prefix (void *cls,
+                                const struct StateRequest *req)
+{
+  struct Client *c = cls;
+  struct GNUNET_SERVICE_Client *client = c->client;
+  struct Channel *chn = c->channel;
+
+  const char *name = (const char *) &req[1];
+  struct Operation *op = op_add (chn, client, req->op_id, 0);
+  GNUNET_PSYCSTORE_state_get_prefix (store, &chn->pub_key, name,
+                                     &store_recv_state_var,
+                                     &store_recv_state_result, op);
+  GNUNET_SERVICE_client_continue (client);
+}
+
+
+/**
+ * Initialize the PSYC service.
+ *
+ * @param cls Closure.
+ * @param server The initialized server.
+ * @param c Configuration to use.
+ */
+static void
+run (void *cls,
+     const struct GNUNET_CONFIGURATION_Handle *c,
+     struct GNUNET_SERVICE_Handle *svc)
+{
+  cfg = c;
+  service = svc;
+  store = GNUNET_PSYCSTORE_connect (cfg);
+  stats = GNUNET_STATISTICS_create ("psyc", cfg);
+  masters = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  slaves = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  channel_slaves = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+  recv_cache = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
+  GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL);
+}
+
+
+/**
+ * Define "main" method using service macro.
+ */
+GNUNET_SERVICE_MAIN
+("psyc",
+ GNUNET_SERVICE_OPTION_NONE,
+ &run,
+ &client_notify_connect,
+ &client_notify_disconnect,
+ NULL,
+ GNUNET_MQ_hd_fixed_size (client_master_start,
+                          GNUNET_MESSAGE_TYPE_PSYC_MASTER_START,
+                          struct MasterStartRequest,
+                          NULL),
+ GNUNET_MQ_hd_var_size (client_slave_join,
+                        GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN,
+                        struct SlaveJoinRequest,
+                        NULL),
+ GNUNET_MQ_hd_var_size (client_join_decision,
+                        GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
+                        struct GNUNET_PSYC_JoinDecisionMessage,
+                        NULL),
+ GNUNET_MQ_hd_fixed_size (client_part_request,
+                          GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST,
+                          struct GNUNET_MessageHeader,
+                          NULL),
+ GNUNET_MQ_hd_var_size (client_psyc_message,
+                        GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
+                        struct GNUNET_MessageHeader,
+                        NULL),
+ GNUNET_MQ_hd_fixed_size (client_membership_store,
+                          GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE,
+                          struct ChannelMembershipStoreRequest,
+                          NULL),
+ GNUNET_MQ_hd_var_size (client_history_replay,
+                        GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY,
+                        struct GNUNET_PSYC_HistoryRequestMessage,
+                        NULL),
+ GNUNET_MQ_hd_var_size (client_state_get,
+                        GNUNET_MESSAGE_TYPE_PSYC_STATE_GET,
+                        struct StateRequest,
+                        NULL),
+ GNUNET_MQ_hd_var_size (client_state_get_prefix,
+                        GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX,
+                        struct StateRequest,
+                        NULL));
+
+/* end of gnunet-service-psyc.c */
diff --git a/src/psyc/psyc.conf.in b/src/psyc/psyc.conf.in
new file mode 100644
index 0000000..764ccfa
--- /dev/null
+++ b/src/psyc/psyc.conf.in
@@ -0,0 +1,12 @@
+[psyc]
+START_ON_DEMAND = @START_ON_DEMAND@
+BINARY = gnunet-service-psyc
+
+UNIXPATH = $GNUNET_RUNTIME_DIR/gnunet-service-psyc.sock
+UNIX_MATCH_UID = YES
+UNIX_MATCH_GID = YES
+
+@UNIXONLY@PORT = 2115
+HOSTNAME = localhost
+ACCEPT_FROM = 127.0.0.1;
+ACCEPT_FROM6 = ::1;
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
new file mode 100644
index 0000000..74bbf3e
--- /dev/null
+++ b/src/psyc/psyc.h
@@ -0,0 +1,178 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/psyc.h
+ * @brief Common type definitions for the PSYC service and API.
+ * @author Gabor X Toth
+ */
+
+#ifndef PSYC_H
+#define PSYC_H
+
+#include "platform.h"
+#include "gnunet_psyc_service.h"
+
+
+int
+GNUNET_PSYC_check_message_parts (uint16_t data_size, const char *data,
+                                 uint16_t *first_ptype, uint16_t *last_ptype);
+
+void
+GNUNET_PSYC_log_message (enum GNUNET_ErrorType kind,
+                         const struct GNUNET_MessageHeader *msg);
+
+
+enum MessageState
+{
+  MSG_STATE_START    = 0,
+  MSG_STATE_HEADER   = 1,
+  MSG_STATE_METHOD   = 2,
+  MSG_STATE_MODIFIER = 3,
+  MSG_STATE_MOD_CONT = 4,
+  MSG_STATE_DATA     = 5,
+  MSG_STATE_END      = 6,
+  MSG_STATE_CANCEL   = 7,
+  MSG_STATE_ERROR    = 8,
+};
+
+
+enum MessageFragmentState
+{
+  MSG_FRAG_STATE_START    = 0,
+  MSG_FRAG_STATE_HEADER   = 1,
+  MSG_FRAG_STATE_DATA     = 2,
+  MSG_FRAG_STATE_END      = 3,
+  MSG_FRAG_STATE_CANCEL   = 4,
+  MSG_FRAG_STATE_DROP     = 5,
+};
+
+
+GNUNET_NETWORK_STRUCT_BEGIN
+
+
+/**** library -> service ****/
+
+
+struct MasterStartRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_MASTER_START
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t policy GNUNET_PACKED;
+
+  struct GNUNET_CRYPTO_EddsaPrivateKey channel_key;
+};
+
+
+struct SlaveJoinRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t relay_count GNUNET_PACKED;
+
+  struct GNUNET_CRYPTO_EddsaPublicKey channel_pub_key;
+
+  struct GNUNET_CRYPTO_EcdsaPrivateKey slave_key;
+
+  struct GNUNET_PeerIdentity origin;
+
+  uint32_t flags GNUNET_PACKED;
+
+  /* Followed by struct GNUNET_PeerIdentity relays[relay_count] */
+
+  /* Followed by struct GNUNET_MessageHeader join_msg */
+};
+
+
+struct ChannelMembershipStoreRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t reserved GNUNET_PACKED;
+
+  uint64_t op_id GNUNET_PACKED;
+
+  struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
+
+  uint64_t announced_at GNUNET_PACKED;
+
+  uint64_t effective_since GNUNET_PACKED;
+
+  uint8_t did_join;
+};
+
+
+struct HistoryRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_HISTORY_REQUEST
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t reserved GNUNET_PACKED;
+
+  /**
+   * ID for this operation.
+   */
+  uint64_t op_id GNUNET_PACKED;
+
+  uint64_t start_message_id GNUNET_PACKED;
+
+  uint64_t end_message_id GNUNET_PACKED;
+
+  uint64_t message_limit GNUNET_PACKED;
+};
+
+
+struct StateRequest
+{
+  /**
+   * Types:
+   * - GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_STATE_GET
+   * - GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_STATE_GET_PREFIX
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t reserved GNUNET_PACKED;
+
+  /**
+   * ID for this operation.
+   */
+  uint64_t op_id GNUNET_PACKED;
+
+  /* Followed by NUL-terminated name. */
+};
+
+
+/**** service -> library ****/
+
+
+GNUNET_NETWORK_STRUCT_END
+
+#endif
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
new file mode 100644
index 0000000..37ea112
--- /dev/null
+++ b/src/psyc/psyc_api.c
@@ -0,0 +1,1584 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/psyc_api.c
+ * @brief PSYC service; high-level access to the PSYC protocol
+ *        note that clients of this API are NOT expected to
+ *        understand the PSYC message format, only the semantics!
+ *        Parsing (and serializing) the PSYC stream format is done
+ *        within the implementation of the libgnunetpsyc library,
+ *        and this API deliberately exposes as little as possible
+ *        of the actual data stream format to the application!
+ * @author Gabor X Toth
+ */
+
+#include <inttypes.h>
+
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_multicast_service.h"
+#include "gnunet_psyc_service.h"
+#include "gnunet_psyc_util_lib.h"
+#include "psyc.h"
+
+#define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
+
+
+/**
+ * Handle to access PSYC channel operations for both the master and slaves.
+ */
+struct GNUNET_PSYC_Channel
+{
+  /**
+   * Configuration to use.
+   */
+  const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+  /**
+   * Client connection to the service.
+   */
+  struct GNUNET_MQ_Handle *mq;
+
+  /**
+   * Message to send on connect.
+   */
+  struct GNUNET_MQ_Envelope *connect_env;
+
+  /**
+   * Time to wait until we try to reconnect on failure.
+   */
+  struct GNUNET_TIME_Relative reconnect_delay;
+
+  /**
+   * Task for reconnecting when the listener fails.
+   */
+  struct GNUNET_SCHEDULER_Task *reconnect_task;
+
+  /**
+   * Async operations.
+   */
+  struct GNUNET_OP_Handle *op;
+
+  /**
+   * Transmission handle;
+   */
+  struct GNUNET_PSYC_TransmitHandle *tmit;
+
+  /**
+   * Receipt handle;
+   */
+  struct GNUNET_PSYC_ReceiveHandle *recv;
+
+  /**
+   * Function called after disconnected from the service.
+   */
+  GNUNET_ContinuationCallback disconnect_cb;
+
+  /**
+   * Closure for @a disconnect_cb.
+   */
+  void *disconnect_cls;
+
+  /**
+   * Are we polling for incoming messages right now?
+   */
+  uint8_t in_receive;
+
+  /**
+   * Is this a master or slave channel?
+   */
+  uint8_t is_master;
+
+  /**
+   * Is this channel in the process of disconnecting from the service?
+   * #GNUNET_YES or #GNUNET_NO
+   */
+  uint8_t is_disconnecting;
+};
+
+
+/**
+ * Handle for the master of a PSYC channel.
+ */
+struct GNUNET_PSYC_Master
+{
+  struct GNUNET_PSYC_Channel chn;
+
+  GNUNET_PSYC_MasterStartCallback start_cb;
+
+  /**
+   * Join request callback.
+   */
+  GNUNET_PSYC_JoinRequestCallback join_req_cb;
+
+  /**
+   * Closure for the callbacks.
+   */
+  void *cb_cls;
+};
+
+
+/**
+ * Handle for a PSYC channel slave.
+ */
+struct GNUNET_PSYC_Slave
+{
+  struct GNUNET_PSYC_Channel chn;
+
+  GNUNET_PSYC_SlaveConnectCallback connect_cb;
+
+  GNUNET_PSYC_JoinDecisionCallback join_dcsn_cb;
+
+  /**
+   * Closure for the callbacks.
+   */
+  void *cb_cls;
+};
+
+
+/**
+ * Handle that identifies a join request.
+ *
+ * Used to match calls to #GNUNET_PSYC_JoinRequestCallback to the
+ * corresponding calls to GNUNET_PSYC_join_decision().
+ */
+struct GNUNET_PSYC_JoinHandle
+{
+  struct GNUNET_PSYC_Master *mst;
+  struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
+};
+
+
+/**
+ * Handle for a pending PSYC transmission operation.
+ */
+struct GNUNET_PSYC_SlaveTransmitHandle
+{
+
+};
+
+
+struct GNUNET_PSYC_HistoryRequest
+{
+  /**
+   * Channel.
+   */
+  struct GNUNET_PSYC_Channel *chn;
+
+  /**
+   * Operation ID.
+   */
+  uint64_t op_id;
+
+  /**
+   * Message handler.
+   */
+  struct GNUNET_PSYC_ReceiveHandle *recv;
+
+  /**
+   * Function to call when the operation finished.
+   */
+  GNUNET_ResultCallback result_cb;
+
+  /**
+   * Closure for @a result_cb.
+   */
+  void *cls;
+};
+
+
+struct GNUNET_PSYC_StateRequest
+{
+  /**
+   * Channel.
+   */
+  struct GNUNET_PSYC_Channel *chn;
+
+  /**
+   * Operation ID.
+   */
+  uint64_t op_id;
+
+  /**
+   * State variable result callback.
+   */
+  GNUNET_PSYC_StateVarCallback var_cb;
+
+  /**
+   * Function to call when the operation finished.
+   */
+  GNUNET_ResultCallback result_cb;
+
+  /**
+   * Closure for @a result_cb.
+   */
+  void *cls;
+};
+
+
+static int
+check_channel_result (void *cls,
+                      const struct GNUNET_OperationResultMessage *res)
+{
+  return GNUNET_OK;
+}
+
+
+static void
+handle_channel_result (void *cls,
+                       const struct GNUNET_OperationResultMessage *res)
+{
+  struct GNUNET_PSYC_Channel *chn = cls;
+
+  uint16_t size = ntohs (res->header.size);
+  if (size < sizeof (*res))
+  { /* Error, message too small. */
+    GNUNET_break (0);
+    return;
+  }
+
+  uint16_t data_size = size - sizeof (*res);
+  const char *data = (0 < data_size) ? (void *) &res[1] : NULL;
+  GNUNET_OP_result (chn->op, GNUNET_ntohll (res->op_id),
+                    GNUNET_ntohll (res->result_code),
+                    data, data_size, NULL);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "handle_channel_result: Received result message with OP ID %" PRIu64 "\n",
+              GNUNET_ntohll (res->op_id));
+}
+
+
+static void
+op_recv_history_result (void *cls, int64_t result,
+                        const void *data, uint16_t data_size)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received history replay result: %" PRId64 ".\n", result);
+
+  struct GNUNET_PSYC_HistoryRequest *hist = cls;
+
+  if (NULL != hist->result_cb)
+    hist->result_cb (hist->cls, result, data, data_size);
+
+  GNUNET_PSYC_receive_destroy (hist->recv);
+  GNUNET_free (hist);
+}
+
+
+static void
+op_recv_state_result (void *cls, int64_t result,
+                      const void *data, uint16_t data_size)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received state request result: %" PRId64 ".\n", result);
+
+  struct GNUNET_PSYC_StateRequest *sr = cls;
+
+  if (NULL != sr->result_cb)
+    sr->result_cb (sr->cls, result, data, data_size);
+
+  GNUNET_free (sr);
+}
+
+
+static int
+check_channel_history_result (void *cls,
+                              const struct GNUNET_OperationResultMessage *res)
+{
+  struct GNUNET_PSYC_MessageHeader *
+    pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res);
+  uint16_t size = ntohs (res->header.size);
+
+  if ( (NULL == pmsg) ||
+       (size < sizeof (*res) + sizeof (*pmsg)) )
+  { /* Error, message too small. */
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  return GNUNET_OK;
+}
+
+
+static void
+handle_channel_history_result (void *cls,
+                               const struct GNUNET_OperationResultMessage *res)
+{
+  struct GNUNET_PSYC_Channel *chn = cls;
+  struct GNUNET_PSYC_MessageHeader *
+    pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res);
+  GNUNET_ResultCallback result_cb = NULL;
+  struct GNUNET_PSYC_HistoryRequest *hist = NULL;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%p Received historic fragment for message #%" PRIu64 ".\n",
+       chn,
+       GNUNET_ntohll (pmsg->message_id));
+
+  if (GNUNET_YES != GNUNET_OP_get (chn->op,
+                                   GNUNET_ntohll (res->op_id),
+                                   &result_cb, (void *) &hist, NULL))
+  { /* Operation not found. */
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "%p Replay operation not found for historic fragment of message #%"
+         PRIu64 ".\n",
+         chn, GNUNET_ntohll (pmsg->message_id));
+    return;
+  }
+
+  GNUNET_PSYC_receive_message (hist->recv,
+                               (const struct GNUNET_PSYC_MessageHeader *) pmsg);
+}
+
+
+static int
+check_channel_state_result (void *cls,
+                            const struct GNUNET_OperationResultMessage *res)
+{
+  const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res);
+  uint16_t mod_size;
+  uint16_t size;
+
+  if (NULL == mod)
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  mod_size = ntohs (mod->size);
+  size = ntohs (res->header.size);
+  if (size - sizeof (*res) != mod_size)
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  return GNUNET_OK;
+}
+
+
+static void
+handle_channel_state_result (void *cls,
+                             const struct GNUNET_OperationResultMessage *res)
+{
+  struct GNUNET_PSYC_Channel *chn = cls;
+
+  GNUNET_ResultCallback result_cb = NULL;
+  struct GNUNET_PSYC_StateRequest *sr = NULL;
+
+  if (GNUNET_YES != GNUNET_OP_get (chn->op,
+                                   GNUNET_ntohll (res->op_id),
+                                   &result_cb, (void *) &sr, NULL))
+  { /* Operation not found. */
+    return;
+  }
+
+  const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res);
+  if (NULL == mod)
+  {
+    GNUNET_break_op (0);
+    return;
+  }
+  uint16_t mod_size = ntohs (mod->size);
+
+  switch (ntohs (mod->type))
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+  {
+    const struct GNUNET_PSYC_MessageModifier *
+      pmod = (const struct GNUNET_PSYC_MessageModifier *) mod;
+
+    const char *name = (const char *) &pmod[1];
+    uint16_t name_size = ntohs (pmod->name_size);
+    if (0 == name_size
+        || mod_size - sizeof (*pmod) < name_size
+        || '\0' != name[name_size - 1])
+    {
+      GNUNET_break_op (0);
+      return;
+    }
+    sr->var_cb (sr->cls, mod, name, name + name_size,
+                ntohs (pmod->header.size) - sizeof (*pmod),
+                ntohs (pmod->value_size));
+    break;
+  }
+
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+    sr->var_cb (sr->cls, mod, NULL, (const char *) &mod[1],
+                mod_size - sizeof (*mod), 0);
+    break;
+  }
+}
+
+
+static int
+check_channel_message (void *cls,
+                       const struct GNUNET_PSYC_MessageHeader *pmsg)
+{
+  return GNUNET_OK;
+}
+
+
+static void
+handle_channel_message (void *cls,
+                        const struct GNUNET_PSYC_MessageHeader *pmsg)
+{
+  struct GNUNET_PSYC_Channel *chn = cls;
+
+  GNUNET_PSYC_receive_message (chn->recv, pmsg);
+}
+
+
+static void
+handle_channel_message_ack (void *cls,
+                            const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_PSYC_Channel *chn = cls;
+
+  GNUNET_PSYC_transmit_got_ack (chn->tmit);
+}
+
+
+static void
+handle_master_start_ack (void *cls,
+                         const struct GNUNET_PSYC_CountersResultMessage *cres)
+{
+  struct GNUNET_PSYC_Master *mst = cls;
+
+  int32_t result = ntohl (cres->result_code);
+  if (GNUNET_OK != result && GNUNET_NO != result)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR, "Could not start master: %ld\n", result);
+    GNUNET_break (0);
+    /* FIXME: disconnect */
+  }
+  if (NULL != mst->start_cb)
+    mst->start_cb (mst->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
+}
+
+
+static int
+check_master_join_request (void *cls,
+                           const struct GNUNET_PSYC_JoinRequestMessage *req)
+{
+  if ( ((sizeof (*req) + sizeof (struct GNUNET_PSYC_Message)) <= ntohs (req->header.size)) &&
+       (NULL == GNUNET_MQ_extract_nested_mh (req)) )
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+  return GNUNET_OK;
+}
+
+
+static void
+handle_master_join_request (void *cls,
+                            const struct GNUNET_PSYC_JoinRequestMessage *req)
+{
+  struct GNUNET_PSYC_Master *mst = cls;
+
+  if (NULL == mst->join_req_cb)
+    return;
+
+  const struct GNUNET_PSYC_Message *join_msg = NULL;
+  if (sizeof (*req) + sizeof (*join_msg) <= ntohs (req->header.size))
+  {
+    join_msg = (struct GNUNET_PSYC_Message *) GNUNET_MQ_extract_nested_mh (req);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Received join_msg of type %u and size %u.\n",
+         ntohs (join_msg->header.type),
+         ntohs (join_msg->header.size));
+  }
+
+  struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh));
+  jh->mst = mst;
+  jh->slave_pub_key = req->slave_pub_key;
+
+  if (NULL != mst->join_req_cb)
+    mst->join_req_cb (mst->cb_cls, req, &req->slave_pub_key, join_msg, jh);
+}
+
+
+static void
+handle_slave_join_ack (void *cls,
+                       const struct GNUNET_PSYC_CountersResultMessage *cres)
+{
+  struct GNUNET_PSYC_Slave *slv = cls;
+
+  int32_t result = ntohl (cres->result_code);
+  if (GNUNET_YES != result && GNUNET_NO != result)
+  {
+    LOG (GNUNET_ERROR_TYPE_ERROR, "Could not join slave.\n");
+    GNUNET_break (0);
+    /* FIXME: disconnect */
+  }
+  if (NULL != slv->connect_cb)
+    slv->connect_cb (slv->cb_cls, result, GNUNET_ntohll (cres->max_message_id));
+}
+
+
+static int
+check_slave_join_decision (void *cls,
+                           const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
+{
+  return GNUNET_OK;
+}
+
+
+static void
+handle_slave_join_decision (void *cls,
+                            const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
+{
+  struct GNUNET_PSYC_Slave *slv = cls;
+
+  struct GNUNET_PSYC_Message *pmsg = NULL;
+  if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg))
+    pmsg = (struct GNUNET_PSYC_Message *) &dcsn[1];
+
+  if (NULL != slv->join_dcsn_cb)
+    slv->join_dcsn_cb (slv->cb_cls, dcsn, ntohl (dcsn->is_admitted), pmsg);
+}
+
+
+static void
+channel_cleanup (struct GNUNET_PSYC_Channel *chn)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "cleaning up channel %p\n",
+              chn);
+  if (NULL != chn->tmit)
+  {
+    GNUNET_PSYC_transmit_destroy (chn->tmit);
+    chn->tmit = NULL;
+  }
+  if (NULL != chn->recv)
+  {
+
+    GNUNET_PSYC_receive_destroy (chn->recv);
+    chn->recv = NULL;
+  }
+  if (NULL != chn->connect_env)
+  {
+    GNUNET_MQ_discard (chn->connect_env);
+    chn->connect_env = NULL;
+  }
+  if (NULL != chn->mq)
+  {
+    GNUNET_MQ_destroy (chn->mq);
+    chn->mq = NULL;
+  }
+  if (NULL != chn->disconnect_cb)
+  {
+    chn->disconnect_cb (chn->disconnect_cls);
+    chn->disconnect_cb = NULL;
+  }
+  GNUNET_free (chn);
+}
+
+
+static void
+handle_channel_part_ack (void *cls,
+                         const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_PSYC_Channel *chn = cls;
+
+  channel_cleanup (chn); 
+}
+
+
+/*** MASTER ***/
+
+
+static void
+master_connect (struct GNUNET_PSYC_Master *mst);
+
+
+static void
+master_reconnect (void *cls)
+{
+  master_connect (cls);
+}
+
+
+/**
+ * Master client disconnected from service.
+ *
+ * Reconnect after backoff period.
+ */
+static void
+master_disconnected (void *cls, enum GNUNET_MQ_Error error)
+{
+  struct GNUNET_PSYC_Master *mst = cls;
+  struct GNUNET_PSYC_Channel *chn = &mst->chn;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Master client disconnected (%d), re-connecting\n",
+       (int) error);
+  if (NULL != chn->tmit)
+  {
+    GNUNET_PSYC_transmit_destroy (chn->tmit);
+    chn->tmit = NULL;
+  }
+  if (NULL != chn->mq)
+  {
+    GNUNET_MQ_destroy (chn->mq);
+    chn->mq = NULL;
+  }
+  chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
+                                                      master_reconnect,
+                                                      mst);
+  chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
+}
+
+
+static void
+master_connect (struct GNUNET_PSYC_Master *mst)
+{
+  struct GNUNET_PSYC_Channel *chn = &mst->chn;
+
+  struct GNUNET_MQ_MessageHandler handlers[] = {
+    GNUNET_MQ_hd_fixed_size (master_start_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK,
+                             struct GNUNET_PSYC_CountersResultMessage,
+                             mst),
+    GNUNET_MQ_hd_var_size (master_join_request,
+                           GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST,
+                           struct GNUNET_PSYC_JoinRequestMessage,
+                           mst),
+    GNUNET_MQ_hd_fixed_size (channel_part_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
+                             struct GNUNET_MessageHeader,
+                             chn),
+    GNUNET_MQ_hd_var_size (channel_message,
+                           GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
+                           struct GNUNET_PSYC_MessageHeader,
+                           chn),
+    GNUNET_MQ_hd_fixed_size (channel_message_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK,
+                             struct GNUNET_MessageHeader,
+                             chn),
+    GNUNET_MQ_hd_var_size (channel_history_result,
+                           GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT,
+                           struct GNUNET_OperationResultMessage,
+                           chn),
+    GNUNET_MQ_hd_var_size (channel_state_result,
+                           GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
+                           struct GNUNET_OperationResultMessage,
+                           chn),
+    GNUNET_MQ_hd_var_size (channel_result,
+                           GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
+                           struct GNUNET_OperationResultMessage,
+                           chn),
+    GNUNET_MQ_handler_end ()
+  };
+
+  chn->mq = GNUNET_CLIENT_connect (chn->cfg,
+                                   "psyc",
+                                   handlers,
+                                   &master_disconnected,
+                                   mst);
+  GNUNET_assert (NULL != chn->mq);
+  chn->tmit = GNUNET_PSYC_transmit_create (chn->mq);
+
+  GNUNET_MQ_send_copy (chn->mq, chn->connect_env);
+}
+
+
+/**
+ * Start a PSYC master channel.
+ *
+ * Will start a multicast group identified by the given ECC key.  Messages
+ * received from group members will be given to the respective handler methods.
+ * If a new member wants to join a group, the "join" method handler will be
+ * invoked; the join handler must then generate a "join" message to approve the
+ * joining of the new member.  The channel can also change group membership
+ * without explicit requests.  Note that PSYC doesn't itself "understand" join
+ * or part messages, the respective methods must call other PSYC functions to
+ * inform PSYC about the meaning of the respective events.
+ *
+ * @param cfg  Configuration to use (to connect to PSYC service).
+ * @param channel_key  ECC key that will be used to sign messages for this
+ *        PSYC session. The public key is used to identify the PSYC channel.
+ *        Note that end-users will usually not use the private key directly, but
+ *        rather look it up in GNS for places managed by other users, or select
+ *        a file with the private key(s) when setting up their own channels
+ *        FIXME: we'll likely want to use NOT the p521 curve here, but a cheaper
+ *        one in the future.
+ * @param policy  Channel policy specifying join and history restrictions.
+ *        Used to automate join decisions.
+ * @param message_cb  Function to invoke on message parts received from slaves.
+ * @param join_request_cb  Function to invoke when a slave wants to join.
+ * @param master_start_cb  Function to invoke after the channel master started.
+ * @param cls  Closure for @a method and @a join_cb.
+ *
+ * @return Handle for the channel master, NULL on error.
+ */
+struct GNUNET_PSYC_Master *
+GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
+                          const struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key,
+                          enum GNUNET_PSYC_Policy policy,
+                          GNUNET_PSYC_MasterStartCallback start_cb,
+                          GNUNET_PSYC_JoinRequestCallback join_request_cb,
+                          GNUNET_PSYC_MessageCallback message_cb,
+                          GNUNET_PSYC_MessagePartCallback message_part_cb,
+                          void *cls)
+{
+  struct GNUNET_PSYC_Master *mst = GNUNET_new (struct GNUNET_PSYC_Master);
+  struct GNUNET_PSYC_Channel *chn = &mst->chn;
+  struct MasterStartRequest *req;
+
+  chn->connect_env = GNUNET_MQ_msg (req,
+                                    GNUNET_MESSAGE_TYPE_PSYC_MASTER_START);
+  req->channel_key = *channel_key;
+  req->policy = policy;
+
+  chn->cfg = cfg;
+  chn->is_master = GNUNET_YES;
+  chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
+
+  chn->op = GNUNET_OP_create ();
+  chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
+
+  mst->start_cb = start_cb;
+  mst->join_req_cb = join_request_cb;
+  mst->cb_cls = cls;
+
+  master_connect (mst);
+  return mst;
+}
+
+
+/**
+ * Stop a PSYC master channel.
+ *
+ * @param master PSYC channel master to stop.
+ * @param keep_active  FIXME
+ */
+void
+GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst,
+                         int keep_active,
+                         GNUNET_ContinuationCallback stop_cb,
+                         void *stop_cls)
+{
+  struct GNUNET_PSYC_Channel *chn = &mst->chn;
+  struct GNUNET_MQ_Envelope *env;
+
+  chn->is_disconnecting = GNUNET_YES;
+  chn->disconnect_cb = stop_cb;
+  chn->disconnect_cls = stop_cls;
+  env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
+  GNUNET_MQ_send (chn->mq, env);
+}
+
+
+/**
+ * Function to call with the decision made for a join request.
+ *
+ * Must be called once and only once in response to an invocation of the
+ * #GNUNET_PSYC_JoinCallback.
+ *
+ * @param jh Join request handle.
+ * @param is_admitted  #GNUNET_YES    if the join is approved,
+ *                     #GNUNET_NO     if it is disapproved,
+ *                     #GNUNET_SYSERR if we cannot answer the request.
+ * @param relay_count Number of relays given.
+ * @param relays Array of suggested peers that might be useful relays to use
+ *        when joining the multicast group (essentially a list of peers that
+ *        are already part of the multicast group and might thus be willing
+ *        to help with routing).  If empty, only this local peer (which must
+ *        be the multicast origin) is a good candidate for building the
+ *        multicast tree.  Note that it is unnecessary to specify our own
+ *        peer identity in this array.
+ * @param join_resp  Application-dependent join response message.
+ *
+ * @return #GNUNET_OK on success,
+ *         #GNUNET_SYSERR if the message is too large.
+ */
+int
+GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
+                           int is_admitted,
+                           uint32_t relay_count,
+                           const struct GNUNET_PeerIdentity *relays,
+                           const struct GNUNET_PSYC_Message *join_resp)
+{
+  struct GNUNET_PSYC_Channel *chn = &jh->mst->chn;
+  struct GNUNET_PSYC_JoinDecisionMessage *dcsn;
+  uint16_t join_resp_size
+    = (NULL != join_resp) ? ntohs (join_resp->header.size) : 0;
+  uint16_t relay_size = relay_count * sizeof (*relays);
+
+  if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
+      < sizeof (*dcsn) + relay_size + join_resp_size)
+    return GNUNET_SYSERR;
+
+  struct GNUNET_MQ_Envelope *
+    env = GNUNET_MQ_msg_extra (dcsn, relay_size + join_resp_size,
+                               GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION);
+  dcsn->is_admitted = htonl (is_admitted);
+  dcsn->slave_pub_key = jh->slave_pub_key;
+
+  if (0 < join_resp_size)
+    GNUNET_memcpy (&dcsn[1], join_resp, join_resp_size);
+
+  GNUNET_MQ_send (chn->mq, env);
+  GNUNET_free (jh);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Send a message to call a method to all members in the PSYC channel.
+ *
+ * @param master Handle to the PSYC channel.
+ * @param method_name Which method should be invoked.
+ * @param notify_mod Function to call to obtain modifiers.
+ * @param notify_data Function to call to obtain fragments of the data.
+ * @param notify_cls Closure for @a notify_mod and @a notify_data.
+ * @param flags Flags for the message being transmitted.
+ *
+ * @return Transmission handle, NULL on error (i.e. more than one request queued).
+ */
+struct GNUNET_PSYC_MasterTransmitHandle *
+GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *mst,
+                             const char *method_name,
+                             GNUNET_PSYC_TransmitNotifyModifier notify_mod,
+                             GNUNET_PSYC_TransmitNotifyData notify_data,
+                             void *notify_cls,
+                             enum GNUNET_PSYC_MasterTransmitFlags flags)
+{
+  if (GNUNET_OK
+      == GNUNET_PSYC_transmit_message (mst->chn.tmit, method_name, NULL,
+                                       notify_mod, notify_data, notify_cls,
+                                       flags))
+    return (struct GNUNET_PSYC_MasterTransmitHandle *) mst->chn.tmit;
+  else
+    return NULL;
+}
+
+
+/**
+ * Resume transmission to the channel.
+ *
+ * @param tmit  Handle of the request that is being resumed.
+ */
+void
+GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *tmit)
+{
+  GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit);
+}
+
+
+/**
+ * Abort transmission request to the channel.
+ *
+ * @param tmit  Handle of the request that is being aborted.
+ */
+void
+GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *tmit)
+{
+  GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit);
+}
+
+
+/**
+ * Convert a channel @a master to a @e channel handle to access the @e channel
+ * APIs.
+ *
+ * @param master Channel master handle.
+ *
+ * @return Channel handle, valid for as long as @a master is valid.
+ */
+struct GNUNET_PSYC_Channel *
+GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
+{
+  return &master->chn;
+}
+
+
+/*** SLAVE ***/
+
+
+static void
+slave_connect (struct GNUNET_PSYC_Slave *slv);
+
+
+static void
+slave_reconnect (void *cls)
+{
+  slave_connect (cls);
+}
+
+
+/**
+ * Slave client disconnected from service.
+ *
+ * Reconnect after backoff period.
+ */
+static void
+slave_disconnected (void *cls,
+                    enum GNUNET_MQ_Error error)
+{
+  struct GNUNET_PSYC_Slave *slv = cls;
+  struct GNUNET_PSYC_Channel *chn = &slv->chn;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Slave client disconnected (%d), re-connecting\n",
+       (int) error);
+  if (NULL != chn->tmit)
+  {
+    GNUNET_PSYC_transmit_destroy (chn->tmit);
+    chn->tmit = NULL;
+  }
+  if (NULL != chn->mq)
+  {
+    GNUNET_MQ_destroy (chn->mq);
+    chn->mq = NULL;
+  }
+  chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
+                                                      &slave_reconnect,
+                                                      slv);
+  chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
+}
+
+
+static void
+slave_connect (struct GNUNET_PSYC_Slave *slv)
+{
+  struct GNUNET_PSYC_Channel *chn = &slv->chn;
+
+  struct GNUNET_MQ_MessageHandler handlers[] = {
+    GNUNET_MQ_hd_fixed_size (slave_join_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK,
+                             struct GNUNET_PSYC_CountersResultMessage,
+                             slv),
+    GNUNET_MQ_hd_var_size (slave_join_decision,
+                           GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
+                           struct GNUNET_PSYC_JoinDecisionMessage,
+                           slv),
+    GNUNET_MQ_hd_fixed_size (channel_part_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
+                             struct GNUNET_MessageHeader,
+                             chn),
+    GNUNET_MQ_hd_var_size (channel_message,
+                           GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
+                           struct GNUNET_PSYC_MessageHeader,
+                           chn),
+    GNUNET_MQ_hd_fixed_size (channel_message_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK,
+                             struct GNUNET_MessageHeader,
+                             chn),
+    GNUNET_MQ_hd_var_size (channel_history_result,
+                           GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT,
+                           struct GNUNET_OperationResultMessage,
+                           chn),
+    GNUNET_MQ_hd_var_size (channel_state_result,
+                           GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT,
+                           struct GNUNET_OperationResultMessage,
+                           chn),
+    GNUNET_MQ_hd_var_size (channel_result,
+                           GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE,
+                           struct GNUNET_OperationResultMessage,
+                           chn),
+    GNUNET_MQ_handler_end ()
+  };
+
+  chn->mq = GNUNET_CLIENT_connect (chn->cfg,
+                                   "psyc",
+                                   handlers,
+                                   &slave_disconnected,
+                                   slv);
+  if (NULL == chn->mq)
+  {
+    chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay,
+                                                        &slave_reconnect,
+                                                        slv);
+    chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay);
+    return;
+  }
+  chn->tmit = GNUNET_PSYC_transmit_create (chn->mq);
+
+  GNUNET_MQ_send_copy (chn->mq, chn->connect_env);
+}
+
+
+/**
+ * Join a PSYC channel.
+ *
+ * The entity joining is always the local peer.  The user must immediately use
+ * the GNUNET_PSYC_slave_transmit() functions to transmit a @e join_msg to the
+ * channel; if the join request succeeds, the channel state (and @e recent
+ * method calls) will be replayed to the joining member.  There is no explicit
+ * notification on failure (as the channel may simply take days to approve,
+ * and disapproval is simply being ignored).
+ *
+ * @param cfg
+ *        Configuration to use.
+ * @param channel_key  ECC public key that identifies the channel we wish to join.
+ * @param slave_key  ECC private-public key pair that identifies the slave, and
+ *        used by multicast to sign the join request and subsequent unicast
+ *        requests sent to the master.
+ * @param origin  Peer identity of the origin.
+ * @param relay_count  Number of peers in the @a relays array.
+ * @param relays  Peer identities of members of the multicast group, which serve
+ *        as relays and used to join the group at.
+ * @param message_cb  Function to invoke on message parts received from the
+ *        channel, typically at least contains method handlers for @e join and
+ *        @e part.
+ * @param slave_connect_cb  Function invoked once we have connected to the
+ *        PSYC service.
+ * @param join_decision_cb  Function invoked once we have received a join
+ *        decision.
+ * @param cls  Closure for @a message_cb and @a slave_joined_cb.
+ * @param method_name  Method name for the join request.
+ * @param env  Environment containing transient variables for the request, or NULL.
+ * @param data  Payload for the join message.
+ * @param data_size  Number of bytes in @a data.
+ *
+ * @return Handle for the slave, NULL on error.
+ */
+struct GNUNET_PSYC_Slave *
+GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
+                        const struct GNUNET_CRYPTO_EddsaPublicKey *channel_pub_key,
+                        const struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key,
+                        enum GNUNET_PSYC_SlaveJoinFlags flags,
+                        const struct GNUNET_PeerIdentity *origin,
+                        uint32_t relay_count,
+                        const struct GNUNET_PeerIdentity *relays,
+                        GNUNET_PSYC_MessageCallback message_cb,
+                        GNUNET_PSYC_MessagePartCallback message_part_cb,
+                        GNUNET_PSYC_SlaveConnectCallback connect_cb,
+                        GNUNET_PSYC_JoinDecisionCallback join_decision_cb,
+                        void *cls,
+                        const struct GNUNET_PSYC_Message *join_msg)
+{
+  struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv));
+  struct GNUNET_PSYC_Channel *chn = &slv->chn;
+  uint16_t relay_size = relay_count * sizeof (*relays);
+  uint16_t join_msg_size;
+  if (NULL == join_msg)
+    join_msg_size = 0;
+  else
+    join_msg_size = ntohs (join_msg->header.size);
+
+  struct SlaveJoinRequest *req;
+  chn->connect_env = GNUNET_MQ_msg_extra (req, relay_size + join_msg_size,
+                                          GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN);
+  req->channel_pub_key = *channel_pub_key;
+  req->slave_key = *slave_key;
+  req->origin = *origin;
+  req->relay_count = htonl (relay_count);
+  req->flags = htonl (flags);
+
+  if (0 < relay_size)
+    GNUNET_memcpy (&req[1], relays, relay_size);
+
+  if (NULL != join_msg)
+    GNUNET_memcpy ((char *) &req[1] + relay_size, join_msg, join_msg_size);
+
+  chn->cfg = cfg;
+  chn->is_master = GNUNET_NO;
+  chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS;
+
+  chn->op = GNUNET_OP_create ();
+  chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
+
+  slv->connect_cb = connect_cb;
+  slv->join_dcsn_cb = join_decision_cb;
+  slv->cb_cls = cls;
+
+  slave_connect (slv);
+  return slv;
+}
+
+
+/**
+ * Part a PSYC channel.
+ *
+ * Will terminate the connection to the PSYC service.  Polite clients should
+ * first explicitly send a part request (via GNUNET_PSYC_slave_transmit()).
+ *
+ * @param slave Slave handle.
+ */
+void
+GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv,
+                        int keep_active,
+                        GNUNET_ContinuationCallback part_cb,
+                        void *part_cls)
+{
+  struct GNUNET_PSYC_Channel *chn = &slv->chn;
+  struct GNUNET_MQ_Envelope *env;
+
+  chn->is_disconnecting = GNUNET_YES;
+  chn->disconnect_cb = part_cb;
+  chn->disconnect_cls = part_cls;
+  env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
+  GNUNET_MQ_send (chn->mq, env);
+}
+
+
+/**
+ * Request a message to be sent to the channel master.
+ *
+ * @param slave Slave handle.
+ * @param method_name Which (PSYC) method should be invoked (on host).
+ * @param notify_mod Function to call to obtain modifiers.
+ * @param notify_data Function to call to obtain fragments of the data.
+ * @param notify_cls Closure for @a notify.
+ * @param flags Flags for the message being transmitted.
+ *
+ * @return Transmission handle, NULL on error (i.e. more than one request
+ *         queued).
+ */
+struct GNUNET_PSYC_SlaveTransmitHandle *
+GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slv,
+                            const char *method_name,
+                            GNUNET_PSYC_TransmitNotifyModifier notify_mod,
+                            GNUNET_PSYC_TransmitNotifyData notify_data,
+                            void *notify_cls,
+                            enum GNUNET_PSYC_SlaveTransmitFlags flags)
+
+{
+  if (GNUNET_OK
+      == GNUNET_PSYC_transmit_message (slv->chn.tmit, method_name, NULL,
+                                       notify_mod, notify_data, notify_cls,
+                                       flags))
+    return (struct GNUNET_PSYC_SlaveTransmitHandle *) slv->chn.tmit;
+  else
+    return NULL;
+}
+
+
+/**
+ * Resume transmission to the master.
+ *
+ * @param tmit Handle of the request that is being resumed.
+ */
+void
+GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *tmit)
+{
+  GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit);
+}
+
+
+/**
+ * Abort transmission request to master.
+ *
+ * @param tmit Handle of the request that is being aborted.
+ */
+void
+GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *tmit)
+{
+  GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit);
+}
+
+
+/**
+ * Convert @a slave to a @e channel handle to access the @e channel APIs.
+ *
+ * @param slv Slave handle.
+ *
+ * @return Channel handle, valid for as long as @a slave is valid.
+ */
+struct GNUNET_PSYC_Channel *
+GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slv)
+{
+  return &slv->chn;
+}
+
+
+/**
+ * Add a slave to the channel's membership list.
+ *
+ * Note that this will NOT generate any PSYC traffic, it will merely update the
+ * local database to modify how we react to <em>membership test</em> queries.
+ * The channel master still needs to explicitly transmit a @e join message to
+ * notify other channel members and they then also must still call this function
+ * in their respective methods handling the @e join message.  This way, how @e
+ * join and @e part operations are exactly implemented is still up to the
+ * application; for example, there might be a @e part_all method to kick out
+ * everyone.
+ *
+ * Note that channel slaves are explicitly trusted to execute such methods
+ * correctly; not doing so correctly will result in either denying other slaves
+ * access or offering access to channel data to non-members.
+ *
+ * @param chn
+ *        Channel handle.
+ * @param slave_pub_key
+ *        Identity of channel slave to add.
+ * @param announced_at
+ *        ID of the message that announced the membership change.
+ * @param effective_since
+ *        Addition of slave is in effect since this message ID.
+ * @param result_cb
+ *        Function to call with the result of the operation.
+ *        The @e result_code argument is #GNUNET_OK on success, or
+ *        #GNUNET_SYSERR on error.  In case of an error, the @e data argument
+ *        can contain an optional error message.
+ * @param cls
+ *        Closure for @a result_cb.
+ */
+void
+GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn,
+                               const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
+                               uint64_t announced_at,
+                               uint64_t effective_since,
+                               GNUNET_ResultCallback result_cb,
+                               void *cls)
+{
+  struct ChannelMembershipStoreRequest *req;
+  struct GNUNET_MQ_Envelope *
+    env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
+  req->slave_pub_key = *slave_pub_key;
+  req->announced_at = GNUNET_htonll (announced_at);
+  req->effective_since = GNUNET_htonll (effective_since);
+  req->did_join = GNUNET_YES;
+  req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL));
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "GNUNET_PSYC_channel_slave_add, OP ID: %" PRIu64 "\n",
+              GNUNET_ntohll (req->op_id));
+  GNUNET_MQ_send (chn->mq, env);
+}
+
+
+/**
+ * Remove a slave from the channel's membership list.
+ *
+ * Note that this will NOT generate any PSYC traffic, it will merely update the
+ * local database to modify how we react to <em>membership test</em> queries.
+ * The channel master still needs to explicitly transmit a @e part message to
+ * notify other channel members and they then also must still call this function
+ * in their respective methods handling the @e part message.  This way, how
+ * @e join and @e part operations are exactly implemented is still up to the
+ * application; for example, there might be a @e part_all message to kick out
+ * everyone.
+ *
+ * Note that channel members are explicitly trusted to perform these
+ * operations correctly; not doing so correctly will result in either
+ * denying members access or offering access to channel data to
+ * non-members.
+ *
+ * @param chn
+ *        Channel handle.
+ * @param slave_pub_key
+ *        Identity of channel slave to remove.
+ * @param announced_at
+ *        ID of the message that announced the membership change.
+ * @param result_cb
+ *        Function to call with the result of the operation.
+ *        The @e result_code argument is #GNUNET_OK on success, or
+ *        #GNUNET_SYSERR on error.  In case of an error, the @e data argument
+ *        can contain an optional error message.
+ * @param cls
+ *        Closure for @a result_cb.
+ */
+void
+GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn,
+                                  const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key,
+                                  uint64_t announced_at,
+                                  GNUNET_ResultCallback result_cb,
+                                  void *cls)
+{
+  struct ChannelMembershipStoreRequest *req;
+  struct GNUNET_MQ_Envelope *
+    env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE);
+  req->slave_pub_key = *slave_pub_key;
+  req->announced_at = GNUNET_htonll (announced_at);
+  req->did_join = GNUNET_NO;
+  req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL));
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "GNUNET_PSYC_channel_slave_remove, OP ID: %" PRIu64 "\n",
+              GNUNET_ntohll (req->op_id));
+  GNUNET_MQ_send (chn->mq, env);
+}
+
+
+static struct GNUNET_PSYC_HistoryRequest *
+channel_history_replay (struct GNUNET_PSYC_Channel *chn,
+                        uint64_t start_message_id,
+                        uint64_t end_message_id,
+                        uint64_t message_limit,
+                        const char *method_prefix,
+                        uint32_t flags,
+                        GNUNET_PSYC_MessageCallback message_cb,
+                        GNUNET_PSYC_MessagePartCallback message_part_cb,
+                        GNUNET_ResultCallback result_cb,
+                        void *cls)
+{
+  struct GNUNET_PSYC_HistoryRequestMessage *req;
+  struct GNUNET_PSYC_HistoryRequest *hist = GNUNET_malloc (sizeof (*hist));
+  hist->chn = chn;
+  hist->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
+  hist->result_cb = result_cb;
+  hist->cls = cls;
+  hist->op_id = GNUNET_OP_add (chn->op, op_recv_history_result, hist, NULL);
+
+  GNUNET_assert (NULL != method_prefix);
+  uint16_t method_size = strnlen (method_prefix,
+                                  GNUNET_MAX_MESSAGE_SIZE
+                                  - sizeof (*req)) + 1;
+  GNUNET_assert ('\0' == method_prefix[method_size - 1]);
+
+  struct GNUNET_MQ_Envelope *
+    env = GNUNET_MQ_msg_extra (req, method_size,
+                               GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY);
+  req->start_message_id = GNUNET_htonll (start_message_id);
+  req->end_message_id = GNUNET_htonll (end_message_id);
+  req->message_limit = GNUNET_htonll (message_limit);
+  req->flags = htonl (flags);
+  req->op_id = GNUNET_htonll (hist->op_id);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "channel_history_replay, OP ID: %" PRIu64 "\n",
+              GNUNET_ntohll (req->op_id));
+  GNUNET_memcpy (&req[1], method_prefix, method_size);
+
+  GNUNET_MQ_send (chn->mq, env);
+  return hist;
+}
+
+
+/**
+ * Request to replay a part of the message history of the channel.
+ *
+ * Historic messages (but NOT the state at the time) will be replayed and given
+ * to the normal method handlers with a #GNUNET_PSYC_MESSAGE_HISTORIC flag set.
+ *
+ * Messages are retrieved from the local PSYCstore if available,
+ * otherwise requested from the network.
+ *
+ * @param channel
+ *        Which channel should be replayed?
+ * @param start_message_id
+ *        Earliest interesting point in history.
+ * @param end_message_id
+ *        Last (inclusive) interesting point in history.
+ * @param method_prefix
+ *        Retrieve only messages with a matching method prefix.
+ * @param flags
+ *        OR'ed enum GNUNET_PSYC_HistoryReplayFlags
+ * @param result_cb
+ *        Function to call when the requested history has been fully replayed.
+ * @param cls
+ *        Closure for the callbacks.
+ *
+ * @return Handle to cancel history replay operation.
+ */
+struct GNUNET_PSYC_HistoryRequest *
+GNUNET_PSYC_channel_history_replay (struct GNUNET_PSYC_Channel *chn,
+                                    uint64_t start_message_id,
+                                    uint64_t end_message_id,
+                                    const char *method_prefix,
+                                    uint32_t flags,
+                                    GNUNET_PSYC_MessageCallback message_cb,
+                                    GNUNET_PSYC_MessagePartCallback message_part_cb,
+                                    GNUNET_ResultCallback result_cb,
+                                    void *cls)
+{
+  return channel_history_replay (chn, start_message_id, end_message_id, 0,
+                                 method_prefix, flags,
+                                 message_cb, message_part_cb, result_cb, cls);
+}
+
+
+/**
+ * Request to replay the latest messages from the message history of the channel.
+ *
+ * Historic messages (but NOT the state at the time) will be replayed (given to
+ * the normal method handlers) if available and if access is permitted.
+ *
+ * @param channel
+ *        Which channel should be replayed?
+ * @param message_limit
+ *        Maximum number of messages to replay.
+ * @param method_prefix
+ *        Retrieve only messages with a matching method prefix.
+ *        Use NULL or "" to retrieve all.
+ * @param flags
+ *        OR'ed enum GNUNET_PSYC_HistoryReplayFlags
+ * @param result_cb
+ *        Function to call when the requested history has been fully replayed.
+ * @param cls
+ *        Closure for the callbacks.
+ *
+ * @return Handle to cancel history replay operation.
+ */
+struct GNUNET_PSYC_HistoryRequest *
+GNUNET_PSYC_channel_history_replay_latest (struct GNUNET_PSYC_Channel *chn,
+                                           uint64_t message_limit,
+                                           const char *method_prefix,
+                                           uint32_t flags,
+                                           GNUNET_PSYC_MessageCallback message_cb,
+                                           GNUNET_PSYC_MessagePartCallback message_part_cb,
+                                           GNUNET_ResultCallback result_cb,
+                                           void *cls)
+{
+  return channel_history_replay (chn, 0, 0, message_limit, method_prefix, flags,
+                                 message_cb, message_part_cb, result_cb, cls);
+}
+
+
+void
+GNUNET_PSYC_channel_history_replay_cancel (struct GNUNET_PSYC_Channel *channel,
+                                           struct GNUNET_PSYC_HistoryRequest *hist)
+{
+  GNUNET_PSYC_receive_destroy (hist->recv);
+  GNUNET_OP_remove (hist->chn->op, hist->op_id);
+  GNUNET_free (hist);
+}
+
+
+/**
+ * Retrieve the best matching channel state variable.
+ *
+ * If the requested variable name is not present in the state, the nearest
+ * less-specific name is matched; for example, requesting "_a_b" will match "_a"
+ * if "_a_b" does not exist.
+ *
+ * @param channel
+ *        Channel handle.
+ * @param full_name
+ *        Full name of the requested variable.
+ *        The actual variable returned might have a shorter name.
+ * @param var_cb
+ *        Function called once when a matching state variable is found.
+ *        Not called if there's no matching state variable.
+ * @param result_cb
+ *        Function called after the operation finished.
+ *        (i.e. all state variables have been returned via @a state_cb)
+ * @param cls
+ *        Closure for the callbacks.
+ */
+static struct GNUNET_PSYC_StateRequest *
+channel_state_get (struct GNUNET_PSYC_Channel *chn,
+                   uint16_t type, const char *name,
+                   GNUNET_PSYC_StateVarCallback var_cb,
+                   GNUNET_ResultCallback result_cb, void *cls)
+{
+  struct StateRequest *req;
+  struct GNUNET_PSYC_StateRequest *sr = GNUNET_malloc (sizeof (*sr));
+  sr->chn = chn;
+  sr->var_cb = var_cb;
+  sr->result_cb = result_cb;
+  sr->cls = cls;
+  sr->op_id = GNUNET_OP_add (chn->op, op_recv_state_result, sr, NULL);
+
+  GNUNET_assert (NULL != name);
+  size_t name_size = strnlen (name, GNUNET_MAX_MESSAGE_SIZE
+                              - sizeof (*req)) + 1;
+  struct GNUNET_MQ_Envelope *
+    env = GNUNET_MQ_msg_extra (req, name_size, type);
+  req->op_id = GNUNET_htonll (sr->op_id);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "channel_state_get, OP ID: %" PRIu64 "\n",
+              GNUNET_ntohll (req->op_id));
+
+  GNUNET_memcpy (&req[1], name, name_size);
+
+  GNUNET_MQ_send (chn->mq, env);
+  return sr;
+}
+
+
+/**
+ * Retrieve the best matching channel state variable.
+ *
+ * If the requested variable name is not present in the state, the nearest
+ * less-specific name is matched; for example, requesting "_a_b" will match "_a"
+ * if "_a_b" does not exist.
+ *
+ * @param channel
+ *        Channel handle.
+ * @param full_name
+ *        Full name of the requested variable.
+ *        The actual variable returned might have a shorter name.
+ * @param var_cb
+ *        Function called once when a matching state variable is found.
+ *        Not called if there's no matching state variable.
+ * @param result_cb
+ *        Function called after the operation finished.
+ *        (i.e. all state variables have been returned via @a state_cb)
+ * @param cls
+ *        Closure for the callbacks.
+ */
+struct GNUNET_PSYC_StateRequest *
+GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *chn,
+                               const char *full_name,
+                               GNUNET_PSYC_StateVarCallback var_cb,
+                               GNUNET_ResultCallback result_cb,
+                               void *cls)
+{
+  return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET,
+                            full_name, var_cb, result_cb, cls);
+
+}
+
+
+/**
+ * Return all channel state variables whose name matches a given prefix.
+ *
+ * A name matches if it starts with the given @a name_prefix, thus requesting
+ * the empty prefix ("") will match all values; requesting "_a_b" will also
+ * return values stored under "_a_b_c".
+ *
+ * The @a state_cb is invoked on all matching state variables asynchronously, as
+ * the state is stored in and retrieved from the PSYCstore,
+ *
+ * @param channel
+ *        Channel handle.
+ * @param name_prefix
+ *        Prefix of the state variable name to match.
+ * @param var_cb
+ *        Function called once when a matching state variable is found.
+ *        Not called if there's no matching state variable.
+ * @param result_cb
+ *        Function called after the operation finished.
+ *        (i.e. all state variables have been returned via @a state_cb)
+ * @param cls
+ *        Closure for the callbacks.
+ */
+struct GNUNET_PSYC_StateRequest *
+GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *chn,
+                                      const char *name_prefix,
+                                      GNUNET_PSYC_StateVarCallback var_cb,
+                                      GNUNET_ResultCallback result_cb,
+                                      void *cls)
+{
+  return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX,
+                            name_prefix, var_cb, result_cb, cls);
+}
+
+
+/**
+ * Cancel a state request operation.
+ *
+ * @param sr
+ *        Handle for the operation to cancel.
+ */
+void
+GNUNET_PSYC_channel_state_get_cancel (struct GNUNET_PSYC_StateRequest *sr)
+{
+  GNUNET_OP_remove (sr->chn->op, sr->op_id);
+  GNUNET_free (sr);
+}
+
+/* end of psyc_api.c */
diff --git a/src/psyc/psyc_test_lib.h b/src/psyc/psyc_test_lib.h
new file mode 100644
index 0000000..0ad9910
--- /dev/null
+++ b/src/psyc/psyc_test_lib.h
@@ -0,0 +1,67 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/test_psyc_api_join.c
+ * @brief library for writing psyc tests
+ * @author xrs
+ */
+
+#define MAX_TESTBED_OPS 32
+
+struct pctx
+{
+  int idx;
+  
+  struct GNUNET_TESTBED_Peer *testbed_peer;
+  
+  const struct GNUNET_PeerIdentity *peer_id;
+
+  const struct GNUNET_PeerIdentity *peer_id_master;
+
+  /**
+   * Used to simulate egos (not peerid)
+   */
+  const struct GNUNET_CRYPTO_EcdsaPrivateKey *id_key;
+
+  const struct GNUNET_CRYPTO_EcdsaPublicKey *id_pub_key;
+
+  /**
+   * Used to store either GNUNET_PSYC_Master or GNUNET_PSYC_Slave handle
+   */
+  void *psyc;
+
+  struct GNUNET_PSYC_Channel *channel;
+
+  const struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key;
+
+  struct GNUNET_CRYPTO_EddsaPublicKey *channel_pub_key;
+
+  int test_ok;
+};
+
+static struct GNUNET_SCHEDULER_Task *timeout_task_id;
+
+static int result = GNUNET_SYSERR;
+
+static struct GNUNET_TESTBED_Operation *op[MAX_TESTBED_OPS];
+
+static int op_cnt = 0;
+
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
new file mode 100644
index 0000000..b6e27bb
--- /dev/null
+++ b/src/psyc/test_psyc.c
@@ -0,0 +1,1018 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/test_psyc.c
+ * @brief Tests for the PSYC API.
+ * @author Gabor X Toth
+ * @author Christian Grothoff
+ */
+
+#include <inttypes.h>
+
+#include "platform.h"
+#include "gnunet_crypto_lib.h"
+#include "gnunet_common.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_testing_lib.h"
+#include "gnunet_psyc_util_lib.h"
+#include "gnunet_psyc_service.h"
+
+#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
+
+/**
+ * Return value from 'main'.
+ */
+static int res;
+
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+static struct GNUNET_PeerIdentity this_peer;
+
+/**
+ * Handle for task for timeout termination.
+ */
+static struct GNUNET_SCHEDULER_Task * end_badly_task;
+
+static struct GNUNET_PSYC_Master *mst;
+static struct GNUNET_PSYC_Slave *slv;
+
+static struct GNUNET_PSYC_Channel *mst_chn, *slv_chn;
+
+static struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key;
+static struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key;
+
+static struct GNUNET_CRYPTO_EddsaPublicKey channel_pub_key;
+static struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
+
+struct TransmitClosure
+{
+  struct GNUNET_PSYC_MasterTransmitHandle *mst_tmit;
+  struct GNUNET_PSYC_SlaveTransmitHandle *slv_tmit;
+  struct GNUNET_PSYC_Environment *env;
+  struct GNUNET_PSYC_Modifier *mod;
+  char *data[16];
+  const char *mod_value;
+  size_t mod_value_size;
+  uint8_t data_delay[16];
+  uint8_t data_count;
+  uint8_t paused;
+  uint8_t n;
+};
+
+static struct TransmitClosure *tmit;
+
+static uint8_t join_req_count, end_count;
+
+enum
+{
+  TEST_NONE                         = 0,
+  TEST_MASTER_START                 = 1,
+  TEST_SLAVE_JOIN_REJECT            = 2,
+  TEST_SLAVE_JOIN_ACCEPT            = 3,
+  TEST_SLAVE_ADD                    = 4,
+  TEST_SLAVE_REMOVE                 = 5,
+  TEST_SLAVE_TRANSMIT               = 6,
+  TEST_MASTER_TRANSMIT              = 7,
+  TEST_MASTER_HISTORY_REPLAY_LATEST = 8,
+  TEST_SLAVE_HISTORY_REPLAY_LATEST  = 9,
+  TEST_MASTER_HISTORY_REPLAY       = 10,
+  TEST_SLAVE_HISTORY_REPLAY        = 11,
+  TEST_MASTER_STATE_GET            = 12,
+  TEST_SLAVE_STATE_GET             = 13,
+  TEST_MASTER_STATE_GET_PREFIX     = 14,
+  TEST_SLAVE_STATE_GET_PREFIX      = 15,
+} test;
+
+
+static void
+master_transmit ();
+
+static void
+master_history_replay_latest ();
+
+
+static void
+master_stopped (void *cls)
+{
+  if (NULL != tmit)
+  {
+    GNUNET_PSYC_env_destroy (tmit->env);
+    GNUNET_free (tmit);
+    tmit = NULL;
+  }
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+
+static void
+slave_parted (void *cls)
+{
+  if (NULL != mst)
+  {
+    GNUNET_PSYC_master_stop (mst, GNUNET_NO, &master_stopped, NULL);
+    mst = NULL;
+  }
+  else
+    master_stopped (NULL);
+}
+
+
+/**
+ * Clean up all resources used.
+ */
+static void
+cleanup ()
+{
+  if (NULL != slv)
+  {
+    GNUNET_PSYC_slave_part (slv, GNUNET_NO, &slave_parted, NULL);
+    slv = NULL;
+  }
+  else
+    slave_parted (NULL);
+}
+
+
+/**
+ * Terminate the test case (failure).
+ *
+ * @param cls NULL
+ */
+static void
+end_badly (void *cls)
+{
+  res = 1;
+  cleanup ();
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Test FAILED.\n");
+}
+
+
+/**
+ * Terminate the test case (success).
+ *
+ * @param cls NULL
+ */
+static void
+end_normally (void *cls)
+{
+  res = 0;
+  cleanup ();
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Test PASSED.\n");
+}
+
+
+/**
+ * Finish the test case (successfully).
+ */
+static void
+end ()
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Ending tests.\n");
+
+  if (end_badly_task != NULL)
+  {
+    GNUNET_SCHEDULER_cancel (end_badly_task);
+    end_badly_task = NULL;
+  }
+  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MILLISECONDS,
+                                &end_normally, NULL);
+}
+
+
+static void
+master_message_cb (void *cls, const struct GNUNET_PSYC_MessageHeader *msg)
+{
+  GNUNET_assert (NULL != msg);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Master got PSYC message fragment of size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, ntohs (msg->header.size),
+              GNUNET_ntohll (msg->message_id), ntohl (msg->flags));
+  // FIXME
+}
+
+
+static void
+master_message_part_cb (void *cls, const struct GNUNET_PSYC_MessageHeader *msg,
+                        const struct GNUNET_MessageHeader *pmsg)
+{
+  GNUNET_assert (NULL != msg && NULL != pmsg);
+
+  uint64_t message_id = GNUNET_ntohll (msg->message_id);
+  uint32_t flags = ntohl (msg->flags);
+
+  uint16_t type = ntohs (pmsg->type);
+  uint16_t size = ntohs (pmsg->size);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Master got message part of type %u and size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, type, size, message_id, flags);
+
+  switch (test)
+  {
+  case TEST_SLAVE_TRANSMIT:
+    if (GNUNET_PSYC_MESSAGE_REQUEST != flags)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Test #%d: Unexpected request flags: %x" PRIu32 "\n",
+                  test, flags);
+      GNUNET_assert (0);
+      return;
+    }
+    // FIXME: check rest of message
+
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END == type)
+      master_transmit ();
+    break;
+
+  case TEST_MASTER_TRANSMIT:
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END == type && 2 == ++end_count)
+      master_history_replay_latest ();
+    break;
+
+  case TEST_MASTER_HISTORY_REPLAY:
+  case TEST_MASTER_HISTORY_REPLAY_LATEST:
+    if (GNUNET_PSYC_MESSAGE_HISTORIC != flags)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Test #%d: Unexpected flags for historic message: %x" PRIu32 "\n",
+                  test, flags);
+      GNUNET_assert (0);
+      return;
+    }
+    break;
+
+  default:
+    GNUNET_assert (0);
+  }
+}
+
+
+static void
+slave_message_cb (void *cls, const struct GNUNET_PSYC_MessageHeader *msg)
+{
+  GNUNET_assert (NULL != msg);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Slave got PSYC message fragment of size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, ntohs (msg->header.size),
+              GNUNET_ntohll (msg->message_id), ntohl (msg->flags));
+  // FIXME
+}
+
+
+static void
+slave_message_part_cb (void *cls,
+                       const struct GNUNET_PSYC_MessageHeader *msg,
+                       const struct GNUNET_MessageHeader *pmsg)
+{
+  GNUNET_assert (NULL != msg && NULL != pmsg);
+
+  uint64_t message_id = GNUNET_ntohll (msg->message_id);
+  uint32_t flags = ntohl (msg->flags);
+
+  uint16_t type = ntohs (pmsg->type);
+  uint16_t size = ntohs (pmsg->size);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Slave got message part of type %u and size %u "
+              "belonging to message ID %" PRIu64 " with flags %x\n",
+              test, type, size, message_id, flags);
+
+  switch (test)
+  {
+  case TEST_MASTER_TRANSMIT:
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END == type && 2 == ++end_count)
+      master_history_replay_latest ();
+    break;
+
+  case TEST_SLAVE_HISTORY_REPLAY:
+  case TEST_SLAVE_HISTORY_REPLAY_LATEST:
+    if (GNUNET_PSYC_MESSAGE_HISTORIC != flags)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Test #%d: Unexpected flags for historic message: %x" PRIu32 "\n",
+                  test, flags);
+      GNUNET_assert (0);
+      return;
+    }
+    break;
+
+  default:
+    GNUNET_assert (0);
+  }
+}
+
+
+static void
+state_get_var (void *cls, const struct GNUNET_MessageHeader *mod,
+               const char *name, const void *value,
+               uint32_t value_size, uint32_t full_value_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Got state var: %s\n%.*s\n",
+              name,
+              (int) value_size,
+              (const char *) value);
+}
+
+
+/*** Slave state_get_prefix() ***/
+
+static void
+slave_state_get_prefix_result (void *cls, int64_t result,
+                               const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: slave_state_get_prefix:\t%" PRId64 " (%.*s)\n",
+              test, result,
+              (int) err_msg_size,
+              (const char *) err_msg);
+  // FIXME: GNUNET_assert (2 == result);
+  end ();
+}
+
+
+static void
+slave_state_get_prefix ()
+{
+  test = TEST_SLAVE_STATE_GET_PREFIX;
+  GNUNET_PSYC_channel_state_get_prefix (slv_chn, "_foo", state_get_var,
+                                        slave_state_get_prefix_result, NULL);
+}
+
+
+/*** Master state_get_prefix() ***/
+
+
+static void
+master_state_get_prefix_result (void *cls, int64_t result,
+                                const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: master_state_get_prefix:\t%" PRId64 " (%s)\n",
+              test, result, (char *) err_msg);
+  // FIXME: GNUNET_assert (2 == result);
+  slave_state_get_prefix ();
+}
+
+
+static void
+master_state_get_prefix ()
+{
+  test = TEST_MASTER_STATE_GET_PREFIX;
+  GNUNET_PSYC_channel_state_get_prefix (mst_chn, "_foo", state_get_var,
+                                        master_state_get_prefix_result, NULL);
+}
+
+
+/*** Slave state_get() ***/
+
+
+static void
+slave_state_get_result (void *cls, int64_t result,
+                        const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: slave_state_get:\t%" PRId64 " (%.*s)\n",
+              test, result, err_msg_size, (char *) err_msg);
+  // FIXME: GNUNET_assert (2 == result);
+  master_state_get_prefix ();
+}
+
+
+static void
+slave_state_get ()
+{
+  test = TEST_SLAVE_STATE_GET;
+  GNUNET_PSYC_channel_state_get (slv_chn, "_foo_bar_baz", state_get_var,
+                                 slave_state_get_result, NULL);
+}
+
+
+/*** Master state_get() ***/
+
+
+static void
+master_state_get_result (void *cls, int64_t result,
+                         const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: master_state_get:\t%" PRId64 " (%.*s)\n",
+              test, result, err_msg_size, (char *) err_msg);
+  // FIXME: GNUNET_assert (1 == result);
+  slave_state_get ();
+}
+
+
+static void
+master_state_get ()
+{
+  test = TEST_MASTER_STATE_GET;
+  GNUNET_PSYC_channel_state_get (mst_chn, "_foo_bar_baz", state_get_var,
+                                 master_state_get_result, NULL);
+}
+
+
+/*** Slave history_replay() ***/
+
+static void
+slave_history_replay_result (void *cls, int64_t result,
+                             const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: slave_history_replay:\t%" PRId64 " (%.*s)\n",
+              test, result,
+              (int) err_msg_size,
+              (const char *) err_msg);
+  GNUNET_assert (9 == result);
+
+  master_state_get ();
+}
+
+
+static void
+slave_history_replay ()
+{
+  test = TEST_SLAVE_HISTORY_REPLAY;
+  GNUNET_PSYC_channel_history_replay (slv_chn, 1, 1, "",
+                                      GNUNET_PSYC_HISTORY_REPLAY_LOCAL,
+                                      slave_message_cb,
+                                      slave_message_part_cb,
+                                      slave_history_replay_result, NULL);
+}
+
+
+/*** Master history_replay() ***/
+
+
+static void
+master_history_replay_result (void *cls, int64_t result,
+                              const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: master_history_replay:\t%" PRId64 " (%.*s)\n",
+              test, result,
+              (int) err_msg_size,
+              (const char *) err_msg);
+  GNUNET_assert (9 == result);
+
+  slave_history_replay ();
+}
+
+
+static void
+master_history_replay ()
+{
+  test = TEST_MASTER_HISTORY_REPLAY;
+  GNUNET_PSYC_channel_history_replay (mst_chn, 1, 1, "",
+                                      GNUNET_PSYC_HISTORY_REPLAY_LOCAL,
+                                      master_message_cb,
+                                      master_message_part_cb,
+                                      master_history_replay_result, NULL);
+}
+
+
+/*** Slave history_replay_latest() ***/
+
+
+static void
+slave_history_replay_latest_result (void *cls, int64_t result,
+                                    const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: slave_history_replay_latest:\t%" PRId64 " (%.*s)\n",
+              test, result,
+              (int) err_msg_size,
+              (const char *) err_msg);
+  GNUNET_assert (9 == result);
+
+  master_history_replay ();
+}
+
+
+static void
+slave_history_replay_latest ()
+{
+  test = TEST_SLAVE_HISTORY_REPLAY_LATEST;
+  GNUNET_PSYC_channel_history_replay_latest (slv_chn, 1, "",
+                                             GNUNET_PSYC_HISTORY_REPLAY_LOCAL,
+                                             &slave_message_cb,
+                                             &slave_message_part_cb,
+                                             &slave_history_replay_latest_result,
+                                             NULL);
+}
+
+
+/*** Master history_replay_latest() ***/
+
+
+static void
+master_history_replay_latest_result (void *cls, int64_t result,
+                                     const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: master_history_replay_latest:\t%" PRId64 " (%.*s)\n",
+              test, result, err_msg_size, (char *) err_msg);
+  GNUNET_assert (9 == result);
+
+  slave_history_replay_latest ();
+}
+
+
+static void
+master_history_replay_latest ()
+{
+  test = TEST_MASTER_HISTORY_REPLAY_LATEST;
+  GNUNET_PSYC_channel_history_replay_latest (mst_chn, 1, "",
+                                             GNUNET_PSYC_HISTORY_REPLAY_LOCAL,
+                                             &master_message_cb,
+                                             &master_message_part_cb,
+                                             &master_history_replay_latest_result,
+                                             NULL);
+}
+
+
+static void
+transmit_resume (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Test #%d: Transmission resumed.\n", test);
+  struct TransmitClosure *tmit = cls;
+  if (NULL != tmit->mst_tmit)
+    GNUNET_PSYC_master_transmit_resume (tmit->mst_tmit);
+  else
+    GNUNET_PSYC_slave_transmit_resume (tmit->slv_tmit);
+}
+
+
+static int
+tmit_notify_data (void *cls, uint16_t *data_size, void *data)
+{
+  struct TransmitClosure *tmit = cls;
+  if (0 == tmit->data_count)
+  {
+    *data_size = 0;
+    return GNUNET_YES;
+  }
+
+  uint16_t size = strlen (tmit->data[tmit->n]);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Test #%d: Transmit notify data: %u bytes available, "
+              "processing fragment %u/%u (size %u).\n",
+              test, *data_size, tmit->n + 1, tmit->data_count, size);
+  if (*data_size < size)
+  {
+    *data_size = 0;
+    GNUNET_assert (0);
+    return GNUNET_SYSERR;
+  }
+
+  if (GNUNET_YES != tmit->paused && 0 < tmit->data_delay[tmit->n])
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Test #%d: Transmission paused.\n", test);
+    tmit->paused = GNUNET_YES;
+    GNUNET_SCHEDULER_add_delayed (
+      GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS,
+                                     tmit->data_delay[tmit->n]),
+      &transmit_resume, tmit);
+    *data_size = 0;
+    return GNUNET_NO;
+  }
+  tmit->paused = GNUNET_NO;
+
+  *data_size = size;
+  GNUNET_memcpy (data, tmit->data[tmit->n], size);
+
+  return ++tmit->n < tmit->data_count ? GNUNET_NO : GNUNET_YES;
+}
+
+
+static int
+tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
+                 uint32_t *full_value_size)
+{
+  struct TransmitClosure *tmit = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Test #%d: Transmit notify modifier: %u bytes available, "
+              "%u modifiers left to process.\n",
+              test, *data_size, (unsigned int) GNUNET_PSYC_env_get_count (tmit->env));
+
+  uint16_t name_size = 0;
+  size_t value_size = 0;
+  const char *value = NULL;
+
+  if (NULL != oper && NULL != tmit->mod)
+  { /* New modifier */
+    tmit->mod = tmit->mod->next;
+    if (NULL == tmit->mod)
+    { /* No more modifiers, continue with data */
+      *data_size = 0;
+      return GNUNET_YES;
+    }
+
+    GNUNET_assert (tmit->mod->value_size < UINT32_MAX);
+    *full_value_size = tmit->mod->value_size;
+    *oper = tmit->mod->oper;
+    name_size = strlen (tmit->mod->name);
+
+    if (name_size + 1 + tmit->mod->value_size <= *data_size)
+    {
+      *data_size = name_size + 1 + tmit->mod->value_size;
+    }
+    else
+    {
+      tmit->mod_value_size = tmit->mod->value_size;
+      value_size = *data_size - name_size - 1;
+      tmit->mod_value_size -= value_size;
+      tmit->mod_value = tmit->mod->value + value_size;
+    }
+
+    GNUNET_memcpy (data, tmit->mod->name, name_size);
+    ((char *)data)[name_size] = '\0';
+    GNUNET_memcpy ((char *)data + name_size + 1, tmit->mod->value, value_size);
+  }
+  else if (NULL != tmit->mod_value && 0 < tmit->mod_value_size)
+  { /* Modifier continuation */
+    value = tmit->mod_value;
+    if (tmit->mod_value_size <= *data_size)
+    {
+      value_size = tmit->mod_value_size;
+      tmit->mod_value = NULL;
+    }
+    else
+    {
+      value_size = *data_size;
+      tmit->mod_value += value_size;
+    }
+    tmit->mod_value_size -= value_size;
+
+    if (*data_size < value_size)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "value larger than buffer: %u < %zu\n",
+                  *data_size, value_size);
+      *data_size = 0;
+      return GNUNET_NO;
+    }
+
+    *data_size = value_size;
+    GNUNET_memcpy (data, value, value_size);
+  }
+
+  return GNUNET_NO;
+}
+
+
+static void
+slave_join ();
+
+
+static void
+slave_transmit ()
+{
+  test = TEST_SLAVE_TRANSMIT;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Slave sending request to master.\n", test);
+
+  tmit = GNUNET_new (struct TransmitClosure);
+  tmit->env = GNUNET_PSYC_env_create ();
+  GNUNET_PSYC_env_add (tmit->env, GNUNET_PSYC_OP_ASSIGN,
+                              "_abc", "abc def", 7);
+  GNUNET_PSYC_env_add (tmit->env, GNUNET_PSYC_OP_ASSIGN,
+                              "_abc_def", "abc def ghi", 11);
+  tmit->mod = GNUNET_PSYC_env_head (tmit->env);
+  tmit->n = 0;
+  tmit->data[0] = "slave test";
+  tmit->data_count = 1;
+  tmit->slv_tmit
+    = GNUNET_PSYC_slave_transmit (slv, "_request_test", &tmit_notify_mod,
+                                  &tmit_notify_data, tmit,
+                                  GNUNET_PSYC_SLAVE_TRANSMIT_NONE);
+}
+
+
+static void
+slave_remove_cb (void *cls, int64_t result,
+                 const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: slave_remove:\t%" PRId64 " (%.*s)\n",
+              test, result, err_msg_size, (char *) err_msg);
+
+  slave_transmit ();
+}
+
+
+static void
+slave_remove ()
+{
+  test = TEST_SLAVE_REMOVE;
+  struct GNUNET_PSYC_Channel *chn = GNUNET_PSYC_master_get_channel (mst);
+  GNUNET_PSYC_channel_slave_remove (chn, &slave_pub_key, 2,
+                                    &slave_remove_cb, chn);
+}
+
+
+static void
+slave_add_cb (void *cls, int64_t result,
+              const void *err_msg, uint16_t err_msg_size)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: slave_add:\t%" PRId64 " (%.*s)\n",
+              test, result, err_msg_size, (char *) err_msg);
+  slave_remove ();
+}
+
+
+static void
+slave_add ()
+{
+  test = TEST_SLAVE_ADD;
+  struct GNUNET_PSYC_Channel *chn = GNUNET_PSYC_master_get_channel (mst);
+  GNUNET_PSYC_channel_slave_add (chn, &slave_pub_key, 2, 2, &slave_add_cb, chn);
+}
+
+
+static void
+schedule_second_slave_join (void *cls)
+{
+  slave_join (TEST_SLAVE_JOIN_ACCEPT);
+}
+
+
+static void
+first_slave_parted (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "First slave parted.\n");
+  GNUNET_SCHEDULER_add_now (&schedule_second_slave_join, NULL);
+}
+
+
+static void
+schedule_first_slave_part (void *cls)
+{
+  GNUNET_PSYC_slave_part (slv, GNUNET_NO, &first_slave_parted, NULL);
+}
+
+
+static void
+join_decision_cb (void *cls,
+                  const struct GNUNET_PSYC_JoinDecisionMessage *dcsn,
+                  int is_admitted,
+                  const struct GNUNET_PSYC_Message *join_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Slave got join decision: %d\n", test, is_admitted);
+
+  switch (test)
+  {
+  case TEST_SLAVE_JOIN_REJECT:
+    GNUNET_assert (0 == is_admitted);
+    GNUNET_assert (1 == join_req_count);
+    GNUNET_SCHEDULER_add_now (&schedule_first_slave_part, NULL);
+    break;
+
+  case TEST_SLAVE_JOIN_ACCEPT:
+    GNUNET_assert (1 == is_admitted);
+    GNUNET_assert (2 == join_req_count);
+    slave_add ();
+    break;
+
+  default:
+    GNUNET_break (0);
+  }
+}
+
+
+static void
+join_request_cb (void *cls,
+                 const struct GNUNET_PSYC_JoinRequestMessage *req,
+                 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
+                 const struct GNUNET_PSYC_Message *join_msg,
+                 struct GNUNET_PSYC_JoinHandle *jh)
+{
+  struct GNUNET_HashCode slave_key_hash;
+  GNUNET_CRYPTO_hash (slave_key, sizeof (*slave_key), &slave_key_hash);
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Got join request #%u from %s.\n",
+              test, join_req_count, GNUNET_h2s (&slave_key_hash));
+
+  /* Reject first request */
+  int is_admitted = (0 < join_req_count++) ? GNUNET_YES : GNUNET_NO;
+  GNUNET_PSYC_join_decision (jh, is_admitted, 0, NULL, NULL);
+}
+
+
+static void
+slave_connect_cb (void *cls, int result, uint64_t max_message_id)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Slave connected: %d, max_message_id: %" PRIu64 "\n",
+              test, result, max_message_id);
+  GNUNET_assert (TEST_SLAVE_JOIN_REJECT == test || TEST_SLAVE_JOIN_ACCEPT == test);
+  GNUNET_assert (GNUNET_OK == result || GNUNET_NO == result);
+}
+
+
+static void
+slave_join (int t)
+{
+  test = t;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Joining slave.\n", t);
+
+  struct GNUNET_PeerIdentity origin = this_peer;
+  struct GNUNET_PSYC_Environment *env = GNUNET_PSYC_env_create ();
+  GNUNET_PSYC_env_add (env, GNUNET_PSYC_OP_ASSIGN,
+                              "_foo", "bar baz", 7);
+  GNUNET_PSYC_env_add (env, GNUNET_PSYC_OP_ASSIGN,
+                              "_foo_bar", "foo bar baz", 11);
+  struct GNUNET_PSYC_Message *
+    join_msg = GNUNET_PSYC_message_create ("_request_join", env, "some data", 9);
+
+  slv = GNUNET_PSYC_slave_join (cfg,
+                                &channel_pub_key,
+                                slave_key,
+                                GNUNET_PSYC_SLAVE_JOIN_NONE,
+                                &origin,
+                                0,
+                                NULL,
+                                &slave_message_cb,
+                                &slave_message_part_cb,
+                                &slave_connect_cb,
+                                &join_decision_cb,
+                                NULL,
+                                join_msg);
+  GNUNET_free (join_msg);
+  slv_chn = GNUNET_PSYC_slave_get_channel (slv);
+  GNUNET_PSYC_env_destroy (env);
+}
+
+
+static void
+master_transmit ()
+{
+  test = TEST_MASTER_TRANSMIT;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Master sending message to all.\n", test);
+  end_count = 0;
+
+  uint32_t i, j;
+
+  char *name_max = "_test_max";
+  uint8_t name_max_size = sizeof ("_test_max");
+  char *val_max = GNUNET_malloc (GNUNET_PSYC_MODIFIER_MAX_PAYLOAD);
+  for (i = 0; i < GNUNET_PSYC_MODIFIER_MAX_PAYLOAD; i++)
+    val_max[i] = (0 == i % 10000) ? '0' + i / 10000 : '.';
+
+  char *name_cont = "_test_cont";
+  uint8_t name_cont_size = sizeof ("_test_cont");
+  char *val_cont = GNUNET_malloc (GNUNET_PSYC_MODIFIER_MAX_PAYLOAD
+                                  + GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD);
+  for (i = 0; i < GNUNET_PSYC_MODIFIER_MAX_PAYLOAD - name_cont_size; i++)
+    val_cont[i] = (0 == i % 10000) ? '0' + i / 10000 : ':';
+  for (j = 0; j < GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD; j++, i++)
+    val_cont[i] = (0 == j % 10000) ? '0' + j / 10000 : '!';
+
+  tmit = GNUNET_new (struct TransmitClosure);
+  tmit->env = GNUNET_PSYC_env_create ();
+  GNUNET_PSYC_env_add (tmit->env, GNUNET_PSYC_OP_ASSIGN,
+                              "_foo", "bar baz", 7);
+  GNUNET_PSYC_env_add (tmit->env, GNUNET_PSYC_OP_ASSIGN,
+                              name_max, val_max,
+                              GNUNET_PSYC_MODIFIER_MAX_PAYLOAD
+                              - name_max_size);
+  GNUNET_PSYC_env_add (tmit->env, GNUNET_PSYC_OP_ASSIGN,
+                              "_foo_bar", "foo bar baz", 11);
+  GNUNET_PSYC_env_add (tmit->env, GNUNET_PSYC_OP_ASSIGN,
+                              name_cont, val_cont,
+                              GNUNET_PSYC_MODIFIER_MAX_PAYLOAD - name_cont_size
+                              + GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD);
+  tmit->mod = GNUNET_PSYC_env_head (tmit->env);
+  tmit->data[0] = "foo";
+  tmit->data[1] =  GNUNET_malloc (GNUNET_PSYC_DATA_MAX_PAYLOAD + 1);
+  for (i = 0; i < GNUNET_PSYC_DATA_MAX_PAYLOAD; i++)
+    tmit->data[1][i] = (0 == i % 10000) ? '0' + i / 10000 : '_';
+  tmit->data[2] = "foo bar";
+  tmit->data[3] = "foo bar baz";
+  tmit->data_delay[1] = 3;
+  tmit->data_count = 4;
+  tmit->mst_tmit
+    = GNUNET_PSYC_master_transmit (mst, "_notice_test", &tmit_notify_mod,
+                                   &tmit_notify_data, tmit,
+                                   GNUNET_PSYC_MASTER_TRANSMIT_INC_GROUP_GEN);
+}
+
+
+static void
+master_start_cb (void *cls, int result, uint64_t max_message_id)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Test #%d: Master started: %d, max_message_id: %" PRIu64 "\n",
+              test, result, max_message_id);
+  GNUNET_assert (TEST_MASTER_START == test);
+  GNUNET_assert (GNUNET_OK == result || GNUNET_NO == result);
+  slave_join (TEST_SLAVE_JOIN_REJECT);
+}
+
+
+static void
+master_start ()
+{
+  test = TEST_MASTER_START;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Test #%d: Starting master.\n", test);
+  mst = GNUNET_PSYC_master_start (cfg, channel_key, GNUNET_PSYC_CHANNEL_PRIVATE,
+                                  &master_start_cb, &join_request_cb,
+                                  &master_message_cb, &master_message_part_cb,
+                                  NULL);
+  mst_chn = GNUNET_PSYC_master_get_channel (mst);
+}
+
+
+static void
+schedule_master_start (void *cls)
+{
+  master_start ();
+}
+
+
+/**
+ * Main function of the test, run from scheduler.
+ *
+ * @param cls NULL
+ * @param cfg configuration we use (also to connect to PSYC service)
+ * @param peer handle to access more of the peer (not used)
+ */
+static void
+#if DEBUG_TEST_PSYC
+run (void *cls, char *const *args, const char *cfgfile,
+     const struct GNUNET_CONFIGURATION_Handle *c)
+#else
+run (void *cls,
+     const struct GNUNET_CONFIGURATION_Handle *c,
+     struct GNUNET_TESTING_Peer *peer)
+#endif
+{
+  cfg = c;
+  end_badly_task = GNUNET_SCHEDULER_add_delayed (TIMEOUT, &end_badly, NULL);
+
+  GNUNET_CRYPTO_get_peer_identity (cfg, &this_peer);
+
+  channel_key = GNUNET_CRYPTO_eddsa_key_create ();
+  slave_key = GNUNET_CRYPTO_ecdsa_key_create ();
+
+  GNUNET_CRYPTO_eddsa_key_get_public (channel_key, &channel_pub_key);
+  GNUNET_CRYPTO_ecdsa_key_get_public (slave_key, &slave_pub_key);
+
+#if DEBUG_TEST_PSYC
+  master_start ();
+#else
+  /* Allow some time for the services to initialize. */
+  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
+                                &schedule_master_start, NULL);
+#endif
+}
+
+
+int
+main (int argc, char *argv[])
+{
+  res = 1;
+#if DEBUG_TEST_PSYC
+  const struct GNUNET_GETOPT_CommandLineOption opts[] = {
+    GNUNET_GETOPT_OPTION_END
+  };
+  if (GNUNET_OK != GNUNET_PROGRAM_run (argc, argv, "test-psyc",
+                                       "test-psyc [options]",
+                                       opts, &run, NULL))
+    return 1;
+#else
+  if (0 != GNUNET_TESTING_peer_run ("test-psyc", "test_psyc.conf", &run, NULL))
+    return 1;
+#endif
+  return res;
+}
+
+/* end of test_psyc.c */
diff --git a/src/psyc/test_psyc2.c b/src/psyc/test_psyc2.c
new file mode 100644
index 0000000..c6e7237
--- /dev/null
+++ b/src/psyc/test_psyc2.c
@@ -0,0 +1,284 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/test_psyc2.c
+ * @brief Testbed test for the PSYC API.
+ * @author xrs
+ */
+
+#include "platform.h"
+#include "gnunet_crypto_lib.h"
+#include "gnunet_common.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_testbed_service.h"
+#include "gnunet_psyc_util_lib.h"
+#include "gnunet_psyc_service.h"
+
+#define PEERS_REQUESTED 2
+
+static int result;
+
+static struct GNUNET_SCHEDULER_Task *timeout_tid;
+static struct pctx **pctx;
+
+static struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key;
+static struct GNUNET_CRYPTO_EddsaPublicKey channel_pub_key;
+
+static struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key;
+static struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
+
+/**
+ * Task To perform tests
+ */
+static struct GNUNET_SCHEDULER_Task *test_task;
+
+/**
+ * Peer id couter
+ */
+static unsigned int pids;
+
+struct pctx
+{
+  int idx;
+  struct GNUNET_TESTBED_Peer *peer;
+  const struct GNUNET_PeerIdentity *id;
+
+  struct GNUNET_TESTBED_Operation *op; 
+
+  /**
+   * psyc service handle
+   */
+  void *psyc;
+  struct GNUNET_PSYC_Master *mst;
+  struct GNUNET_PSYC_Slave *slv;
+
+  /**
+   * result for test on peer
+   */
+  int test_ok;
+};
+
+static void
+shutdown_task (void *cls)
+{
+  if (NULL != pctx)
+  {
+    if (NULL != pctx[0]->mst)
+      GNUNET_PSYC_master_stop (pctx[0]->mst, GNUNET_NO, NULL, NULL);  
+
+    for (int i=0; i < PEERS_REQUESTED; i++)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Operation done.\n");
+      GNUNET_TESTBED_operation_done (pctx[i]->op);
+      GNUNET_free_non_null (pctx[i]);
+    }
+    GNUNET_free (pctx);
+  }
+
+  if (NULL != timeout_tid)
+    GNUNET_SCHEDULER_cancel (timeout_tid);
+}
+
+static void
+timeout_task (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Timeout!\n");
+  result = GNUNET_SYSERR;
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+static void 
+start_test (void *cls)
+{
+}
+
+static void
+pinfo_cb (void *cls,
+          struct GNUNET_TESTBED_Operation *operation,
+          const struct GNUNET_TESTBED_PeerInformation *pinfo,
+          const char *emsg)
+{
+  struct pctx *pc = (struct pctx*) cls;
+
+  pc->id = pinfo->result.id;
+
+  pids++;
+  if (pids < (PEERS_REQUESTED - 1))
+    return;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got all IDs, starting test\n");
+  test_task = GNUNET_SCHEDULER_add_now (&start_test, NULL);
+}
+
+static void
+mst_start_cb () 
+{
+}
+
+static void 
+join_request_cb ()
+{
+}
+
+static void
+mst_message_cb ()
+{
+}
+
+static void
+mst_message_part_cb ()
+{
+}
+
+static void 
+slv_message_cb ()
+{
+}
+
+static void 
+slv_message_part_cb ()
+{
+}
+
+static void
+slv_connect_cb () 
+{
+}
+
+static void
+join_decision_cb ()
+{
+}
+
+static void *
+psyc_ca (void *cls,
+         const struct GNUNET_CONFIGURATION_Handle *cfg)
+{
+  struct GNUNET_PSYC_Message *join_msg = NULL;
+  struct pctx *pc = (struct pctx *) cls;
+
+  if (0 == pc->idx)
+  {
+    pc->mst = GNUNET_PSYC_master_start (cfg, channel_key, 
+                                        GNUNET_PSYC_CHANNEL_PRIVATE,
+                                        &mst_start_cb, &join_request_cb,
+                                        &mst_message_cb, &mst_message_part_cb,
+                                        NULL);
+    return pc->mst;
+  }
+
+  pc->slv = GNUNET_PSYC_slave_join (cfg, &channel_pub_key, slave_key,
+                                    GNUNET_PSYC_SLAVE_JOIN_NONE,
+                                    &pid, 0, NULL, &slv_message_cb, 
+                                    &slv_message_part_cb,
+                                    &slv_connect_cb, &join_decision_cb, 
+                                    NULL, join_msg);
+  return pc->slv;
+} 
+
+static void
+psyc_da (void *cls,
+         void *op_result)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Disconnected from service.\n");
+} 
+
+static void
+service_connect (void *cls,
+                 struct GNUNET_TESTBED_Operation *op,
+                 void *ca_result,
+                 const char *emsg)
+{
+  struct pctx *pc = (struct pctx *) cls;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "Connected to service\n");
+
+  GNUNET_assert (NULL != ca_result);
+
+  // FIXME: we need a simple service handle to connect to the service, then 
+  // get peer information and AFTER that make PSYC ops. Compare to CADET. 
+  pc->psyc = ca_result;
+
+  GNUNET_TESTBED_peer_get_information (pc->peer, 
+                                       GNUNET_TESTBED_PIT_IDENTITY, 
+                                       pinfo_cb, pc);
+}
+
+static void
+testbed_master (void *cls,
+     struct GNUNET_TESTBED_RunHandle *h,
+     unsigned int num_peers,
+     struct GNUNET_TESTBED_Peer **p,
+     unsigned int links_succeeded,
+     unsigned int links_failed)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Connected to testbed_master()\n");
+
+  // Create ctx for peers
+  pctx = GNUNET_new_array (PEERS_REQUESTED, struct pctx*);
+  for (int i = 0; i<PEERS_REQUESTED; i++) 
+  {
+    pctx[i] = GNUNET_new (struct pctx);
+    pctx[i]->idx = i;
+    pctx[i]->peer = p[i];
+    pctx[i]->id = NULL;
+    pctx[i]->mst = NULL;
+    pctx[i]->op = NULL;
+    pctx[i]->test_ok = GNUNET_NO;
+  }
+
+  channel_key = GNUNET_CRYPTO_eddsa_key_create ();
+  slave_key = GNUNET_CRYPTO_ecdsa_key_create ();
+
+  GNUNET_CRYPTO_eddsa_key_get_public (channel_key, &channel_pub_key);
+  GNUNET_CRYPTO_ecdsa_key_get_public (slave_key, &slave_pub_key);
+
+  pctx[0]->op = 
+    GNUNET_TESTBED_service_connect (NULL, p[0], "psyc", service_connect, 
+                                    pctx[0], psyc_ca, psyc_da, pctx[0]);
+
+  GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL); 
+
+  timeout_tid = 
+    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 5),
+                                  &timeout_task, NULL);
+}
+
+int
+main (int argc, char *argv[])
+{
+  int ret;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "test\n");
+
+  result = GNUNET_SYSERR;
+
+  ret = GNUNET_TESTBED_test_run ("test-psyc2", "test_psyc.conf",
+                                 PEERS_REQUESTED, 0LL, NULL, NULL, 
+                                 testbed_master, NULL);
+
+  if ((GNUNET_OK != ret) || (GNUNET_OK != result))
+    return 1;
+
+  return 0;
+}
+
+/* end of test-psyc2.c */
diff --git a/src/psyc/test_psyc_api_join.c b/src/psyc/test_psyc_api_join.c
new file mode 100644
index 0000000..419fa11
--- /dev/null
+++ b/src/psyc/test_psyc_api_join.c
@@ -0,0 +1,282 @@
+/*
+ * This file is part of GNUnet
+ * Copyright (C) 2013 GNUnet e.V.
+ *
+ * GNUnet is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License,
+ * 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
+ * Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+     SPDX-License-Identifier: AGPL3.0-or-later
+ */
+
+/**
+ * @file psyc/test_psyc_api_join.c
+ * @brief Testbed test for the PSYC API.
+ * @author xrs
+ */
+
+/**
+ * Lessons Learned:
+ * - define topology in config
+ * - psyc slave join needs part to end (same with master)
+ * - GNUNET_SCHEDULER_add_delayed return value will outdate at call time
+ * - main can not contain GNUNET_log()
+ */
+
+#include "platform.h"
+#include "gnunet_crypto_lib.h"
+#include "gnunet_common.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_testbed_service.h"
+#include "gnunet_psyc_util_lib.h"
+#include "gnunet_psyc_service.h"
+#include "psyc_test_lib.h"
+
+static struct pctx PEERS[2];
+
+static int pids;
+
+
+static void 
+shutdown_task (void *cls)
+{
+  if (NULL != timeout_task_id) {
+    GNUNET_SCHEDULER_cancel (timeout_task_id);
+    timeout_task_id = NULL;
+  }
+
+  for (int i=0;i<2;i++) {
+    GNUNET_free (PEERS[i].channel_pub_key);
+
+    if (NULL != PEERS[i].psyc)
+    {
+      if (0 == i) 
+        GNUNET_PSYC_master_stop (PEERS[i].psyc, GNUNET_NO, NULL, NULL);
+      else 
+        GNUNET_PSYC_slave_part (PEERS[i].psyc, GNUNET_NO, NULL, NULL);
+    }
+  }
+
+  for (int i=0;i<MAX_TESTBED_OPS;i++)
+    if (NULL != op[i])
+      GNUNET_TESTBED_operation_done (op[i]);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Shut down!\n");
+}
+
+static void
+timeout_task (void *cls)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Timeout!\n");
+
+  timeout_task_id = NULL;
+
+  result = GNUNET_SYSERR;
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+static void
+join_decision_cb (void *cls,
+                  const struct GNUNET_PSYC_JoinDecisionMessage *dcsn,
+                  int is_admitted,
+                  const struct GNUNET_PSYC_Message *join_msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+              "slave: got join decision: %s\n", 
+              (GNUNET_YES == is_admitted) ? "admitted":"rejected");
+
+  result = (GNUNET_YES == is_admitted) ? GNUNET_OK : GNUNET_SYSERR;
+
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+static void
+join_request_cb (void *cls,
+                 const struct GNUNET_PSYC_JoinRequestMessage *req,
+                 const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
+                 const struct GNUNET_PSYC_Message *join_msg,
+                 struct GNUNET_PSYC_JoinHandle *jh)
+{
+  struct GNUNET_HashCode slave_key_hash;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "master: got join request.\n");
+
+  GNUNET_CRYPTO_hash (slave_key, sizeof (*slave_key), &slave_key_hash);
+
+  GNUNET_PSYC_join_decision (jh, GNUNET_YES, 0, NULL, NULL);
+}
+
+static void
+psyc_da ()
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "disconnect form PSYC service\n");
+}
+
+static void *
+psyc_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg) 
+{
+  struct pctx *peer = (struct pctx*) cls;
+
+  // Case: master role
+  if (0 == peer->idx) {
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Connecting to PSYC as master ...\n");
+
+    peer->psyc = (struct GNUNET_PSYC_Master *) 
+      GNUNET_PSYC_master_start (cfg,
+                                peer->channel_key,
+                                GNUNET_PSYC_CHANNEL_PRIVATE,
+                                NULL,
+                                join_request_cb,
+                                NULL,
+                                NULL,
+                                cls);
+    return peer->psyc;
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Connecting to PSYC as slave ...\n");
+
+  struct GNUNET_PSYC_Environment *env = GNUNET_PSYC_env_create ();
+  GNUNET_PSYC_env_add (env, GNUNET_PSYC_OP_ASSIGN, "_foo", "bar baz", 7);
+  GNUNET_PSYC_env_add (env, GNUNET_PSYC_OP_ASSIGN, "_foo_bar", "foo bar baz", 11);
+
+  struct GNUNET_PSYC_Message *
+    join_msg = GNUNET_PSYC_message_create ("_request_join", env, "some data", 40);
+
+  peer->psyc = (struct GNUNET_PSYC_Slave *)
+    GNUNET_PSYC_slave_join (cfg,
+                            peer->channel_pub_key,
+                            peer->id_key,
+                            GNUNET_PSYC_SLAVE_JOIN_NONE,
+                            peer->peer_id_master, 
+                            0,
+                            NULL,
+                            NULL,
+                            NULL,
+                            NULL,
+                            join_decision_cb,
+                            cls,
+                            join_msg);
+
+  GNUNET_free (join_msg);
+  peer->channel = GNUNET_PSYC_slave_get_channel (peer->psyc);
+  GNUNET_PSYC_env_destroy (env);
+
+  return peer->psyc;
+}
+
+static void
+service_connect (void *cls,
+                 struct GNUNET_TESTBED_Operation *op,
+                 void *ca_result,
+                 const char *emsg)
+{
+  GNUNET_assert (NULL != ca_result);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Connected to the service\n");
+}
+
+static void 
+connect_to_services (void *cls)
+{
+  for (int i = 0; i < 2; i++)
+  {
+    PEERS[i].peer_id_master = PEERS[0].peer_id;
+
+    op[op_cnt++] = 
+      GNUNET_TESTBED_service_connect (NULL, PEERS[i].testbed_peer, "psyc", 
+                                      &service_connect, &PEERS[i], &psyc_ca, 
+                                      &psyc_da, &PEERS[i]);
+  }
+}
+
+static void
+pinfo_cb (void *cls,
+          struct GNUNET_TESTBED_Operation *operation,
+          const struct GNUNET_TESTBED_PeerInformation *pinfo,
+          const char *emsg)
+{
+  struct pctx *peer = (struct pctx*) cls;
+
+  peer->peer_id = pinfo->result.id;
+
+  pids++;
+  if (pids < 2)
+    return;
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Got all IDs, starting test\n");
+
+  GNUNET_SCHEDULER_add_now (&connect_to_services, NULL);
+}
+
+static void
+testbed_master (void *cls,
+     struct GNUNET_TESTBED_RunHandle *h,
+     unsigned int num_peers,
+     struct GNUNET_TESTBED_Peer **p,
+     unsigned int links_succeeded,
+     unsigned int links_failed)
+{
+  struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key = NULL;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "Connected to testbed_master\n");
+
+  // Set up shutdown logic
+  GNUNET_SCHEDULER_add_shutdown (&shutdown_task, NULL); 
+  timeout_task_id = 
+    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, 15),
+                                  &timeout_task, NULL);
+  GNUNET_assert (NULL != timeout_task_id);
+
+  // Set up channel key
+  channel_key = GNUNET_CRYPTO_eddsa_key_create ();
+  GNUNET_assert (NULL != channel_key);
+  
+  // Set up information contexts for peers
+  for (int i=0 ; i < 2 ; i++)
+  {
+    PEERS[i].idx = i;
+    PEERS[i].testbed_peer = p[i];
+
+    // Create "egos"
+    PEERS[i].id_key = GNUNET_CRYPTO_ecdsa_key_create ();
+
+    // Set up channel keys shared by master and slave
+    PEERS[i].channel_key = channel_key; 
+
+    PEERS[i].channel_pub_key = 
+      GNUNET_malloc (sizeof (struct GNUNET_CRYPTO_EddsaPublicKey));
+    // Get public key 
+    GNUNET_CRYPTO_eddsa_key_get_public (PEERS[i].channel_key, 
+                                        PEERS[i].channel_pub_key);
+    // Get peerinfo
+    op[op_cnt++] = 
+      GNUNET_TESTBED_peer_get_information (p[i], 
+                                           GNUNET_TESTBED_PIT_IDENTITY, 
+                                           pinfo_cb, &PEERS[i]);
+  }
+}
+
+int 
+main (int argc, char *argv[])
+{
+  int ret; 
+
+  ret = GNUNET_TESTBED_test_run ("test_psyc_api_join", "test_psyc.conf",
+                                 2, 0LL, NULL, NULL, 
+                                 &testbed_master, NULL);
+
+  if ( (GNUNET_OK != ret) || (GNUNET_OK != result) )
+    return 1;
+
+  return 0;
+}
+
+/* end of test_psyc_api_join.c */