PSYC: implement slave to master requests, tests, fixes, reorg

Multicast lib: handle member to origin requests.
Keep track of members and origins and call their callbacks when necessary.

6 files changed, 729 insertions, 330 deletions

diff --git a/src/psyc/Makefile.am b/src/psyc/Makefile.am
index 7860b3995..162b42b2b 100644
--- a/src/psyc/Makefile.am
+++ b/src/psyc/Makefile.am
@@ -21,7 +21,7 @@ lib_LTLIBRARIES = libgnunetpsyc.la
 
 libgnunetpsyc_la_SOURCES = \
   psyc_api.c \
-  psyc.h
+  psyc_common.c
 libgnunetpsyc_la_LIBADD = \
   $(top_builddir)/src/util/libgnunetutil.la \
   $(top_builddir)/src/env/libgnunetenv.la \
@@ -39,7 +39,8 @@ libexec_PROGRAMS = \
  gnunet-service-psyc
 
 gnunet_service_psyc_SOURCES = \
- gnunet-service-psyc.c
+ gnunet-service-psyc.c \
+ psyc_common.c
 gnunet_service_psyc_LDADD = \
   $(top_builddir)/src/statistics/libgnunetstatistics.la \
   $(top_builddir)/src/util/libgnunetutil.la \
@@ -51,6 +52,7 @@ gnunet_service_psyc_DEPENDENCIES = \
   $(top_builddir)/src/util/libgnunetutil.la \
   $(top_builddir)/src/multicast/libgnunetmulticast.la \
   $(top_builddir)/src/psycstore/libgnunetpsycstore.la
+gnunet_service_psyc_CFLAGS = $(AM_CFLAGS)
 
 
 if HAVE_TESTING
diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index e5de7dcda..dcb5031f1 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -56,6 +56,7 @@ static struct GNUNET_SERVER_NotificationContext *nc;
 static struct GNUNET_PSYCSTORE_Handle *store;
 
 /**
+ * All connected masters and slaves.
  * Channel's pub_key_hash -> struct Channel
  */
 static struct GNUNET_CONTAINER_MultiHashMap *clients;
@@ -105,6 +106,15 @@ struct Channel
 
   uint8_t in_transmit;
   uint8_t is_master;
+
+  /**
+   * Ready to receive messages from client.
+   */
+  uint8_t ready;
+
+  /**
+   * Client disconnected.
+   */
   uint8_t disconnected;
 };
 
@@ -116,7 +126,6 @@ struct Master
   struct Channel channel;
   struct GNUNET_CRYPTO_EddsaPrivateKey priv_key;
   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
-  struct GNUNET_HashCode pub_key_hash;
 
   struct GNUNET_MULTICAST_Origin *origin;
   struct GNUNET_MULTICAST_OriginMessageHandle *tmit_handle;
@@ -144,6 +153,8 @@ struct Master
    * @see enum GNUNET_PSYC_Policy
    */
   uint32_t policy;
+
+  struct GNUNET_HashCode pub_key_hash;
 };
 
 
@@ -155,24 +166,26 @@ struct Slave
   struct Channel channel;
   struct GNUNET_CRYPTO_EddsaPrivateKey slave_key;
   struct GNUNET_CRYPTO_EddsaPublicKey chan_key;
-  struct GNUNET_HashCode chan_key_hash;
 
   struct GNUNET_MULTICAST_Member *member;
   struct GNUNET_MULTICAST_MemberRequestHandle *tmit_handle;
 
   struct GNUNET_PeerIdentity origin;
+
+  uint32_t relay_count;
   struct GNUNET_PeerIdentity *relays;
+
   struct GNUNET_MessageHeader *join_req;
 
   uint64_t max_message_id;
   uint64_t max_request_id;
 
-  uint32_t relay_count;
+  struct GNUNET_HashCode chan_key_hash;
 };
 
 
 static inline void
-transmit_message (struct Channel *ch);
+transmit_message (struct Channel *ch, uint8_t inc_msg_id);
 
 
 /**
@@ -235,14 +248,14 @@ client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
   if (NULL == client)
     return;
 
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p disconnected\n", client);
-
   struct Channel *ch
     = GNUNET_SERVER_client_get_user_context (client, struct Channel);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p Client disconnected\n", ch);
+
   if (NULL == ch)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "User context is NULL in client_disconnect()\n");
+                "%p User context is NULL in client_disconnect()\n", ch);
     GNUNET_break (0);
     return;
   }
@@ -252,7 +265,7 @@ client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
   /* Send pending messages to multicast before cleanup. */
   if (NULL != ch->tmit_head)
   {
-    transmit_message (ch);
+    transmit_message (ch, GNUNET_NO);
   }
   else
   {
@@ -311,53 +324,22 @@ fragment_store_result (void *cls, int64_t result, const char *err_msg)
 
 
 /**
- * Iterator callback for sending a message to a client.
- *
- * @see message_cb()
- */
-static int
-message_to_client (void *cls, const struct GNUNET_HashCode *chan_key_hash,
-                   void *chan)
-{
-  const struct GNUNET_MessageHeader *msg = cls;
-  struct Channel *ch = chan;
-
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Sending message of type %u and size %u to client 0x%zx.\n",
-              ntohs (msg->type), ntohs (msg->size), ch->client);
-
-  GNUNET_SERVER_notification_context_add (nc, ch->client);
-  GNUNET_SERVER_notification_context_unicast (nc, ch->client, msg, GNUNET_NO);
-
-  return GNUNET_YES;
-}
-
-
-/**
  * Incoming message fragment from multicast.
  *
- * Store it using PSYCstore and send it to all clients of the channel.
+ * Store it using PSYCstore and send it to the client of the channel.
  */
 static void
-message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
+message_cb (struct Channel *ch,
+            const struct GNUNET_CRYPTO_EddsaPublicKey *chan_key,
+            const struct GNUNET_HashCode *chan_key_hash,
+            const struct GNUNET_MessageHeader *msg)
 {
   uint16_t type = ntohs (msg->type);
   uint16_t size = ntohs (msg->size);
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Received message of type %u and size %u from multicast.\n",
-              type, size);
-
-  struct Channel *ch = cls;
-  struct Master *mst = cls;
-  struct Slave *slv = cls;
-
-  /* const struct GNUNET_MULTICAST_MessageHeader *mmsg
-     = (const struct GNUNET_MULTICAST_MessageHeader *) msg; */
-  struct GNUNET_CRYPTO_EddsaPublicKey *chan_key
-    = ch->is_master ? &mst->pub_key : &slv->chan_key;
-  struct GNUNET_HashCode *chan_key_hash
-    = ch->is_master ? &mst->pub_key_hash : &slv->chan_key_hash;
+              "%p Received message of type %u and size %u from multicast.\n",
+              ch, type, size);
 
   switch (type)
   {
@@ -378,29 +360,19 @@ message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 
     const struct GNUNET_MULTICAST_MessageHeader *mmsg
       = (const struct GNUNET_MULTICAST_MessageHeader *) msg;
-    struct GNUNET_PSYC_MessageHeader *pmsg;
-
-    uint16_t size = ntohs (msg->size);
-    uint16_t psize = 0;
-    uint16_t pos = 0;
 
-    for (pos = 0; sizeof (*mmsg) + pos < size; pos += psize)
+    if (GNUNET_YES != GNUNET_PSYC_check_message_parts (size - sizeof (*mmsg),
+                                                       (const char *) &mmsg[1]))
     {
-      const struct GNUNET_MessageHeader *pmsg
-        = (const struct GNUNET_MessageHeader *) ((char *) &mmsg[1] + pos);
-      psize = ntohs (pmsg->size);
-      if (psize < sizeof (*pmsg) || sizeof (*mmsg) + pos + psize > size)
-      {
-        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                    "Received invalid message part of type %u and size %u "
-                    "from multicast. Not sending to clients.\n",
-                    ntohs (pmsg->type), psize);
-        GNUNET_break_op (0);
-        return;
-      }
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "%p Received message with invalid parts from multicast. "
+                  "Dropping message.\n", ch);
+      GNUNET_break_op (0);
+      break;
     }
 
