social: implement enter/leave/messaging; psyc: improvements and fixes

- social: implement enter/leave, send/receive messages, slicer
- psyc, social: add struct GNUNET_PSYC_Message for single-fragment join messages
- psyc: add message callback in addition to message part callback
- client_manager, social, psyc, multicast: add disconnect callback

5 files changed, 456 insertions, 166 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 866275a79..8fc080baf 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -168,7 +168,7 @@ struct FragmentQueue
    * Is the message queued for delivery to the client?
    * i.e. added to the recv_msgs queue
    */
-  uint8_t queued;
+  uint8_t is_queued;
 };
 
 
@@ -382,7 +382,7 @@ struct Slave
   /**
    * Join request to be transmitted to the master on join.
    */
-  struct GNUNET_MessageHeader *join_req;
+  struct GNUNET_PSYC_Message *join_msg;
 
   /**
    * Join decision received from multicast.
@@ -435,7 +435,7 @@ cleanup_master (struct Master *mst)
   struct Channel *chn = &mst->chn;
 
   if (NULL != mst->origin)
-    GNUNET_MULTICAST_origin_stop (mst->origin);
+    GNUNET_MULTICAST_origin_stop (mst->origin, NULL, NULL); // FIXME
   GNUNET_CONTAINER_multihashmap_destroy (mst->join_reqs);
   GNUNET_CONTAINER_multihashmap_remove (masters, &chn->pub_key_hash, chn);
 }
@@ -462,12 +462,21 @@ cleanup_slave (struct Slave *slv)
   }
   GNUNET_CONTAINER_multihashmap_remove (slaves, &chn->pub_key_hash, slv);
 
-  if (NULL != slv->join_req)
-    GNUNET_free (slv->join_req);
+  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;
+  }
   if (NULL != slv->member)
-    GNUNET_MULTICAST_member_part (slv->member);
+  {
+    GNUNET_MULTICAST_member_part (slv->member, NULL, NULL); // FIXME
+    slv->member = NULL;
+  }
   GNUNET_CONTAINER_multihashmap_remove (slaves, &chn->pub_key_hash, chn);
 }
 
@@ -482,7 +491,10 @@ cleanup_channel (struct Channel *chn)
   GNUNET_CONTAINER_multihashmap_remove_all (recv_cache, &chn->pub_key_hash);
 
   if (NULL != chn->store_op)
+  {
     GNUNET_PSYCSTORE_operation_cancel (chn->store_op);
+    chn->store_op = NULL;
+  }
 
   (GNUNET_YES == chn->is_master)
     ? cleanup_master ((struct Master *) chn)
@@ -574,7 +586,7 @@ struct JoinMemTestClosure
   struct GNUNET_CRYPTO_EcdsaPublicKey slave_key;
   struct Channel *chn;
   struct GNUNET_MULTICAST_JoinHandle *jh;
-  struct MasterJoinRequest *master_join_req;
+  struct GNUNET_PSYC_JoinRequestMessage *join_msg;
 };
 
 
@@ -594,14 +606,14 @@ join_mem_test_cb (void *cls, int64_t result, const char *err_msg)
                         &slave_key_hash);
     GNUNET_CONTAINER_multihashmap_put (mst->join_reqs, &slave_key_hash, jcls->jh,
                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
-    client_send_msg (jcls->chn, &jcls->master_join_req->header);
+    client_send_msg (jcls->chn, &jcls->join_msg->header);
   }
   else
   {
     // FIXME: add relays
     GNUNET_MULTICAST_join_decision (jcls->jh, result, 0, NULL, NULL);
   }
-  GNUNET_free (jcls->master_join_req);
+  GNUNET_free (jcls->join_msg);
   GNUNET_free (jcls);
 }
 
@@ -633,7 +645,8 @@ mcast_recv_join_request (void *cls,
     }
   }
 
-  struct MasterJoinRequest *req = GNUNET_malloc (sizeof (*req) + join_msg_size);
+  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_key = *slave_key;
@@ -644,7 +657,7 @@ mcast_recv_join_request (void *cls,
   jcls->slave_key = *slave_key;
   jcls->chn = chn;
   jcls->jh = jh;
-  jcls->master_join_req = req;
+  jcls->join_msg = req;
 
   GNUNET_PSYCSTORE_membership_test (store, &chn->pub_key, slave_key,
                                     chn->max_message_id, 0,
@@ -780,6 +793,7 @@ client_send_mcast_msg (struct Channel *chn,
   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;
 
   memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
   client_send_msg (chn, &pmsg->header);
@@ -810,6 +824,7 @@ client_send_mcast_req (struct Master *mst,
   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);
 
   memcpy (&pmsg[1], &req[1], size - sizeof (*req));
@@ -870,11 +885,12 @@ fragment_queue_insert (struct Channel *chn,
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                 "%p Adding message fragment to cache. "
-                "message_id: %" PRIu64 ", fragment_id: %" PRIu64 ", "
-                "header_size: %" PRIu64 " + %u).\n",
+                "message_id: %" PRIu64 ", fragment_id: %" PRIu64 "\n",
                 chn, GNUNET_ntohll (mmsg->message_id),
-                GNUNET_ntohll (mmsg->fragment_id),
-                fragq->header_size, size);
+                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_new (struct RecvCacheEntry);
     cache_entry->ref_count = 1;
     cache_entry->mmsg = GNUNET_malloc (size);
@@ -955,11 +971,11 @@ fragment_queue_insert (struct Channel *chn,
   case MSG_FRAG_STATE_DATA:
   case MSG_FRAG_STATE_END:
   case MSG_FRAG_STATE_CANCEL:
-    if (GNUNET_NO == fragq->queued)
+    if (GNUNET_NO == fragq->is_queued)
     {
       GNUNET_CONTAINER_heap_insert (chn->recv_msgs, NULL,
                                     GNUNET_ntohll (mmsg->message_id));
-      fragq->queued = GNUNET_YES;
+      fragq->is_queued = GNUNET_YES;
     }
   }
 
@@ -1034,7 +1050,7 @@ fragment_queue_run (struct Channel *chn, uint64_t msg_id,
   if (MSG_FRAG_STATE_END <= fragq->state)
   {
     struct GNUNET_HashCode msg_id_hash;
-    hash_key_from_nll (&msg_id_hash, msg_id);
+    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);
@@ -1042,7 +1058,7 @@ fragment_queue_run (struct Channel *chn, uint64_t msg_id,
   }
   else
   {
-    fragq->queued = GNUNET_NO;
+    fragq->is_queued = GNUNET_NO;
   }
 }
 
@@ -1331,13 +1347,18 @@ store_recv_slave_counters (void *cls, int result, uint64_t max_fragment_id,
       = GNUNET_MULTICAST_member_join (cfg, &chn->pub_key, &slv->priv_key,
                                       &slv->origin,
                                       slv->relay_count, slv->relays,
-                                      slv->join_req,
+                                      &slv->join_msg->header,
                                       &mcast_recv_join_request,
                                       &mcast_recv_join_decision,
                                       &mcast_recv_membership_test,
                                       &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
   {
@@ -1435,6 +1456,7 @@ client_recv_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
 {
   const struct SlaveJoinRequest *req
     = (const struct SlaveJoinRequest *) msg;
+  uint16_t req_size = ntohs (req->header.size);
 
   struct GNUNET_CRYPTO_EcdsaPublicKey slv_pub_key;
   struct GNUNET_HashCode pub_key_hash, slv_pub_key_hash;
@@ -1460,15 +1482,32 @@ client_recv_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
     slv->pub_key_hash = slv_pub_key_hash;
     slv->origin = req->origin;
     slv->relay_count = ntohl (req->relay_count);
+
+    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)
+    {
+      join_msg_size = ntohs (slv->join_msg->header.size);
+      slv->join_msg = GNUNET_malloc (join_msg_size);
+      memcpy (slv->join_msg, ((char *) &req[1]) + relay_size, join_msg_size);
+    }
+    if (sizeof (*req) + relay_size + join_msg_size != req_size)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "%u + %u + %u != %u\n",
+                  sizeof (*req), relay_size, join_msg_size, req_size);
+      GNUNET_break (0);
+      GNUNET_SERVER_client_disconnect (client);
+      return;
+    }
     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));
+      slv->relays = GNUNET_malloc (relay_size);
+      memcpy (slv->relays, &req[1], relay_size);
     }
 
     chn = &slv->chn;
@@ -1510,14 +1549,18 @@ client_recv_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
         = GNUNET_MULTICAST_member_join (cfg, &chn->pub_key, &slv->priv_key,
                                         &slv->origin,
                                         slv->relay_count, slv->relays,
-                                        slv->join_req,
+                                        &slv->join_msg->header,
                                         &mcast_recv_join_request,
                                         &mcast_recv_join_decision,
                                         &mcast_recv_membership_test,
                                         &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)
     {
@@ -1549,13 +1592,14 @@ struct JoinDecisionClosure
 
 
 /**
- * Iterator callback for responding to join requests of a slave.
+ * Iterator callback for sending join decisions to multicast.
  */
 static int
 mcast_send_join_decision (void *cls, const struct GNUNET_HashCode *pub_key_hash,
-                          void *jh)
+                          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;
@@ -1579,8 +1623,7 @@ client_recv_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
   struct JoinDecisionClosure jcls;
   jcls.is_admitted = ntohl (dcsn->is_admitted);
   jcls.msg