-    psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+    struct GNUNET_PSYC_MessageHeader *pmsg;
+    uint16_t psize = sizeof (*pmsg) + size - sizeof (*mmsg);
     pmsg = GNUNET_malloc (psize);
     pmsg->header.size = htons (psize);
     pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
@@ -408,39 +380,116 @@ message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 
     memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
 
-    GNUNET_CONTAINER_multihashmap_get_multiple (clients, chan_key_hash,
-                                                message_to_client,
-                                                (void *) pmsg);
+    GNUNET_SERVER_notification_context_add (nc, ch->client);
+    GNUNET_SERVER_notification_context_unicast (nc, ch->client,
+                                                (const struct GNUNET_MessageHeader *) pmsg,
+                                                GNUNET_NO);
     GNUNET_free (pmsg);
     break;
   }
   default:
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                "Discarding unknown message of type %u and size %u.\n",
-                type, size);
+                "%p Dropping unknown message of type %u and size %u.\n",
+                ch, type, size);
   }
 }
 
 
 /**
- * Send a request received from multicast to a client.
+ * Incoming message fragment from multicast for a master.
  */
-static int
-request_to_client (void *cls, const struct GNUNET_HashCode *chan_key_hash,
-                   void *chan)
+static void
+master_message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
+{
+  struct Master *mst = cls;
+  GNUNET_assert (NULL != mst);
+
+  struct GNUNET_CRYPTO_EddsaPublicKey *chan_key = &mst->pub_key;
+  struct GNUNET_HashCode *chan_key_hash = &mst->pub_key_hash;
+
+  message_cb (&mst->channel, chan_key, chan_key_hash, msg);
+}
+
+
+/**
+ * Incoming message fragment from multicast for a slave.
+ */
+static void
+slave_message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 {
-  /* TODO */
+  struct Slave *slv = cls;
+  GNUNET_assert (NULL != slv);
+
+  struct GNUNET_CRYPTO_EddsaPublicKey *chan_key = &slv->chan_key;
+  struct GNUNET_HashCode *chan_key_hash = &slv->chan_key_hash;
 
-  return GNUNET_YES;
+  message_cb (&slv->channel, chan_key, chan_key_hash, msg);
 }
 
 
+/**
+ * Incoming request fragment from multicast for a master.
+ *
+ * @param cls           Master.
+ * @param member_key    Sending member's public key.
+ * @param msg           The message.
+ * @param flags         Request flags.
+ */
 static void
 request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
-            const struct GNUNET_MessageHeader *req,
+            const struct GNUNET_MessageHeader *msg,
             enum GNUNET_MULTICAST_MessageFlags flags)
 {
+  struct Master *mst = cls;
+  struct Channel *ch = &mst->channel;
 
+  uint16_t type = ntohs (msg->type);
+  uint16_t size = ntohs (msg->size);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received request of type %u and size %u from multicast.\n",
+              ch, type, size);
+
+  switch (type)
+  {
+  case GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST:
+  {
+    const struct GNUNET_MULTICAST_RequestHeader *req
+      = (const struct GNUNET_MULTICAST_RequestHeader *) msg;
+
+    if (GNUNET_YES != GNUNET_PSYC_check_message_parts (size - sizeof (*req),
+                                                       (const char *) &req[1]))
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "%p Dropping message with invalid parts "
+                  "received from multicast.\n", ch);
+      GNUNET_break_op (0);
+      break;
+    }
+
+    struct GNUNET_PSYC_MessageHeader *pmsg;
+    uint16_t psize = sizeof (*pmsg) + size - sizeof (*req);
+    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->flags = htonl (GNUNET_PSYC_MESSAGE_REQUEST);
+
+    memcpy (&pmsg[1], &req[1], size - sizeof (*req));
+
+    GNUNET_SERVER_notification_context_add (nc, ch->client);
+    GNUNET_SERVER_notification_context_unicast (nc, ch->client,
+                                                (const struct GNUNET_MessageHeader *) pmsg,
+                                                GNUNET_NO);
+    GNUNET_free (pmsg);
+    break;
+  }
+  default:
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Dropping unknown request of type %u and size %u.\n",
+                ch, type, size);
+    GNUNET_break_op (0);
+  }
 }
 
 
@@ -470,7 +519,8 @@ master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
                                        max_fragment_id + 1,
                                        join_cb, membership_test_cb,
                                        replay_fragment_cb, replay_message_cb,
-                                       request_cb, message_cb, ch);
+                                       request_cb, master_message_cb, ch);
+    ch->ready = GNUNET_YES;
   }
   GNUNET_SERVER_notification_context_add (nc, ch->client);
   GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
@@ -505,7 +555,8 @@ slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
                                       slv->join_req, join_cb,
                                       membership_test_cb,
                                       replay_fragment_cb, replay_message_cb,
-                                      message_cb, ch);
+                                      slave_message_cb, ch);
+    ch->ready = GNUNET_YES;
   }
 
   GNUNET_SERVER_notification_context_add (nc, ch->client);
@@ -529,9 +580,11 @@ handle_master_start (void *cls, struct GNUNET_SERVER_Client *client,
   mst->channel.is_master = GNUNET_YES;
   mst->policy = ntohl (req->policy);
   mst->priv_key = req->channel_key;
-  GNUNET_CRYPTO_eddsa_key_get_public (&mst->priv_key,
-                                                  &mst->pub_key);
+  GNUNET_CRYPTO_eddsa_key_get_public (&mst->priv_key, &mst->pub_key);
   GNUNET_CRYPTO_hash (&mst->pub_key, sizeof (mst->pub_key), &mst->pub_key_hash);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Master connected to channel %s.\n",
+              mst, GNUNET_h2s (&mst->pub_key_hash));
 
   GNUNET_PSYCSTORE_counters_get (store, &mst->pub_key,
                                  master_counters_cb, mst);
@@ -561,14 +614,20 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
                       &slv->chan_key_hash);
   slv->origin = req->origin;
   slv->relay_count = ntohl (req->relay_count);
+  if (0 < slv->relay_count)
+  {
+    const struct GNUNET_PeerIdentity *relays
+      = (const struct GNUNET_PeerIdentity *) &req[1];
+    slv->relays
+      = GNUNET_malloc (slv->relay_count * sizeof (struct GNUNET_PeerIdentity));
+    uint32_t i;
+    for (i = 0; i < slv->relay_count; i++)
+      memcpy (&slv->relays[i], &relays[i], sizeof (*relays));
+  }
 
-  const struct GNUNET_PeerIdentity *relays
-    = (const struct GNUNET_PeerIdentity *) &req[1];
-  slv->relays
-    = GNUNET_malloc (slv->relay_count * sizeof (struct GNUNET_PeerIdentity));
-  uint32_t i;
-  for (i = 0; i < slv->relay_count; i++)
-    memcpy (&slv->relays[i], &relays[i], sizeof (*relays));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Slave connected to channel %s.\n",
+              slv, GNUNET_h2s (&slv->chan_key_hash));
 
   GNUNET_PSYCSTORE_counters_get (store, &slv->chan_key,
                                  slave_counters_cb, slv);
@@ -609,13 +668,14 @@ transmit_notify (void *cls, size_t *data_size, void *data)
 
   if (NULL == tmit_msg || *data_size < tmit_msg->size)
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "transmit_notify: nothing to send.\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p transmit_notify: nothing to send.\n", ch);
     *data_size = 0;
     return GNUNET_NO;
   }
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "transmit_notify: sending %u bytes.\n", tmit_msg->size);
+              "%p transmit_notify: sending %u bytes.\n", ch, tmit_msg->size);
 
   *data_size = tmit_msg->size;
   memcpy (data, tmit_msg->buf, *data_size);
@@ -630,7 +690,7 @@ transmit_notify (void *cls, size_t *data_size, void *data)
   {
     if (NULL != ch->tmit_head)
     {
-      transmit_message (ch);
+      transmit_message (ch, GNUNET_NO);
     }
     else if (ch->disconnected)
     {
@@ -644,19 +704,55 @@ transmit_notify (void *cls, size_t *data_size, void *data)
 
 
 /**
+ * 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)
+master_transmit_message (struct Master *mst, uint8_t inc_msg_id)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "master_transmit_message()\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p master_transmit_message()\n", mst);
   mst->channel.tmit_task = 0;
   if (NULL == mst->tmit_handle)
   {
+    if (GNUNET_YES == inc_msg_id)
+      mst->max_message_id++;
     mst->tmit_handle
-      = GNUNET_MULTICAST_origin_to_all (mst->origin, ++mst->max_message_id,
+      = GNUNET_MULTICAST_origin_to_all (mst->origin, mst->max_message_id,
                                         mst->max_group_generation,
-                                        transmit_notify, mst);
+                                        master_transmit_notify, mst);
   }
   else
   {
@@ -669,14 +765,16 @@ master_transmit_message (struct Master *mst)
  * Transmit a message from a channel slave to the multicast group.
  */
 static void
-slave_transmit_message (struct Slave *slv)
+slave_transmit_message (struct Slave *slv, uint8_t inc_msg_id)
 {
   slv->channel.tmit_task = 0;
   if (NULL == slv->tmit_handle)
   {
+    if (GNUNET_YES == inc_msg_id)
+      slv->max_message_id++;
     slv->tmit_handle
-      = GNUNET_MULTICAST_member_to_origin(slv->member, ++slv->max_request_id,
-                                          transmit_notify, slv);
+      = GNUNET_MULTICAST_member_to_origin (slv->member, slv->max_request_id,
+                                           slave_transmit_notify, slv);
   }
   else
   {
@@ -686,11 +784,11 @@ slave_transmit_message (struct Slave *slv)
 
 
 static inline void
-transmit_message (struct Channel *ch)
+transmit_message (struct Channel *ch, uint8_t inc_msg_id)
 {
   ch->is_master
-    ? master_transmit_message ((struct Master *) ch)
-    : slave_transmit_message ((struct Slave *) ch);
+    ? master_transmit_message ((struct Master *) ch, inc_msg_id)
+    : slave_transmit_message ((struct Slave *) ch, inc_msg_id);
 }
 
 
@@ -708,10 +806,9 @@ transmit_error (struct Channel *ch)
   tmit_msg->size = sizeof (*msg);
   tmit_msg->state = ch->tmit_state;
   GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, tmit_msg);
-  transmit_message (ch);
+  transmit_message (ch, GNUNET_NO);
 
   /* FIXME: cleanup */
-  GNUNET_SERVER_client_disconnect (ch->client);
 }
 
 
@@ -720,40 +817,60 @@ transmit_error (struct Channel *ch)
  */
 static void
 handle_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
-                         const struct GNUNET_MessageHeader *msg)
+                     const struct GNUNET_MessageHeader *msg)
 {
   struct Channel *ch
     = GNUNET_SERVER_client_get_user_context (client, struct Channel);
   GNUNET_assert (NULL != ch);
 
+  if (GNUNET_YES != ch->ready)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p Ignoring message from client, channel is not ready yet.\n",
+                ch);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
+    return;
+  }
+
+  uint8_t inc_msg_id = GNUNET_NO;
   uint16_t size = ntohs (msg->size);
-  uint16_t psize = 0, pos = 0;
+  uint16_t psize = 0, ptype = 0, pos = 0;
 
   if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < size - sizeof (*msg))
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Message payload too large\n");
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "%p Message payload too large\n", ch);
     GNUNET_break (0);
     transmit_error (ch);
+    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
     return;
   }
 
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Received message from client.\n", ch);
+  GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, msg);
+
   for (pos = 0; sizeof (*msg) + pos < size; pos += psize)
   {
     const struct GNUNET_MessageHeader *pmsg
       = (const struct GNUNET_MessageHeader *) ((char *) &msg[1] + pos);
     psize = ntohs (pmsg->size);
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Received message part of type %u and size %u "
-                "from client.\n", ntohs (pmsg->type), psize);
+    ptype = ntohs (pmsg->type);
     if (psize < sizeof (*pmsg) || sizeof (*msg) + pos + psize > size)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                  "Received invalid message part of type %u and size %u "
-                  "from client.\n", ntohs (pmsg->type), psize);
+                  "%p Received invalid message part of type %u and size %u "
+                  "from client.\n", ch, ptype, psize);
       GNUNET_break (0);
       transmit_error (ch);
+      GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
       return;
     }
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "%p Received message part from client.\n", ch);
+    GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, pmsg);
+
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == ptype)
+      inc_msg_id = GNUNET_YES;
   }
 
   size -= sizeof (*msg);
@@ -763,7 +880,7 @@ handle_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
   tmit_msg->size = size;
   tmit_msg->state = ch->tmit_state;
   GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, tmit_msg);
-  transmit_message (ch);
+  transmit_message (ch, inc_msg_id);
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 };
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index 90c07480a..1ffda5d08 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -27,7 +27,16 @@
 #ifndef PSYC_H
 #define PSYC_H
 