-    = (sizeof (*dcsn) + sizeof (struct GNUNET_PSYC_MessageHeader)
-       <= ntohs (msg->size))
+    = (sizeof (*dcsn) + sizeof (*jcls.msg) <= ntohs (msg->size))
     ? (struct GNUNET_MessageHeader *) &dcsn[1]
     : NULL;
 
@@ -1901,6 +1944,9 @@ client_recv_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
     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);
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index 995fb1fa4..82800a334 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -229,22 +229,6 @@ struct OperationResult
    */
 };
 
-
-struct MasterJoinRequest
-{
-  /**
-   * Type: GNUNET_MESSAGE_TYPE_PSYC_MASTER_JOIN_REQUEST
-   */
-  struct GNUNET_MessageHeader header;
-  /**
-   * Public key of the joining slave.
-   */
-  struct GNUNET_CRYPTO_EcdsaPublicKey slave_key;
-
-  /* Followed by struct GNUNET_MessageHeader join_request */
-};
-
-
 GNUNET_NETWORK_STRUCT_END
 
 #endif
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index 88b007a0f..a43b1ef5f 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -74,6 +74,16 @@ struct GNUNET_PSYC_Channel
   struct GNUNET_MessageHeader *connect_msg;
 
   /**
+   * 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;
@@ -82,6 +92,12 @@ struct GNUNET_PSYC_Channel
    * 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;
 };
 
 
@@ -232,19 +248,26 @@ master_recv_join_request (void *cls,
   struct GNUNET_PSYC_Master *
     mst = GNUNET_CLIENT_MANAGER_get_user_context_ (client,
                                                    sizeof (struct GNUNET_PSYC_Channel));
-
-  const struct MasterJoinRequest *req = (const struct MasterJoinRequest *) msg;
-
-  struct GNUNET_PSYC_MessageHeader *pmsg = NULL;
-  if (ntohs (req->header.size) <= sizeof (*req) + sizeof (*pmsg))
-    pmsg = (struct GNUNET_PSYC_MessageHeader *) &req[1];
+  if (NULL == mst->join_req_cb)
+    return;
+
+  const struct GNUNET_PSYC_JoinRequestMessage *
+    req = (const struct GNUNET_PSYC_JoinRequestMessage *) msg;
+  const struct GNUNET_PSYC_Message *join_msg = NULL;
+  if (sizeof (*req) + sizeof (*join_msg) <= ntohs (req->header.size))
+  {
+    join_msg = (struct GNUNET_PSYC_Message *) &req[1];
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+         "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_key = req->slave_key;
 
   if (NULL != mst->join_req_cb)
-    mst->join_req_cb (mst->cb_cls, &req->slave_key, pmsg, jh);
+    mst->join_req_cb (mst->cb_cls, req, &req->slave_key, join_msg, jh);
 }
 
 
@@ -273,13 +296,12 @@ slave_recv_join_decision (void *cls,
   const struct GNUNET_PSYC_JoinDecisionMessage *
     dcsn = (const struct GNUNET_PSYC_JoinDecisionMessage *) msg;
 
-  struct GNUNET_PSYC_MessageHeader *pmsg = NULL;
+  struct GNUNET_PSYC_Message *pmsg = NULL;
   if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg))
-    pmsg = (struct GNUNET_PSYC_MessageHeader *) &dcsn[1];
+    pmsg = (struct GNUNET_PSYC_Message *) &dcsn[1];
 
-  struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh));
   if (NULL != slv->join_dcsn_cb)
-    slv->join_dcsn_cb (slv->cb_cls, ntohl (dcsn->is_admitted), pmsg);
+    slv->join_dcsn_cb (slv->cb_cls, dcsn, ntohl (dcsn->is_admitted), pmsg);
 }
 
 
@@ -299,7 +321,7 @@ static struct GNUNET_CLIENT_MANAGER_MessageHandler master_handlers[] =
 
   { &master_recv_join_request, NULL,
     GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST,
-    sizeof (struct MasterJoinRequest), GNUNET_YES },
+    sizeof (struct GNUNET_PSYC_JoinRequestMessage), GNUNET_YES },
 
   { &channel_recv_disconnect, NULL, 0, 0, GNUNET_NO },
 
@@ -331,6 +353,35 @@ static struct GNUNET_CLIENT_MANAGER_MessageHandler slave_handlers[] =
 };
 
 
+static void
+channel_cleanup (struct GNUNET_PSYC_Channel *chn)
+{
+  GNUNET_PSYC_transmit_destroy (chn->tmit);
+  GNUNET_PSYC_receive_destroy (chn->recv);
+  GNUNET_free (chn->connect_msg);
+  if (NULL != chn->disconnect_cb)
+    chn->disconnect_cb (chn->disconnect_cls);
+}
+
+
+static void
+master_cleanup (void *cls)
+{
+  struct GNUNET_PSYC_Master *mst = cls;
+  channel_cleanup (&mst->chn);
+  GNUNET_free (mst);
+}
+
+
+static void
+slave_cleanup (void *cls)
+{
+  struct GNUNET_PSYC_Slave *slv = cls;
+  channel_cleanup (&slv->chn);
+  GNUNET_free (slv);
+}
+
+
 /**
  * Start a PSYC master channel.
  *
@@ -367,6 +418,7 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
                           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_malloc (sizeof (*mst));
@@ -390,7 +442,7 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
   GNUNET_CLIENT_MANAGER_set_user_context_ (chn->client, mst, sizeof (*chn));
 
   chn->tmit = GNUNET_PSYC_transmit_create (chn->client);
-  chn->recv = GNUNET_PSYC_receive_create (message_cb, message_cb, cls);
+  chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
 
   channel_send_connect_msg (chn);
   return mst;
@@ -404,10 +456,21 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
  * @param keep_active  FIXME
  */
 void
-GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst)
+GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst,
+                         int keep_active,
+                         GNUNET_ContinuationCallback stop_cb,
+                         void *stop_cls)
 {
-  GNUNET_CLIENT_MANAGER_disconnect (mst->chn.client, GNUNET_YES);
-  GNUNET_free (mst);
+  struct GNUNET_PSYC_Channel *chn = &mst->chn;
+
+  /* FIXME: send msg to service */
+
+  chn->is_disconnecting = GNUNET_YES;
+  chn->disconnect_cb = stop_cb;
+  chn->disconnect_cls = stop_cls;
+
+  GNUNET_CLIENT_MANAGER_disconnect (mst->chn.client, GNUNET_YES,
+                                    &master_cleanup, mst);
 }
 
 
@@ -439,7 +502,7 @@ 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_MessageHeader *join_resp)
+                           const struct GNUNET_PSYC_Message *join_resp)
 {
   struct GNUNET_PSYC_Channel *chn = &jh->mst->chn;
   struct GNUNET_PSYC_JoinDecisionMessage *dcsn;
@@ -461,6 +524,7 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
     memcpy (&dcsn[1], join_resp, join_resp_size);
 
   GNUNET_CLIENT_MANAGER_transmit (chn->client, &dcsn->header);
+  GNUNET_free (jh);
   return GNUNET_OK;
 }
 