-#include "gnunet_common.h"
+#include "platform.h"
+#include "gnunet_psyc_service.h"
+
+
+int
+GNUNET_PSYC_check_message_parts (uint16_t data_size, const char *data);
+
+void
+GNUNET_PSYC_log_message (enum GNUNET_ErrorType kind,
+                         const struct GNUNET_MessageHeader *msg);
 
 
 enum MessageState
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index 20394bbce..8a1c9ffaa 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -48,12 +48,30 @@ struct OperationHandle
   struct GNUNET_MessageHeader *msg;
 };
 
+
+/**
+ * Handle for a pending PSYC transmission operation.
+ */
+struct GNUNET_PSYC_ChannelTransmitHandle
+{
+  struct GNUNET_PSYC_Channel *ch;
+  GNUNET_PSYC_TransmitNotifyModifier notify_mod;
+  GNUNET_PSYC_TransmitNotifyData notify_data;
+  void *notify_cls;
+  enum MessageState state;
+};
+
 /**
  * Handle to access PSYC channel operations for both the master and slaves.
  */
 struct GNUNET_PSYC_Channel
 {
   /**
+   * Transmission handle;
+   */
+  struct GNUNET_PSYC_ChannelTransmitHandle tmit;
+
+  /**
    * Configuration to use.
    */
   const struct GNUNET_CONFIGURATION_Handle *cfg;
@@ -124,6 +142,11 @@ struct GNUNET_PSYC_Channel
   uint64_t recv_message_id;
 
   /**
+   * Public key of the slave from which a message is being received.
+   */
+  struct GNUNET_CRYPTO_EddsaPublicKey recv_slave_key;
+
+  /**
    * State of the currently being received message from the PSYC service.
    */
   enum MessageState recv_state;
@@ -171,27 +194,12 @@ struct GNUNET_PSYC_Channel
 
 
 /**
- * Handle for a pending PSYC transmission operation.
- */
-struct GNUNET_PSYC_MasterTransmitHandle
-{
-  struct GNUNET_PSYC_Master *master;
-  GNUNET_PSYC_MasterTransmitNotifyModifier notify_mod;
-  GNUNET_PSYC_MasterTransmitNotify notify_data;
-  void *notify_cls;
-  enum MessageState state;
-};
-
-
-/**
  * Handle for the master of a PSYC channel.
  */
 struct GNUNET_PSYC_Master
 {
   struct GNUNET_PSYC_Channel ch;
 
-  struct GNUNET_PSYC_MasterTransmitHandle *tmit;
-
   GNUNET_PSYC_MasterStartCallback start_cb;
 
   uint64_t max_message_id;
@@ -204,6 +212,10 @@ struct GNUNET_PSYC_Master
 struct GNUNET_PSYC_Slave
 {
   struct GNUNET_PSYC_Channel ch;
+
+  GNUNET_PSYC_SlaveJoinCallback join_cb;
+
+  uint64_t max_message_id;
 };
 
 
@@ -251,7 +263,7 @@ reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 
 
 static void
-master_transmit_data (struct GNUNET_PSYC_Master *mst);
+channel_transmit_data (struct GNUNET_PSYC_Channel *ch);
 
 
 /**
@@ -302,7 +314,8 @@ recv_reset (struct GNUNET_PSYC_Channel *ch)
   ch->recv_state = MSG_STATE_START;
   ch->recv_flags = 0;
   ch->recv_message_id = 0;
-  ch->recv_mod_value_size =0;
+  //FIXME: ch->recv_slave_key = { 0 };
+  ch->recv_mod_value_size = 0;
   ch->recv_mod_value_size_expected = 0;
 }
 
@@ -379,8 +392,9 @@ queue_message (struct GNUNET_PSYC_Channel *ch,
   }
  
   if (NULL != op
-      && (end || (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
-                  < op->msg->size + sizeof (struct GNUNET_MessageHeader))))
+      && (GNUNET_YES == end
+          || (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
+              < op->msg->size + sizeof (struct GNUNET_MessageHeader))))
   {
     /* End of message or buffer is full, add it to transmission queue. */
     op->msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
@@ -390,6 +404,9 @@ queue_message (struct GNUNET_PSYC_Channel *ch,
     ch->tmit_ack_pending++;
   }
 
+  if (GNUNET_YES == end)
+    ch->in_transmit = GNUNET_NO;
+
   transmit_next (ch);
 }
 
@@ -400,15 +417,14 @@ queue_message (struct GNUNET_PSYC_Channel *ch,
  * @param mst Master handle.
  */
 static void
-master_transmit_mod (struct GNUNET_PSYC_Master *mst)
+channel_transmit_mod (struct GNUNET_PSYC_Channel *ch)
 {
-  struct GNUNET_PSYC_Channel *ch = &mst->ch;
   uint16_t max_data_size, data_size;
   char data[GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD] = "";
   struct GNUNET_MessageHeader *msg = (struct GNUNET_MessageHeader *) data;
   int notify_ret;
 
-  switch (mst->tmit->state)
+  switch (ch->tmit.state)
   {
   case MSG_STATE_MODIFIER:
   {
@@ -417,12 +433,11 @@ master_transmit_mod (struct GNUNET_PSYC_Master *mst)
     max_data_size = data_size = GNUNET_PSYC_MODIFIER_MAX_PAYLOAD;
     msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER);
     msg->size = sizeof (struct GNUNET_PSYC_MessageModifier);
-    notify_ret = mst->tmit->notify_mod (mst->tmit->notify_cls,
-                                        &data_size, &mod[1], &mod->oper);
+    notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls,
+                                      &data_size, &mod[1], &mod->oper);
     mod->name_size = strnlen ((char *) &mod[1], data_size);
     if (mod->name_size < data_size)
     {
-      mod->oper = htons (mod->oper);
       mod->value_size = htons (data_size - 1 - mod->name_size);
       mod->name_size = htons (mod->name_size);
     }
@@ -438,8 +453,8 @@ master_transmit_mod (struct GNUNET_PSYC_Master *mst)
     max_data_size = data_size = GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD;
     msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT);    
     msg->size = sizeof (struct GNUNET_MessageHeader);
-    notify_ret = mst->tmit->notify_mod (mst->tmit->notify_cls,
-                                        &data_size, &msg[1], NULL);
+    notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls,
+                                      &data_size, &msg[1], NULL);
     break;
   }
   default:
@@ -454,27 +469,28 @@ master_transmit_mod (struct GNUNET_PSYC_Master *mst)
       ch->tmit_paused = GNUNET_YES;
       return;
     }
-    mst->tmit->state = MSG_STATE_MOD_CONT;
+    ch->tmit.state = MSG_STATE_MOD_CONT;
     break;
 
   case GNUNET_YES:
     if (0 == data_size)
     {
       /* End of modifiers. */
-      mst->tmit->state = MSG_STATE_DATA;
+      ch->tmit.state = MSG_STATE_DATA;
       if (0 == ch->tmit_ack_pending)
-        master_transmit_data (mst);
+        channel_transmit_data (ch);
 
       return;
     }
-    mst->tmit->state = MSG_STATE_MODIFIER;
+    ch->tmit.state = MSG_STATE_MODIFIER;
     break;
 
   default:
     LOG (GNUNET_ERROR_TYPE_ERROR,
-         "MasterTransmitNotify returned error when requesting a modifier.\n");
+         "MasterTransmitNotifyModifier returned error "
+         "when requesting a modifier.\n");
 