@@ -576,15 +640,19 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
                         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_MessageHeader *join_msg)
+                        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 = ntohs (join_msg->header.size);
   struct SlaveJoinRequest *req
-    = GNUNET_malloc (sizeof (*req) + relay_count * sizeof (*relays));
+    = GNUNET_malloc (sizeof (*req) + relay_size + join_msg_size);
   req->header.size = htons (sizeof (*req)
                             + relay_count * sizeof (*relays));
   req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN);
@@ -592,7 +660,12 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   req->slave_key = *slave_key;
   req->origin = *origin;
   req->relay_count = htonl (relay_count);
-  memcpy (&req[1], relays, relay_count * sizeof (*relays));
+
+  if (0 < relay_size)
+    memcpy (&req[1], relays, relay_size);
+
+  if (0 < join_msg_size)
+    memcpy ((char *) &req[1] + relay_size, join_msg, join_msg_size);
 
   chn->connect_msg = (struct GNUNET_MessageHeader *) req;
   chn->cfg = cfg;
@@ -605,7 +678,7 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   chn->client = GNUNET_CLIENT_MANAGER_connect (cfg, "psyc", slave_handlers);
   GNUNET_CLIENT_MANAGER_set_user_context_ (chn->client, slv, sizeof (*chn));
 
-  chn->recv = GNUNET_PSYC_receive_create (message_cb, message_cb, cls);
+  chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls);
   chn->tmit = GNUNET_PSYC_transmit_create (chn->client);
 
   channel_send_connect_msg (chn);
@@ -622,10 +695,21 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
  * @param slave Slave handle.
  */
 void
-GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv)
+GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv,
+                        int keep_active,
+                        GNUNET_ContinuationCallback part_cb,
+                        void *part_cls)
 {
-  GNUNET_CLIENT_MANAGER_disconnect (slv->chn.client, GNUNET_YES);
-  GNUNET_free (slv);
+  struct GNUNET_PSYC_Channel *chn = &slv->chn;
+
+  /* FIXME: send msg to service */
+
+  chn->is_disconnecting = GNUNET_YES;
+  chn->disconnect_cb = part_cb;
+  chn->disconnect_cls = part_cls;
+
+  GNUNET_CLIENT_MANAGER_disconnect (slv->chn.client, GNUNET_YES,
+                                    &slave_cleanup, slv);
 }
 
 
@@ -796,6 +880,7 @@ GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *channel,
                                 uint64_t start_message_id,
                                 uint64_t end_message_id,
                                 GNUNET_PSYC_MessageCallback message_cb,
+                                GNUNET_PSYC_MessagePartCallback message_part_cb,
                                 GNUNET_PSYC_FinishCallback finish_cb,
                                 void *cls)
 {
diff --git a/src/psyc/psyc_util_lib.c b/src/psyc/psyc_util_lib.c
index 8ab0bdad2..6177976de 100644
--- a/src/psyc/psyc_util_lib.c
+++ b/src/psyc/psyc_util_lib.c
@@ -76,7 +76,11 @@ struct GNUNET_PSYC_TransmitHandle
    *
    */
   const char *mod_value;
-  size_t mod_value_size;
+
+  /**
+   * Number of bytes remaining to be transmitted from the current modifier value.
+   */
+  uint32_t mod_value_remaining;
 
   /**
    * State of the current message being received from client.
@@ -104,14 +108,14 @@ struct GNUNET_PSYC_TransmitHandle
 struct GNUNET_PSYC_ReceiveHandle
 {
   /**
-   * Message part callback.
+   * Message callback.
    */
   GNUNET_PSYC_MessageCallback message_cb;
 
   /**
-   * Message part callback for historic message.
+   * Message part callback.
    */
-  GNUNET_PSYC_MessageCallback hist_message_cb;
+  GNUNET_PSYC_MessagePartCallback message_part_cb;
 
   /**
    * Closure for the callbacks.
@@ -149,6 +153,7 @@ struct GNUNET_PSYC_ReceiveHandle
   uint32_t mod_value_size;
 };
 
+
 /**** Messages ****/
 
 
@@ -167,7 +172,7 @@ struct GNUNET_PSYC_ReceiveHandle
  * @return Message header with size information,
  *         followed by the message parts.
  */
-struct GNUNET_MessageHeader *
+struct GNUNET_PSYC_Message *
 GNUNET_PSYC_message_create (const char *method_name,
                             const struct GNUNET_ENV_Environment *env,
                             const void *data,
@@ -188,7 +193,7 @@ GNUNET_PSYC_message_create (const char *method_name,
     }
   }
 
-  struct GNUNET_MessageHeader *msg;
+  struct GNUNET_PSYC_Message *msg;
   uint16_t method_name_size = strlen (method_name) + 1;
   if (method_name_size == 1)
     return NULL;
@@ -199,12 +204,13 @@ GNUNET_PSYC_message_create (const char *method_name,
     + ((0 < data_size) ? sizeof (*pmsg) + data_size : 0)/* data */
     + sizeof (*pmsg);                                   /* end of message */
   msg = GNUNET_malloc (msg_size);
-  msg->size = htons (msg_size);
-  msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+  msg->header.size = htons (msg_size);
+  msg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE); /* FIXME */
 
   pmeth = (struct GNUNET_PSYC_MessageMethod *) &msg[1];
+  pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD);
   pmeth->header.size = htons (sizeof (*pmeth) + method_name_size);