-    mst->tmit->state = MSG_STATE_START;
+    ch->tmit.state = MSG_STATE_CANCEL;
     msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL);
     msg->size = htons (sizeof (*msg));
 
@@ -489,7 +505,7 @@ master_transmit_mod (struct GNUNET_PSYC_Master *mst)
     queue_message (ch, msg, GNUNET_NO);
   }
 
-  master_transmit_mod (mst);
+  channel_transmit_mod (ch);
 }
 
 
@@ -499,17 +515,16 @@ master_transmit_mod (struct GNUNET_PSYC_Master *mst)
  * @param mst Master handle.
  */
 static void
-master_transmit_data (struct GNUNET_PSYC_Master *mst)
+channel_transmit_data (struct GNUNET_PSYC_Channel *ch)
 {
-  struct GNUNET_PSYC_Channel *ch = &mst->ch;
   uint16_t data_size = GNUNET_PSYC_DATA_MAX_PAYLOAD;
   char data[GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD] = "";
   struct GNUNET_MessageHeader *msg = (struct GNUNET_MessageHeader *) data;
 
   msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA);
 
-  int notify_ret = mst->tmit->notify_data (mst->tmit->notify_cls,
-                                           &data_size, &msg[1]);
+  int notify_ret = ch->tmit.notify_data (ch->tmit.notify_cls,
+                                         &data_size, &msg[1]);
   switch (notify_ret)
   {
   case GNUNET_NO:
@@ -522,14 +537,14 @@ master_transmit_data (struct GNUNET_PSYC_Master *mst)
     break;
 
   case GNUNET_YES:
-    mst->tmit->state = MSG_STATE_START;
+    ch->tmit.state = MSG_STATE_END;
     break;
 
   default:
     LOG (GNUNET_ERROR_TYPE_ERROR,
          "MasterTransmitNotify returned error when requesting data.\n");
 
-    mst->tmit->state = MSG_STATE_START;
+    ch->tmit.state = MSG_STATE_CANCEL;
     msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL);
     msg->size = htons (sizeof (*msg));
     queue_message (ch, msg, GNUNET_YES);
@@ -554,6 +569,86 @@ master_transmit_data (struct GNUNET_PSYC_Master *mst)
 
 
 /**
+ * Send a message to a channel.
+ *
+ * @param ch 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).
+ */
+static struct GNUNET_PSYC_ChannelTransmitHandle *
+channel_transmit (struct GNUNET_PSYC_Channel *ch,
+                  const char *method_name,
+                  GNUNET_PSYC_TransmitNotifyModifier notify_mod,
+                  GNUNET_PSYC_TransmitNotifyData notify_data,
+                  void *notify_cls,
+                  uint32_t flags)
+{
+  if (GNUNET_NO != ch->in_transmit)
+    return NULL;
+  ch->in_transmit = GNUNET_YES;
+
+  size_t size = strlen (method_name) + 1;
+  struct GNUNET_PSYC_MessageMethod *pmeth;
+  struct OperationHandle *op;
+
+  ch->tmit_msg = op = GNUNET_malloc (sizeof (*op) + sizeof (*op->msg)
+                                     + sizeof (*pmeth) + size);
+  op->msg = (struct GNUNET_MessageHeader *) &op[1];
+  op->msg->size = sizeof (*op->msg) + sizeof (*pmeth) + size;
+
+  pmeth = (struct GNUNET_PSYC_MessageMethod *) &op->msg[1];
+  pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD);
+  pmeth->header.size = htons (sizeof (*pmeth) + size);
+  pmeth->flags = htonl (flags);
+  memcpy (&pmeth[1], method_name, size);
+
+  ch->tmit.ch = ch;
+  ch->tmit.notify_mod = notify_mod;
+  ch->tmit.notify_data = notify_data;
+  ch->tmit.notify_cls = notify_cls;
+  ch->tmit.state = MSG_STATE_MODIFIER;
+
+  channel_transmit_mod (ch);
+  return &ch->tmit;
+}
+
+
+/**
+ * Resume transmission to the channel.
+ *
+ * @param th Handle of the request that is being resumed.
+ */
+static void
+channel_transmit_resume (struct GNUNET_PSYC_ChannelTransmitHandle *th)
+{
+  struct GNUNET_PSYC_Channel *ch = th->ch;
+  if (0 == ch->tmit_ack_pending)
+  {
+    ch->tmit_paused = GNUNET_NO;
+    channel_transmit_data (ch);
+  }
+}
+
+
+/**
+ * Abort transmission request to channel.
+ *
+ * @param th Handle of the request that is being aborted.
+ */
+static void
+channel_transmit_cancel (struct GNUNET_PSYC_ChannelTransmitHandle *th)
+{
+  struct GNUNET_PSYC_Channel *ch = th->ch;
+  if (GNUNET_NO == ch->in_transmit)
+    return;
+}
+
+
+/**
  * Handle incoming message from the PSYC service.
  *
  * @param ch The channel the message is sent to.
@@ -564,14 +659,20 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
                      const struct GNUNET_PSYC_MessageHeader *msg)
 {
   uint16_t size = ntohs (msg->header.size);
+  uint32_t flags = ntohl (msg->flags);
+
+  GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG,
+                           (struct GNUNET_MessageHeader *) msg);
 
   if (MSG_STATE_START == ch->recv_state)
   {
     ch->recv_message_id = GNUNET_ntohll (msg->message_id);
-    ch->recv_flags = ntohl (msg->flags);
+    ch->recv_flags = flags;
+    ch->recv_slave_key = msg->slave_key;
   }
   else if (GNUNET_ntohll (msg->message_id) != ch->recv_message_id)
   {
+    // FIXME
     LOG (GNUNET_ERROR_TYPE_WARNING,
          "Unexpected message ID. Got: %" PRIu64 ", expected: %" PRIu64 "\n",
          GNUNET_ntohll (msg->message_id), ch->recv_message_id);
@@ -579,11 +680,11 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
     recv_error (ch);
     return;
   }
-  else if (ntohl (msg->flags) != ch->recv_flags)
+  else if (flags != ch->recv_flags)
   {
     LOG (GNUNET_ERROR_TYPE_WARNING,
          "Unexpected message flags. Got: %lu, expected: %lu\n",
-         ntohl (msg->flags), ch->recv_flags);
+         flags, ch->recv_flags);
     GNUNET_break_op (0);
     recv_error (ch);
     return;
@@ -599,10 +700,6 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
     ptype = ntohs (pmsg->type);
     size_eq = size_min = 0;
 
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Received message part of type %u and size %u from PSYC.\n",
-                ptype, psize);
-
     if (psize < sizeof (*pmsg) || sizeof (*msg) + pos + psize > size)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
@@ -612,6 +709,10 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
       return;
     }
 
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Received message part from PSYC.\n");
+    GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, pmsg);
+
     switch (ptype)
     {
     case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
@@ -758,6 +859,46 @@ handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
 
 
 /**
+ * Handle incoming message acknowledgement from the PSYC service.
+ *
+ * @param ch The channel the acknowledgement is sent to.
+ */
+static void
+handle_psyc_message_ack (struct GNUNET_PSYC_Channel *ch)
+{
+  if (0 == ch->tmit_ack_pending)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING, "Ignoring extraneous message ACK\n");
+    GNUNET_break (0);
+    return;
+  }
+  ch->tmit_ack_pending--;
+
+  switch (ch->tmit.state)
+  {
+  case MSG_STATE_MODIFIER:
+  case MSG_STATE_MOD_CONT:
+    if (GNUNET_NO == ch->tmit_paused)
+      channel_transmit_mod (ch);
+    break;
+
+  case MSG_STATE_DATA:
+    if (GNUNET_NO == ch->tmit_paused)
+      channel_transmit_data (ch);
+    break;
+
+  case MSG_STATE_END:
+  case MSG_STATE_CANCEL:
+    break;
+
+  default:
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Ignoring message ACK in state %u.\n", ch->tmit.state);
+  }
+}
+
+
+/**
  * Type of a function to call when we receive a message
  * from the service.
  *
@@ -775,7 +916,7 @@ message_handler (void *cls,
 
   if (NULL == msg)
   {
-    GNUNET_break (0);
+    // timeout / disconnected from server, reconnect
     reschedule_connect (ch);
     return;
   }
@@ -824,63 +965,15 @@ message_handler (void *cls,
   }
   case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
   {
-#if TODO
     struct CountersResult *cres = (struct CountersResult *) msg;
     slv->max_message_id = GNUNET_ntohll (cres->max_message_id);
-    if (NULL != slv->join_ack_cb)
-      mst->join_ack_cb (ch->cb_cls, mst->max_message_id);
-#endif
+    if (NULL != slv->join_cb)
+      slv->join_cb (ch->cb_cls, slv->max_message_id);
     break;
   }
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK:
   {
-    if (0 == ch->tmit_ack_pending)
-    {
-      LOG (GNUNET_ERROR_TYPE_WARNING, "Ignoring extraneous message ACK\n");
-      GNUNET_break (0);
-      break;
-    }
-    ch->tmit_ack_pending--;
-
-    if (ch->is_master)
-    {
-      GNUNET_assert (NULL != mst->tmit);
-      switch (mst->tmit->state)
-      {
-      case MSG_STATE_MODIFIER:
-      case MSG_STATE_MOD_CONT:
-        if (GNUNET_NO == ch->tmit_paused)
-          master_transmit_mod (mst);
-        break;
-
-      case MSG_STATE_DATA:
-        if (GNUNET_NO == ch->tmit_paused)
-          master_transmit_data (mst);
-        break;
-
-      case MSG_STATE_END:
-      case MSG_STATE_CANCEL:
-        if (NULL != mst->tmit)
-        {
-          GNUNET_free (mst->tmit);
-          mst->tmit = NULL;
-        }
-        else
-        {
-          LOG (GNUNET_ERROR_TYPE_WARNING,
-               "Ignoring message ACK, there's no transmission going on.\n");
-          GNUNET_break (0);
-        }
-        break;
-      default:
-        LOG (GNUNET_ERROR_TYPE_DEBUG,
-             "Ignoring message ACK in state %u.\n", mst->tmit->state);
-      }
-    }
-    else
-    {
-      /* TODO: slave */
-    }
+    handle_psyc_message_ack (ch);
     break;
   }
 
@@ -1106,8 +1199,6 @@ void
 GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master)
 {
   disconnect (master);
-  if (NULL != master->tmit)
-    GNUNET_free (master->tmit);
   GNUNET_free (master);
 }
 
@@ -1162,41 +1253,14 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
 struct GNUNET_PSYC_MasterTransmitHandle *
 GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
                              const char *method_name,
-                             GNUNET_PSYC_MasterTransmitNotifyModifier notify_mod,
-                             GNUNET_PSYC_MasterTransmitNotify notify_data,
+                             GNUNET_PSYC_TransmitNotifyModifier notify_mod,
+                             GNUNET_PSYC_TransmitNotifyData notify_data,
                              void *notify_cls,
                              enum GNUNET_PSYC_MasterTransmitFlags flags)
 {
-  GNUNET_assert (NULL != master);
-  struct GNUNET_PSYC_Channel *ch = &master->ch;
-  if (GNUNET_NO != ch->in_transmit)
-    return NULL;
-  ch->in_transmit = GNUNET_YES;
-
-  size_t size = strlen (method_name) + 1;
-  struct GNUNET_PSYC_MessageMethod *pmeth;
-  struct OperationHandle *op;
-
-  ch->tmit_msg = op = GNUNET_malloc (sizeof (*op) + sizeof (*op->msg)
-                                     + sizeof (*pmeth) + size);
-  op->msg = (struct GNUNET_MessageHeader *) &op[1];
-  op->msg->size = sizeof (*op->msg) + sizeof (*pmeth) + size;
-
-  pmeth = (struct GNUNET_PSYC_MessageMethod *) &op->msg[1];
-  pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD);
-  pmeth->header.size = htons (sizeof (*pmeth) + size);
-  pmeth->flags = htonl (flags);
-  memcpy (&pmeth[1], method_name, size);
-
-  master->tmit = GNUNET_malloc (sizeof (*master->tmit));
-  master->tmit->master = master;
-  master->tmit->notify_mod = notify_mod;
-  master->tmit->notify_data = notify_data;
-  master->tmit->notify_cls = notify_cls;
-  master->tmit->state = MSG_STATE_MODIFIER;
-
-  master_transmit_mod (master);
-  return master->tmit;
+  return (struct GNUNET_PSYC_MasterTransmitHandle *)
+    channel_transmit (&master->ch, method_name, notify_mod, notify_data,
+                      notify_cls, flags);
 }
 
 
@@ -1208,12 +1272,7 @@ GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
 void
 GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th)
 {
-  struct GNUNET_PSYC_Channel *ch = &th->master->ch;
-  if (0 == ch->tmit_ack_pending)
-  {
-    ch->tmit_paused = GNUNET_NO;
-    master_transmit_data (th->master);
-  }
+  channel_transmit_resume ((struct GNUNET_PSYC_ChannelTransmitHandle *) th);
 }
 
 
@@ -1225,10 +1284,7 @@ GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th)
 void
 GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th)
 {
-  struct GNUNET_PSYC_Master *master = th->master;
-  struct GNUNET_PSYC_Channel *ch = &master->ch;
-  if (GNUNET_NO != ch->in_transmit)
-    return;
+  channel_transmit_cancel ((struct GNUNET_PSYC_ChannelTransmitHandle *) th);
 }
 
 
@@ -1282,15 +1338,15 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
 {
   struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv));
   struct GNUNET_PSYC_Channel *ch = &slv->ch;
-  struct SlaveJoinRequest *req = GNUNET_malloc (sizeof (*req)
-                                                + relay_count * sizeof (*relays));
+  struct SlaveJoinRequest *req
+    = GNUNET_malloc (sizeof (*req) + relay_count * sizeof (*relays));
   req->header.size = htons (sizeof (*req)
                             + relay_count * sizeof (*relays));
   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN);
   req->channel_key = *channel_key;
   req->slave_key = *slave_key;
   req->origin = *origin;