-  memcpy (pmeth, method_name, method_name_size);
+  memcpy (&pmeth[1], method_name, method_name_size);
 
   uint16_t p = sizeof (*msg) + sizeof (*pmeth) + method_name_size;
   if (NULL != env)
@@ -215,7 +221,7 @@ GNUNET_PSYC_message_create (const char *method_name,
       uint16_t mod_name_size = strlen (mod->name) + 1;
       pmod = (struct GNUNET_PSYC_MessageModifier *) ((char *) msg + p);
       pmod->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER);
-      pmod->header.size = sizeof (*pmod) + mod_name_size + 1 + mod->value_size;
+      pmod->header.size = sizeof (*pmod) + mod_name_size + mod->value_size;
       p += pmod->header.size;
       pmod->header.size = htons (pmod->header.size);
 
@@ -241,6 +247,7 @@ GNUNET_PSYC_message_create (const char *method_name,
   pmsg->size = htons (sizeof (*pmsg));
   pmsg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END);
 
+  GNUNET_assert (p + sizeof (*pmsg) == msg_size);
   return msg;
 }
 
@@ -276,8 +283,8 @@ GNUNET_PSYC_log_message (enum GNUNET_ErrorType kind,
     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],
-                size - sizeof (*mod) - name_size - 1,
-                ((char *) &mod[1]) + name_size + 1);
+                size - sizeof (*mod) - name_size,
+                ((char *) &mod[1]) + name_size);
     break;
   }
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
@@ -331,7 +338,7 @@ transmit_queue_insert (struct GNUNET_PSYC_TransmitHandle *tmit,
   uint16_t size = (NULL != msg) ? ntohs (msg->size) : 0;
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Queueing message of type %u and size %u (end: %u)).\n",
+       "Queueing message part of type %u and size %u (end: %u)).\n",
        ntohs (msg->type), size, end);
 
   if (NULL != tmit->msg)
@@ -396,6 +403,9 @@ transmit_data (struct GNUNET_PSYC_TransmitHandle *tmit)
   msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA);
 
   int notify_ret = tmit->notify_data (tmit->notify_data_cls, &data_size, &msg[1]);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "transmit_data (ret: %d, size: %u): %.*s\n",
+       notify_ret, data_size, data_size, &msg[1]);
   switch (notify_ret)
   {
   case GNUNET_NO:
@@ -463,9 +473,15 @@ transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
     msg->size = sizeof (struct GNUNET_PSYC_MessageModifier);
     notify_ret = tmit->notify_mod (tmit->notify_mod_cls, &data_size, &mod[1],
                                    &mod->oper, &mod->value_size);
-    mod->name_size = strnlen ((char *) &mod[1], data_size);
+
+    mod->name_size = strnlen ((char *) &mod[1], data_size) + 1;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "transmit_mod (ret: %d, size: %u + %u): %.*s\n",
+         notify_ret, mod->name_size, mod->value_size, data_size, &mod[1]);
     if (mod->name_size < data_size)
     {
+      tmit->mod_value_remaining
+        = mod->value_size - (data_size - mod->name_size);
       mod->value_size = htonl (mod->value_size);
       mod->name_size = htons (mod->name_size);
     }
@@ -483,6 +499,10 @@ transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
     msg->size = sizeof (struct GNUNET_MessageHeader);
     notify_ret = tmit->notify_mod (tmit->notify_mod_cls,
                                    &data_size, &msg[1], NULL, NULL);
+    tmit->mod_value_remaining -= data_size;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "transmit_mod (ret: %d, size: %u): %.*s\n",
+         notify_ret, data_size, data_size, &msg[1]);
     break;
   }
   default:
@@ -497,26 +517,19 @@ transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
       tmit->paused = GNUNET_YES;
       return;
     }
-    tmit->state = GNUNET_PSYC_MESSAGE_STATE_MOD_CONT;
+    tmit->state
+      = (0 == tmit->mod_value_remaining)
+      ? GNUNET_PSYC_MESSAGE_STATE_MODIFIER
+      : GNUNET_PSYC_MESSAGE_STATE_MOD_CONT;
     break;
 
-  case GNUNET_YES:
-    if (0 == data_size)
-    {
-      /* End of modifiers. */
-      tmit->state = GNUNET_PSYC_MESSAGE_STATE_DATA;
-      if (0 == tmit->acks_pending)
-        transmit_data (tmit);
-
-      return;
-    }
-    tmit->state = GNUNET_PSYC_MESSAGE_STATE_MODIFIER;
+  case GNUNET_YES: /* End of modifiers. */
+    GNUNET_assert (0 == tmit->mod_value_remaining);
     break;
 
   default:
     LOG (GNUNET_ERROR_TYPE_ERROR,
-         "TransmitNotifyModifier callback returned error "
-         "when requesting a modifier.\n");
+         "TransmitNotifyModifier callback returned with error.\n");
 
     tmit->state = GNUNET_PSYC_MESSAGE_STATE_CANCEL;
     msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL);
@@ -533,7 +546,16 @@ transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
     transmit_queue_insert (tmit, msg, GNUNET_NO);
   }
 
-  transmit_mod (tmit);
+  if (GNUNET_YES == notify_ret)
+  {
+    tmit->state = GNUNET_PSYC_MESSAGE_STATE_DATA;
+    if (0 == tmit->acks_pending)
+      transmit_data (tmit);
+  }
+  else
+  {
+    transmit_mod (tmit);
+  }
 }
 
 