-  req->relay_count = relay_count;
+  req->relay_count = htonl (relay_count);
   memcpy (&req[1], relays, relay_count * sizeof (*relays));
 
   ch->message_cb = message_cb;
@@ -1303,6 +1359,7 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
   ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, slv);
 
+  slv->join_cb = slave_joined_cb;
   return slv;
 }
 
@@ -1328,9 +1385,8 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave)
  *
  * @param slave Slave handle.
  * @param method_name Which (PSYC) method should be invoked (on host).
- * @param env Environment containing transient variables for the message, or
- *            NULL.
- * @param notify Function to call when we are allowed to transmit (to get data).
+ * @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
@@ -1339,12 +1395,14 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave)
 struct GNUNET_PSYC_SlaveTransmitHandle *
 GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slave,
                             const char *method_name,
-                            const struct GNUNET_ENV_Environment *env,
-                            GNUNET_PSYC_SlaveTransmitNotify notify,
+                            GNUNET_PSYC_TransmitNotifyModifier notify_mod,
+                            GNUNET_PSYC_TransmitNotifyData notify_data,
                             void *notify_cls,
                             enum GNUNET_PSYC_SlaveTransmitFlags flags)
 {
-  return NULL;
+  return (struct GNUNET_PSYC_SlaveTransmitHandle *)
+    channel_transmit (&slave->ch, method_name,
+                      notify_mod, notify_data, notify_cls, flags);
 }
 
 
@@ -1356,7 +1414,7 @@ GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slave,
 void
 GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th)
 {
-
+  channel_transmit_resume ((struct GNUNET_PSYC_ChannelTransmitHandle *) th);
 }
 
 
@@ -1368,7 +1426,7 @@ GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th)
 void
 GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *th)
 {
-
+  channel_transmit_cancel ((struct GNUNET_PSYC_ChannelTransmitHandle *) th);
 }
 
 
@@ -1382,7 +1440,7 @@ GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *th)
 struct GNUNET_PSYC_Channel *
 GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
 {
-  return (struct GNUNET_PSYC_Channel *) master;
+  return &master->ch;
 }
 
 
@@ -1395,7 +1453,7 @@ GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
 struct GNUNET_PSYC_Channel *
 GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slave)
 {
-  return (struct GNUNET_PSYC_Channel *) slave;
+  return &slave->ch;
 }
 
 
diff --git a/src/psyc/psyc_common.c b/src/psyc/psyc_common.c
new file mode 100644
index 000000000..7368011fc
--- /dev/null
+++ b/src/psyc/psyc_common.c
@@ -0,0 +1,100 @@
+/*
+ * This file is part of GNUnet
+ * (C) 2013 Christian Grothoff (and other contributing authors)
+ *
+ * GNUnet is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published
+ * by the Free Software Foundation; either version 3, or (at your
+ * option) any later version.
+ *
+ * GNUnet is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with GNUnet; see the file COPYING.  If not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/**
+ * @file psyc/psyc_common.c
+ * @brief Common functions for PSYC
+ * @author Gabor X Toth
+ */
+
+#include <inttypes.h>
+#include "psyc.h"
+
+/**
+ * Check if @a data contains a series of valid message parts.
+ *
+ * @param data_size  Size of @a data.
+ * @param data       Data.
+ *
+ * @return GNUNET_YES or GNUNET_NO
+ */
+int
+GNUNET_PSYC_check_message_parts (uint16_t data_size, const char *data)
+{
+  const struct GNUNET_MessageHeader *pmsg;
+  uint16_t psize = 0;
+  uint16_t pos = 0;
+
+  for (pos = 0; data_size + pos < data_size; pos += psize)
+  {
+    pmsg = (const struct GNUNET_MessageHeader *) (data + pos);
+    psize = ntohs (pmsg->size);
+    if (psize < sizeof (*pmsg) || data_size + pos + psize > data_size)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Invalid message part of type %u and size %u.",
+                  ntohs (pmsg->type), psize);
+      return GNUNET_NO;
+    }
+  }
+  return GNUNET_YES;
+}
+
+
+void
+GNUNET_PSYC_log_message (enum GNUNET_ErrorType kind,
+                         const struct GNUNET_MessageHeader *msg)
+{
+  uint16_t size = ntohs (msg->size);
+  uint16_t type = ntohs (msg->type);
+  GNUNET_log (kind, "Message of type %d and size %u:\n", type, size);
+  switch (type)
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE:
+  {
+    struct GNUNET_PSYC_MessageHeader *pmsg
+      = (struct GNUNET_PSYC_MessageHeader *) msg;
+    GNUNET_log (kind, "\tID: %" PRIu64 "\tflags: %" PRIu32 "\n",
+                GNUNET_ntohll (pmsg->message_id), ntohl (pmsg->flags));
+    break;
+  }
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+  {
+    struct GNUNET_PSYC_MessageMethod *meth
+      = (struct GNUNET_PSYC_MessageMethod *) msg;
+    GNUNET_log (kind, "\t%.*s\n", size - sizeof (*meth), &meth[1]);
+    break;
+  }
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+  {
+    struct GNUNET_PSYC_MessageModifier *mod
+      = (struct GNUNET_PSYC_MessageModifier *) msg;
+    uint16_t name_size = ntohs (mod->name_size);
+    char oper = ' ' < mod->oper ? mod->oper : ' ';
+    GNUNET_log (kind, "\t%c%.*s\t%.*s\n", oper, name_size, &mod[1],
+                ntohs (mod->value_size), ((char *) &mod[1]) + name_size + 1);
+    break;
+  }
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+    GNUNET_log (kind, "\t%.*s\n", size - sizeof (*msg), &msg[1]);
+    break;
+  }
+}
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 33684b125..88947be60 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -37,7 +37,7 @@
 
 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10)
 
-#define DEBUG_SERVICE 0
+#define DEBUG_SERVICE 1
 
 
 /**
@@ -66,7 +66,8 @@ struct GNUNET_PSYC_MasterTransmitHandle *mth;
 
 struct TransmitClosure
 {
-  struct GNUNET_PSYC_MasterTransmitHandle *handle;
+  struct GNUNET_PSYC_MasterTransmitHandle *mst_tmit;
+  struct GNUNET_PSYC_SlaveTransmitHandle *slv_tmit;
   struct GNUNET_ENV_Environment *env;
   char *data[16];
   const char *mod_value;
@@ -78,12 +79,30 @@ struct TransmitClosure
 
 struct TransmitClosure *tmit;
 
+
+enum
+{
+  TEST_NONE,
+  TEST_SLAVE_TRANSMIT,
+  TEST_MASTER_TRANSMIT,
+} test;
+
+
+static void
+master_transmit ();
+
+
 /**
  * Clean up all resources used.
  */
 static void
 cleanup ()
 {
+  if (NULL != slv)
+  {
+    GNUNET_PSYC_slave_part (slv);
+    slv = NULL;
+  }
   if (NULL != mst)
   {
     GNUNET_PSYC_master_stop (mst);
@@ -133,6 +152,8 @@ end_normally (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 static void
 end ()
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Ending tests.\n");
+
   if (end_badly_task != GNUNET_SCHEDULER_NO_TASK)
   {
     GNUNET_SCHEDULER_cancel (end_badly_task);
@@ -144,8 +165,8 @@ end ()
 
 
 static void
-message (void *cls, uint64_t message_id, uint32_t flags,
-         const struct GNUNET_MessageHeader *msg)
+master_message (void *cls, uint64_t message_id, uint32_t flags,
+                const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
   {
@@ -158,12 +179,64 @@ message (void *cls, uint64_t message_id, uint32_t flags,
   uint16_t size = ntohs (msg->size);
 
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Got message part of type %u and size %u "
+              "Master got message part of type %u and size %u "
               "belonging to message ID %llu with flags %u\n",
               type, size, message_id, flags);
 
-  if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END == type)
-    end ();
+  switch (test)
+  {
+  case TEST_SLAVE_TRANSMIT:
+    if (GNUNET_PSYC_MESSAGE_REQUEST != flags)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Unexpected request flags: %lu\n", 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:
+    break;
+
+  default:
+    GNUNET_assert (0);
+  }
+}
+
+
+static void
+slave_message (void *cls, uint64_t message_id, uint32_t flags,
+               const struct GNUNET_MessageHeader *msg)
+{
+  if (NULL == msg)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Error while receiving message %llu\n", message_id);
+    return;
+  }
+
+  uint16_t type = ntohs (msg->type);
+  uint16_t size = ntohs (msg->size);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Slave got message part of type %u and size %u "
+              "belonging to message ID %llu with flags %u\n",
+              type, size, message_id, flags);
+
+  switch (test)
+  {
+  case TEST_MASTER_TRANSMIT:
+    if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END == type)
+      end ();
+    break;
+
+  default:
+    GNUNET_assert (0);
+  }
 }
 
 
@@ -175,7 +248,9 @@ join_request (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
               struct GNUNET_PSYC_JoinHandle *jh)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Got join request.");
+              "Got join request: %s (%zu vars)", method_name, variable_count);
+  GNUNET_PSYC_join_decision (jh, GNUNET_YES, 0, NULL, "_notice_join", NULL,
+                             "you're in", 9);
 }
 
 
@@ -185,7 +260,7 @@ transmit_resume (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission resumed.\n");
   struct TransmitClosure *tmit = cls;
   tmit->paused = GNUNET_NO;
-  GNUNET_PSYC_master_transmit_resume (tmit->handle);
+  GNUNET_PSYC_master_transmit_resume (tmit->mst_tmit);
 }
 
 
@@ -204,35 +279,8 @@ tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper)
   uint16_t name_size = 0;
   size_t value_size = 0;
 
-  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;
-    memcpy (data, value, value_size);
-  }
-  else if (NULL != oper)
-  {
+  if (NULL != oper)
+  { /* New modifier */
     if (GNUNET_NO == GNUNET_ENV_environment_shift (tmit->env, &op, &name,
                                                    (void *) &value, &value_size))
     { /* No more modifiers, continue with data */
@@ -259,6 +307,33 @@ tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper)
     ((char *)data)[name_size] = '\0';
     memcpy ((char *)data + name_size + 1, 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;
+    memcpy (data, value, value_size);
+  }
 
   return 0 == tmit->mod_value_size ? GNUNET_YES : GNUNET_NO;
 }
@@ -268,6 +343,12 @@ 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,
               "Transmit notify data: %lu bytes available, "
@@ -300,10 +381,52 @@ tmit_notify_data (void *cls, uint16_t *data_size, void *data)
 
 
 static void
-master_started (void *cls, uint64_t max_message_id)
+slave_joined (void *cls, uint64_t max_message_id)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Master started: %" PRIu64 "\n", max_message_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave joined: %lu\n", max_message_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave sending request to master.\n");
+
+  test = TEST_SLAVE_TRANSMIT;
+
+  tmit = GNUNET_new (struct TransmitClosure);
+  tmit->env = GNUNET_ENV_environment_create ();
+  GNUNET_ENV_environment_add (tmit->env, GNUNET_ENV_OP_ASSIGN,
+                              "_abc", "abc def", 7);
+  GNUNET_ENV_environment_add (tmit->env, GNUNET_ENV_OP_ASSIGN,
+                              "_abc_def", "abc def ghi", 11);
+  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_join ()
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Joining slave.\n");
+
+  struct GNUNET_PeerIdentity origin;
+  struct GNUNET_PeerIdentity relays[16];
+  struct GNUNET_ENV_Environment *env = GNUNET_ENV_environment_create ();
+  GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
+                              "_foo", "bar baz", 7);
+  GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
+                              "_foo_bar", "foo bar baz", 11);
+  slv = GNUNET_PSYC_slave_join (cfg, &channel_pub_key, slave_key, &origin,
+                                16, relays, &slave_message, &join_request, &slave_joined,
+                                NULL, "_request_join", env, "some data", 9);
+  GNUNET_ENV_environment_destroy (env);
+}
+
+
+static void
+master_transmit ()
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Master sending message to all.\n");
+  test = TEST_MASTER_TRANSMIT;
 
   tmit = GNUNET_new (struct TransmitClosure);
   tmit->env = GNUNET_ENV_environment_create ();
@@ -315,17 +438,19 @@ master_started (void *cls, uint64_t max_message_id)
   tmit->data[1] = "foo bar";
   tmit->data[2] = "foo bar baz";
   tmit->data_count = 3;
-  tmit->handle
-    = GNUNET_PSYC_master_transmit (mst, "_test", tmit_notify_mod,
+  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
-slave_joined (void *cls, uint64_t max_message_id)
+master_started (void *cls, uint64_t max_message_id)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave joined: %lu\n", max_message_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Master started: %" PRIu64 "\n", max_message_id);
+  slave_join ();
 }
 
 
@@ -355,21 +480,9 @@ run (void *cls,
   GNUNET_CRYPTO_eddsa_key_get_public (channel_key, &channel_pub_key);
   GNUNET_CRYPTO_eddsa_key_get_public (slave_key, &slave_pub_key);
 
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Starting master.\n");
   mst = GNUNET_PSYC_master_start (cfg, channel_key, GNUNET_PSYC_CHANNEL_PRIVATE,
-                                  &message, &join_request, &master_started, NULL);
-  return; /* FIXME: test slave */
-
-  struct GNUNET_PeerIdentity origin;
-  struct GNUNET_PeerIdentity relays[16];
-  struct GNUNET_ENV_Environment *env = GNUNET_ENV_environment_create ();
-  GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
-                              "_foo", "bar baz", 7);
-  GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
-                              "_foo_bar", "foo bar baz", 11);
-  slv = GNUNET_PSYC_slave_join (cfg, &channel_pub_key, slave_key, &origin,
-                                16, relays, &message, &join_request, &slave_joined,
-                                NULL, "_request_join", env, "some data", 9);
-  GNUNET_ENV_environment_destroy (env);
+                                  &master_message, &join_request, &master_started, NULL);
 }