@@ -547,9 +569,10 @@ transmit_notify_env (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
   size_t value_size = 0;
   const char *value = NULL;
 
-  if (NULL != oper && NULL != tmit->mod)
+  if (NULL != oper)
   { /* New modifier */
-    tmit->mod = tmit->mod->next;
+    if (NULL != tmit->mod)
+      tmit->mod = tmit->mod->next;
     if (NULL == tmit->mod)
     { /* No more modifiers, continue with data */
       *data_size = 0;
@@ -559,30 +582,29 @@ transmit_notify_env (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
     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);
+    name_size = strlen (tmit->mod->name) + 1;
 
-    if (name_size + 1 + tmit->mod->value_size <= *data_size)
+    if (name_size + tmit->mod->value_size <= *data_size)
     {
-      *data_size = name_size + 1 + tmit->mod->value_size;
+      *data_size = name_size + 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;
+      value_size = *data_size - name_size;
       tmit->mod_value = tmit->mod->value + value_size;
     }
 
     memcpy (data, tmit->mod->name, name_size);
-    ((char *)data)[name_size] = '\0';
-    memcpy ((char *)data + name_size + 1, tmit->mod->value, value_size);
+    memcpy ((char *)data + name_size, tmit->mod->value, value_size);
+    return GNUNET_NO;
   }
-  else if (NULL != tmit->mod_value && 0 < tmit->mod_value_size)
+  else
   { /* Modifier continuation */
+    GNUNET_assert (NULL != tmit->mod_value && 0 < tmit->mod_value_remaining);
     value = tmit->mod_value;
-    if (tmit->mod_value_size <= *data_size)
+    if (tmit->mod_value_remaining <= *data_size)
     {
-      value_size = tmit->mod_value_size;
+      value_size = tmit->mod_value_remaining;
       tmit->mod_value = NULL;
     }
     else
@@ -590,7 +612,6 @@ transmit_notify_env (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
       value_size = *data_size;
       tmit->mod_value += value_size;
     }
-    tmit->mod_value_size -= value_size;
 
     if (*data_size < value_size)
     {
@@ -603,9 +624,8 @@ transmit_notify_env (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
 
     *data_size = value_size;
     memcpy (data, value, value_size);
+    return (tmit->mod_value = NULL) ? GNUNET_YES : GNUNET_NO;
   }
-
-  return 0 == tmit->mod_value_size ? GNUNET_YES : GNUNET_NO;
 }
 
 
@@ -663,10 +683,16 @@ GNUNET_PSYC_transmit_message (struct GNUNET_PSYC_TransmitHandle *tmit,
   {
     tmit->notify_mod = &transmit_notify_env;
     tmit->notify_mod_cls = tmit;
-    tmit->mod
-      = (NULL != env)
-      ? GNUNET_ENV_environment_head (env)
-      : NULL;
+    if (NULL != env)
+    {
+      struct GNUNET_ENV_Modifier mod = {};
+      mod.next = GNUNET_ENV_environment_head (env);
+      tmit->mod = &mod;
+    }
+    else
+    {
+      tmit->mod = NULL;
+    }
   }
 
   transmit_mod (tmit);
@@ -762,12 +788,12 @@ GNUNET_PSYC_transmit_got_ack (struct GNUNET_PSYC_TransmitHandle *tmit)
  */
 struct GNUNET_PSYC_ReceiveHandle *
 GNUNET_PSYC_receive_create (GNUNET_PSYC_MessageCallback message_cb,
-                            GNUNET_PSYC_MessageCallback hist_message_cb,
+                            GNUNET_PSYC_MessagePartCallback message_part_cb,
                             void *cb_cls)
 {
   struct GNUNET_PSYC_ReceiveHandle *recv = GNUNET_malloc (sizeof (*recv));
   recv->message_cb = message_cb;
-  recv->hist_message_cb = hist_message_cb;
+  recv->message_part_cb = message_part_cb;
   recv->cb_cls = cb_cls;
   return recv;
 }
@@ -800,13 +826,11 @@ GNUNET_PSYC_receive_reset (struct GNUNET_PSYC_ReceiveHandle *recv)
 static void
 recv_error (struct GNUNET_PSYC_ReceiveHandle *recv)
 {
-  GNUNET_PSYC_MessageCallback message_cb
-    = recv->flags & GNUNET_PSYC_MESSAGE_HISTORIC
-    ? recv->hist_message_cb
-    : recv->message_cb;
+  if (NULL != recv->message_part_cb)
+    recv->message_part_cb (recv->cb_cls, recv->message_id, 0, recv->flags, NULL);
 
-  if (NULL != message_cb)
-    message_cb (recv->cb_cls, recv->message_id, recv->flags, NULL);
+  if (NULL != recv->message_cb)
+    recv->message_cb (recv->cb_cls, recv->message_id, recv->flags, NULL);
 
   GNUNET_PSYC_receive_reset (recv);
 }
@@ -827,6 +851,7 @@ GNUNET_PSYC_receive_message (struct GNUNET_PSYC_ReceiveHandle *recv,
 {
   uint16_t size = ntohs (msg->header.size);
   uint32_t flags = ntohl (msg->flags);
+  uint64_t message_id;
 
   GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG,
                            (struct GNUNET_MessageHeader *) msg);
@@ -858,6 +883,7 @@ GNUNET_PSYC_receive_message (struct GNUNET_PSYC_ReceiveHandle *recv,
     recv_error (recv);
     return GNUNET_SYSERR;
   }
+  message_id = recv->message_id;
 
   uint16_t pos = 0, psize = 0, ptype, size_eq, size_min;
 
@@ -964,10 +990,10 @@ GNUNET_PSYC_receive_message (struct GNUNET_PSYC_ReceiveHandle *recv,
 
       uint16_t name_size = ntohs (mod->name_size);
       recv->mod_value_size_expected = ntohl (mod->value_size);
-      recv->mod_value_size = psize - sizeof (*mod) - name_size - 1;
+      recv->mod_value_size = psize - sizeof (*mod) - name_size;
 
-      if (psize < sizeof (*mod) + name_size + 1
-          || '\0' != *((char *) &mod[1] + name_size)
+      if (psize < sizeof (*mod) + name_size
+          || '\0' != *((char *) &mod[1] + name_size - 1)
           || recv->mod_value_size_expected < recv->mod_value_size)
       {
         LOG (GNUNET_ERROR_TYPE_WARNING, "Dropping malformed modifier.\n");
@@ -1018,13 +1044,9 @@ GNUNET_PSYC_receive_message (struct GNUNET_PSYC_ReceiveHandle *recv,
     }
     }
 
-    GNUNET_PSYC_MessageCallback message_cb
-      = recv->flags & GNUNET_PSYC_MESSAGE_HISTORIC
-      ? recv->hist_message_cb
-      : recv->message_cb;
-
-    if (NULL != message_cb)
-      message_cb (recv->cb_cls, recv->message_id, recv->flags, pmsg);
+    if (NULL != recv->message_part_cb)
+      recv->message_part_cb (recv->cb_cls, recv->message_id, 0, // FIXME: data_offset
+                             recv->flags, pmsg);
 
     switch (ptype)
     {
@@ -1034,6 +1056,9 @@ GNUNET_PSYC_receive_message (struct GNUNET_PSYC_ReceiveHandle *recv,
       break;
     }
   }
+
+  if (NULL != recv->message_cb)
+    recv->message_cb (recv->cb_cls, message_id, flags, msg);
   return GNUNET_OK;
 }
 
@@ -1063,6 +1088,7 @@ GNUNET_PSYC_receive_check_parts (uint16_t data_size, const char *data,
   for (pos = 0; pos < data_size; pos += psize, parts++)
   {
     pmsg = (const struct GNUNET_MessageHeader *) (data + pos);
+    GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, pmsg);
     psize = ntohs (pmsg->size);
     ptype = ntohs (pmsg->type);
     if (0 == parts && NULL != first_ptype)
@@ -1084,3 +1110,111 @@ GNUNET_PSYC_receive_check_parts (uint16_t data_size, const char *data,
   }
   return parts;
 }
+
+
+struct ParseMessageClosure
+{
+  struct GNUNET_ENV_Environment *env;
+  const char **method_name;
+  const void **data;
+  uint16_t *data_size;
+  enum GNUNET_PSYC_MessageState msg_state;
+};
+
+
+static void
+parse_message_part_cb (void *cls, uint64_t message_id, uint64_t data_offset,
+                       uint32_t flags, const struct GNUNET_MessageHeader *msg)
+{
+  struct ParseMessageClosure *pmc = cls;
+  if (NULL == msg)
+  {
+    pmc->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
+    return;
+  }
+
+  switch (ntohs (msg->type))
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+  {
+    struct GNUNET_PSYC_MessageMethod *
+      pmeth = (struct GNUNET_PSYC_MessageMethod *) msg;
+    *pmc->method_name = (const char *) &pmeth[1];
+    pmc->msg_state = GNUNET_PSYC_MESSAGE_STATE_METHOD;
+    break;
+  }
+
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+  {
+    struct GNUNET_PSYC_MessageModifier *
+      pmod = (struct GNUNET_PSYC_MessageModifier *) msg;
+
+    const char *name = (const char *) &pmod[1];
+    const void *value = name + pmod->name_size;
+    GNUNET_ENV_environment_add (pmc->env, pmod->oper, name, value,
+                                pmod->value_size);
+    pmc->msg_state = GNUNET_PSYC_MESSAGE_STATE_MODIFIER;
+    break;
+  }
+
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+    *pmc->data = &msg[1];
+    *pmc->data_size = ntohs (msg->size) - sizeof (*msg);
+    pmc->msg_state = GNUNET_PSYC_MESSAGE_STATE_DATA;
+    break;
+
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+    pmc->msg_state = GNUNET_PSYC_MESSAGE_STATE_END;
+    break;
+
+  default:
+    pmc->msg_state = GNUNET_PSYC_MESSAGE_STATE_ERROR;
+  }
+}
+
+
+/**
+ * Parse PSYC message.
+ *
+ * @param msg
+ *        The PSYC message to parse.
+ * @param[out] method_name
+ *        Pointer to the method name inside @a pmsg.
+ * @param env
+ *        The environment for the message with a list of modifiers.
+ * @param[out] data
+ *        Pointer to data inside @a pmsg.
+ * @param[out] data_size
+ *        Size of @data is written here.
+ *
+ * @return #GNUNET_OK on success,
+ *         #GNUNET_SYSERR on parse error.
+ */
+int
+GNUNET_PSYC_message_parse (const struct GNUNET_PSYC_Message *msg,
+                           const char **method_name,
+                           struct GNUNET_ENV_Environment *env,
+                           const void **data,
+                           uint16_t *data_size)
+{
+  struct ParseMessageClosure cls;
+  cls.env = env;
+  cls.method_name = method_name;
+  cls.data = data;
+  cls.data_size = data_size;
+
+  uint16_t msg_size = ntohs (msg->header.size);
+  struct GNUNET_PSYC_MessageHeader *
+    pmsg = GNUNET_malloc (sizeof (*pmsg) + msg_size - sizeof (*msg));
+  memcpy (&pmsg[1], &msg[1], msg_size - sizeof (*msg));
+
+  struct GNUNET_PSYC_ReceiveHandle *
+    recv = GNUNET_PSYC_receive_create (NULL, &parse_message_part_cb, &cls);
+  GNUNET_PSYC_receive_message (recv, pmsg);
+  GNUNET_PSYC_receive_destroy (recv);
+  GNUNET_free (pmsg);
+
+  return (GNUNET_PSYC_MESSAGE_STATE_END == cls.msg_state)
+    ? GNUNET_OK
+    : GNUNET_SYSERR;
+}
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 0077bc9b7..495e3be47 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -62,8 +62,6 @@ static struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key;
 static struct GNUNET_CRYPTO_EddsaPublicKey channel_pub_key;
 static struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key;
 
-struct GNUNET_PSYC_MasterTransmitHandle *mth;
-
 struct TransmitClosure
 {
   struct GNUNET_PSYC_MasterTransmitHandle *mst_tmit;
@@ -95,6 +93,28 @@ static void
 master_transmit ();
 
 
+void master_stopped (void *cls)
+{
+  if (NULL != tmit)
+  {
+    GNUNET_ENV_environment_destroy (tmit->env);
+    GNUNET_free (tmit);
+    tmit = NULL;
+  }
+  GNUNET_SCHEDULER_shutdown ();
+}
+
+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.
  */
@@ -103,21 +123,11 @@ cleanup ()
 {
   if (NULL != slv)
   {
-    GNUNET_PSYC_slave_part (slv);
+    GNUNET_PSYC_slave_part (slv, GNUNET_NO, &slave_parted, NULL);
     slv = NULL;
   }
-  if (NULL != mst)
-  {
-    GNUNET_PSYC_master_stop (mst);
-    mst = NULL;
-  }
-  if (NULL != tmit)
-  {
-    GNUNET_ENV_environment_destroy (tmit->env);
-    GNUNET_free (tmit);
-    tmit = NULL;
-  }
-  GNUNET_SCHEDULER_shutdown ();
+  else
+    slave_parted (NULL);
 }
 
 
@@ -171,7 +181,20 @@ end ()
 
 static void
 master_message_cb (void *cls, uint64_t message_id, uint32_t flags,
-                   const struct GNUNET_MessageHeader *msg)
+                   const struct GNUNET_PSYC_MessageHeader *msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Master got PSYC message fragment of size %u "
+              "belonging to message ID %llu with flags %x\n",
+              ntohs (msg->header.size), message_id, flags);
+  // FIXME
+}
+
+
+static void
+master_message_part_cb (void *cls, uint64_t message_id,
+                        uint64_t data_offset, uint32_t flags,
+                        const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
   {
@@ -215,7 +238,20 @@ master_message_cb (void *cls, uint64_t message_id, uint32_t flags,
 
 static void
 slave_message_cb (void *cls, uint64_t message_id, uint32_t flags,
-                  const struct GNUNET_MessageHeader *msg)
+                  const struct GNUNET_PSYC_MessageHeader *msg)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Slave got PSYC message fragment of size %u "
+              "belonging to message ID %llu with flags %x\n",
+              ntohs (msg->header.size), message_id, flags);
+  // FIXME
+}
+
+
+static void
+slave_message_part_cb (void *cls, uint64_t message_id,
+                       uint64_t data_offset, uint32_t flags,
+                       const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
   {
@@ -371,7 +407,7 @@ tmit_notify_mod (void *cls, uint16_t *data_size, void *data, uint8_t *oper,
     memcpy (data, value, value_size);
   }
 
-  return 0 == tmit->mod_value_size ? GNUNET_YES : GNUNET_NO;
+  return GNUNET_NO;
 }
 
 
@@ -380,8 +416,10 @@ slave_join ();
 
 
 static void
-join_decision_cb (void *cls, int is_admitted,
-                  const struct GNUNET_PSYC_MessageHeader *join_msg)
+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_WARNING,
               "Slave got join decision: %d\n", is_admitted);
@@ -415,8 +453,10 @@ join_decision_cb (void *cls, int is_admitted,
 
 
 static void
-join_request_cb (void *cls, const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
-                 const struct GNUNET_PSYC_MessageHeader *msg,
+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;
@@ -450,11 +490,11 @@ slave_join ()
                               "_foo", "bar baz", 7);
   GNUNET_ENV_environment_add (env, GNUNET_ENV_OP_ASSIGN,
                               "_foo_bar", "foo bar baz", 11);
-  struct GNUNET_MessageHeader *
+  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, &origin,
-                                0, NULL, &slave_message_cb,
+  slv = GNUNET_PSYC_slave_join (cfg, &channel_pub_key, slave_key, &origin, 0, NULL,
+                                &slave_message_cb, &slave_message_part_cb,
                                 &slave_connect_cb, &join_decision_cb, NULL,
                                 join_msg);
   GNUNET_ENV_environment_destroy (env);
@@ -551,7 +591,8 @@ run (void *cls,
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Starting master.\n");
   mst = GNUNET_PSYC_master_start (cfg, channel_key, GNUNET_PSYC_CHANNEL_PRIVATE,
                                   &master_start_cb, &join_request_cb,
-                                  &master_message_cb, NULL);
+                                  &master_message_cb, &master_message_part_cb,
+                                  NULL);
 }