9 files changed, 125 insertions, 73 deletions

diff --git a/src/include/gnunet_psyc_service.h b/src/include/gnunet_psyc_service.h
index 1346e77c7..80404f837 100644
--- a/src/include/gnunet_psyc_service.h
+++ b/src/include/gnunet_psyc_service.h
@@ -537,11 +537,18 @@ typedef void
 /**
  * Method called from PSYC upon receiving part of a message.
  *
- * @param cls  Closure.
- * @param message_id  Sequence number of the message.
- * @param data_offset  Byte offset of data, only set if @a msg has a type
- *                     #GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA
- * @param flags  OR'ed GNUNET_PSYC_MessageFlags
+ * @param cls
+ *        Closure.
+ * @param slave_key
+ *        Public key of the slave sending the message.
+ *        Only set for channel master.
+ * @param message_id
+ *        Sequence number of the message.
+ * @param flags
+ *        OR'ed GNUNET_PSYC_MessageFlags
+ * @param data_offset
+ *        Byte offset of data, only set if @a msg has a type
+ *        #GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA
  * @param msg  Message part, one of the following types:
  * - #GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_HEADER
  * - #GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD
@@ -552,19 +559,24 @@ typedef void
  */
 typedef void
 (*GNUNET_PSYC_MessagePartCallback) (void *cls,
+                                    const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
                                     uint64_t message_id,
-                                    uint64_t data_offset,
                                     uint32_t flags,
+                                    uint64_t data_offset,
                                     const struct GNUNET_MessageHeader *msg);
 
 
 /**
  * Method called from PSYC upon receiving a join request.
  *
- * @param cls  Closure.
- * @param slave_key  Public key of the slave requesting join.
- * @param join_msg  Join message sent along with the request.
- * @param jh  Join handle to use with GNUNET_PSYC_join_decision()
+ * @param cls
+ *        Closure.
+ * @param slave_key
+ *        Public key of the slave requesting join.
+ * @param join_msg
+ *        Join message sent along with the request.
+ * @param jh
+ *        Join handle to use with GNUNET_PSYC_join_decision()
  */
 typedef void
 (*GNUNET_PSYC_JoinRequestCallback) (void *cls,
diff --git a/src/multicast/gnunet-service-multicast.c b/src/multicast/gnunet-service-multicast.c
index afd5102ec..a9b15cef1 100644
--- a/src/multicast/gnunet-service-multicast.c
+++ b/src/multicast/gnunet-service-multicast.c
@@ -788,9 +788,11 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "%p Client connected to group %s..\n",
               mem, GNUNET_h2s (&grp->pub_key_hash));
+  char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&mem->pub_key);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "%p ..as member %s.\n",
-              mem, GNUNET_h2s (&mem_pub_key_hash));
+              "%p ..as member %s (%s).\n",
+              mem, GNUNET_h2s (&mem->pub_key_hash), str);
+  GNUNET_free (str);
 
   GNUNET_SERVER_client_set_user_context (client, grp);
 
@@ -833,6 +835,7 @@ client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
     if (0 < join_msg_size)
       memcpy (&req[1], join_msg, join_msg_size);
 
+    req->member_key = mem->pub_key;
     req->purpose.size = htonl (msg_size
                                - sizeof (req->header)
                                - sizeof (req->reserved)
@@ -1000,6 +1003,7 @@ client_recv_multicast_request (void *cls, struct GNUNET_SERVER_Client *client,
   /* FIXME: yucky, should use separate message structs for P2P and CS! */
   out = (struct GNUNET_MULTICAST_RequestHeader *) GNUNET_copy_message (m);
 
+  out->member_key = mem->pub_key;
   out->fragment_id = GNUNET_ntohll (++mem->max_fragment_id);
   out->purpose.size = htonl (ntohs (out->header.size)
                              - sizeof (out->header)
diff --git a/src/multicast/multicast_api.c b/src/multicast/multicast_api.c
index ce663542d..5cd729203 100644
--- a/src/multicast/multicast_api.c
+++ b/src/multicast/multicast_api.c
@@ -267,7 +267,7 @@ group_recv_message (void *cls,
               "Calling message callback with a message of size %u.\n",
               ntohs (mmsg->header.size));
 
-  if (NULL != grp->message_cb)
+  if (GNUNET_YES != grp->is_disconnecting && NULL != grp->message_cb)
     grp->message_cb (grp->cb_cls, mmsg);
 }
 
@@ -848,7 +848,7 @@ GNUNET_MULTICAST_member_part (struct GNUNET_MULTICAST_Member *mem,
   grp->disconnect_cls = part_cls;
 
   GNUNET_CLIENT_MANAGER_disconnect (mem->grp.client, GNUNET_YES,
-                                    &member_cleanup, mem);
+                                    member_cleanup, mem);
 }
 
 
diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 2afc98040..e20b2280e 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -534,8 +534,12 @@ cleanup_slave (struct Slave *slv)
 static void
 cleanup_channel (struct Channel *chn)
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+              "%p Cleaning up channel %s. master? %u\n",
+              chn, GNUNET_h2s (&chn->pub_key_hash), chn->is_master);
   message_queue_drop (chn);
-  GNUNET_CONTAINER_multihashmap_remove_all (recv_cache, &chn->pub_key_hash);
+  GNUNET_CONTAINER_multihashmap_destroy (chn->recv_frags);
+  chn->recv_frags = NULL;
 
   if (NULL != chn->store_op)
   {
@@ -565,6 +569,7 @@ client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 
   struct Channel *
     chn = GNUNET_SERVER_client_get_user_context (client, struct Channel);
+  chn->is_disconnected = GNUNET_YES;
 
   if (NULL == chn)
   {
@@ -1046,6 +1051,7 @@ client_send_mcast_req (struct Master *mst,
   pmsg->message_id = req->request_id;
   pmsg->fragment_offset = req->fragment_offset;
   pmsg->flags = htonl (GNUNET_PSYC_MESSAGE_REQUEST);
+  pmsg->slave_key = req->member_key;
 
   memcpy (&pmsg[1], &req[1], size - sizeof (*req));
   client_send_msg (chn, &pmsg->header);
@@ -1229,7 +1235,7 @@ fragment_queue_run (struct Channel *chn, uint64_t msg_id,
   struct GNUNET_CONTAINER_MultiHashMap
     *chan_msgs = GNUNET_CONTAINER_multihashmap_get (recv_cache,
                                                     &chn->pub_key_hash);
-  GNUNET_assert (NULL != chan_msgs); // FIXME
+  GNUNET_assert (NULL != chan_msgs);
   uint64_t frag_id;
 
   while (GNUNET_YES == GNUNET_CONTAINER_heap_peek2 (fragq->fragments, NULL,
@@ -1382,7 +1388,7 @@ message_queue_run (struct Channel *chn)
       /* Check if there's a missing message before the current one */
       if (GNUNET_PSYC_STATE_NOT_MODIFIED == fragq->state_delta)
       {
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p state NOT modified\n");
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p state NOT modified\n", chn);
 
         if (!(fragq->flags & GNUNET_PSYC_MESSAGE_ORDER_ANY)
             && (chn->max_message_id != msg_id - 1
@@ -1399,7 +1405,7 @@ message_queue_run (struct Channel *chn)
       }
       else
       {
-        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p state modified\n");
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p state modified\n", chn);
         if (GNUNET_YES != fragq->state_is_modified)
         {
           if (msg_id - fragq->state_delta != chn->max_state_message_id)
@@ -1594,12 +1600,12 @@ store_recv_master_counters (void *cls, int result, uint64_t max_fragment_id,
     mst->max_group_generation = max_group_generation;
     mst->origin
       = GNUNET_MULTICAST_origin_start (cfg, &mst->priv_key, max_fragment_id,
-                                       &mcast_recv_join_request,
-                                       &mcast_recv_membership_test,
-                                       &mcast_recv_replay_fragment,
-                                       &mcast_recv_replay_message,
-                                       &mcast_recv_request,
-                                       &mcast_recv_message, chn);
+                                       mcast_recv_join_request,
+                                       mcast_recv_membership_test,
+                                       mcast_recv_replay_fragment,
+                                       mcast_recv_replay_message,
+                                       mcast_recv_request,
+                                       mcast_recv_message, chn);
     chn->is_ready = GNUNET_YES;
   }
   else
@@ -1641,12 +1647,12 @@ store_recv_slave_counters (void *cls, int result, uint64_t max_fragment_id,
                                       &slv->origin,
                                       slv->relay_count, slv->relays,
                                       &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);
+                                      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);
@@ -2004,7 +2010,7 @@ transmit_notify (void *cls, size_t *data_size, void *data)
   else if (GNUNET_YES == chn->is_disconnected)
   {
     /* FIXME: handle partial message (when still in_transmit) */
-    cleanup_channel (chn);
+    return GNUNET_SYSERR;
   }
   return ret;
 }
@@ -2113,13 +2119,17 @@ master_queue_message (struct Master *mst, struct TransmitMessage *tmit_msg,
     }
     else if (pmeth->flags & GNUNET_PSYC_MASTER_TRANSMIT_STATE_MODIFY)
     {
-      GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "%p master_queue_message: setting state_modify flag\n", mst);
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "%p master_queue_message: state_delta=%" PRIu64 "\n",
+                  mst, tmit_msg->id - mst->max_state_message_id);
       pmeth->state_delta = GNUNET_htonll (tmit_msg->id
                                           - mst->max_state_message_id);
+      mst->max_state_message_id = tmit_msg->id;
     }
     else
     {
-        GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "%p master_queue_message: setting state_not_modified flag\n", mst);
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "%p master_queue_message: state not modified\n", mst);
       pmeth->state_delta = GNUNET_htonll (GNUNET_PSYC_STATE_NOT_MODIFIED);
     }
 
@@ -2234,7 +2244,9 @@ client_recv_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
   uint16_t size = ntohs (msg->size);
   if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < size - sizeof (*msg))
   {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "%p Message payload too large.\n", chn);
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "%p Message payload too large: %u < %u.\n",
+                chn, GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD, size - sizeof (*msg));
     GNUNET_break (0);
     transmit_cancel (chn, client);
     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
@@ -2613,7 +2625,7 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
   masters = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
   slaves = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
   channel_slaves = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
-  recv_cache = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+  recv_cache = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
   nc = GNUNET_SERVER_notification_context_create (server, 1);
   GNUNET_SERVER_add_handlers (server, server_handlers);
   GNUNET_SERVER_disconnect_notify (server, &client_disconnect, NULL);
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 4e7979a4d..863eaab88 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -213,8 +213,9 @@ master_message_cb (void *cls, uint64_t message_id, uint32_t flags,
 
 
 void
-master_message_part_cb (void *cls, uint64_t message_id,
-                        uint64_t data_offset, uint32_t flags,
+master_message_part_cb (void *cls,
+                        const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
+                        uint64_t message_id, uint32_t flags, uint64_t data_offset,
                         const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
@@ -284,8 +285,9 @@ slave_message_cb (void *cls, uint64_t message_id, uint32_t flags,
 
 
 void
-slave_message_part_cb (void *cls, uint64_t message_id,
-                       uint64_t data_offset, uint32_t flags,
+slave_message_part_cb (void *cls,
+                       const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
+                       uint64_t message_id, uint32_t flags, uint64_t data_offset,
                        const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
@@ -453,9 +455,9 @@ slave_history_replay ()
   test = TEST_SLAVE_HISTORY_REPLAY;
   GNUNET_PSYC_channel_history_replay (slv_chn, 1, 1, "",
                                       GNUNET_PSYC_HISTORY_REPLAY_LOCAL,
-                                      &slave_message_cb,
-                                      &slave_message_part_cb,
-                                      &slave_history_replay_result, NULL);
+                                      slave_message_cb,
+                                      slave_message_part_cb,
+                                      slave_history_replay_result, NULL);
 }
 
 
@@ -481,9 +483,9 @@ master_history_replay ()
   test = TEST_MASTER_HISTORY_REPLAY;
   GNUNET_PSYC_channel_history_replay (mst_chn, 1, 1, "",
                                       GNUNET_PSYC_HISTORY_REPLAY_LOCAL,
-                                      &master_message_cb,
-                                      &master_message_part_cb,
-                                      &master_history_replay_result, NULL);
+                                      master_message_cb,
+                                      master_message_part_cb,
+                                      master_history_replay_result, NULL);
 }
 
 
diff --git a/src/psycstore/gnunet-service-psycstore.c b/src/psycstore/gnunet-service-psycstore.c
index 1f9de54f8..01eb5d707 100644
--- a/src/psycstore/gnunet-service-psycstore.c
+++ b/src/psycstore/gnunet-service-psycstore.c
@@ -508,8 +508,10 @@ struct StateModifyClosure
 
 
 static void
-recv_state_message_part (void *cls, uint64_t message_id, uint64_t data_offset,
-                         uint32_t flags, const struct GNUNET_MessageHeader *msg)
+recv_state_message_part (void *cls,
+                         const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
+                         uint64_t message_id, uint32_t flags, uint64_t data_offset,
+                         const struct GNUNET_MessageHeader *msg)
 {
   struct StateModifyClosure *scls = cls;
   uint16_t psize;
@@ -618,7 +620,7 @@ recv_state_fragment (void *cls, struct GNUNET_MULTICAST_MessageHeader *msg,
 
   if (NULL == scls->recv)
   {
-    scls->recv = GNUNET_PSYC_receive_create (NULL, &recv_state_message_part,
+    scls->recv = GNUNET_PSYC_receive_create (NULL, recv_state_message_part,
                                              scls);
   }
 
@@ -660,7 +662,7 @@ handle_state_modify (void *cls,
   {
     ret = db->message_get (db->cls, &req->channel_key,
                            message_id, message_id,
-                           &ret_frags, &recv_state_fragment, &scls);
+                           &ret_frags, recv_state_fragment, &scls);
     if (GNUNET_OK != ret)
     {
       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
diff --git a/src/psycstore/plugin_psycstore_sqlite.c b/src/psycstore/plugin_psycstore_sqlite.c
index 1bf14644b..8f9a3ef63 100644
--- a/src/psycstore/plugin_psycstore_sqlite.c
+++ b/src/psycstore/plugin_psycstore_sqlite.c
@@ -598,7 +598,7 @@ database_setup (struct Plugin *plugin)
                "SELECT name, value_current\n"
                "FROM state\n"
                "WHERE channel_id = (SELECT id FROM channels WHERE pub_key = ?)\n"
-               "      AND (name = ? OR name LIKE ?);",
+               "      AND (name = ? OR substr(name, 1, ?) = ? || '_');",
                &plugin->select_state_prefix);
 
   sql_prepare (plugin->dbh,
@@ -1763,16 +1763,12 @@ state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_
   int ret = GNUNET_SYSERR;
   sqlite3_stmt *stmt = plugin->select_state_prefix;
   size_t name_len = strlen (name);
-  char *name_prefix;
 
-  GNUNET_asprintf (&name_prefix,
-                   "%s_%%",
-                   name);
   if (SQLITE_OK != sqlite3_bind_blob (stmt, 1, channel_key,
                                       sizeof (*channel_key), SQLITE_STATIC)
       || SQLITE_OK != sqlite3_bind_text (stmt, 2, name, name_len, SQLITE_STATIC)
-      || SQLITE_OK != sqlite3_bind_text (stmt, 3, name_prefix, name_len + 2,
-                                         SQLITE_STATIC))
+      || SQLITE_OK != sqlite3_bind_int (stmt, 3, name_len + 1)
+      || SQLITE_OK != sqlite3_bind_text (stmt, 4, name, name_len, SQLITE_STATIC))
   {
     LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite3_bind");
@@ -1808,7 +1804,6 @@ state_get_prefix (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *channel_
     LOG_SQLITE (plugin, GNUNET_ERROR_TYPE_ERROR | GNUNET_ERROR_TYPE_BULK,
                 "sqlite3_reset");
   }
-  GNUNET_free (name_prefix);
   return ret;
 }
 
diff --git a/src/psycstore/psyc_util_lib.c b/src/psycstore/psyc_util_lib.c
index 7b13ac57f..e45bcafb3 100644
--- a/src/psycstore/psyc_util_lib.c
+++ b/src/psycstore/psyc_util_lib.c
@@ -401,12 +401,17 @@ transmit_queue_insert (struct GNUNET_PSYC_TransmitHandle *tmit,
 static void
 transmit_data (struct GNUNET_PSYC_TransmitHandle *tmit)
 {
-  uint16_t data_size = GNUNET_PSYC_DATA_MAX_PAYLOAD;
+  int notify_ret = GNUNET_YES;
+  uint16_t data_size = 0;
   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 = tmit->notify_data (tmit->notify_data_cls, &data_size, &msg[1]);
+  if (NULL != tmit->notify_data)
+  {
+    data_size = GNUNET_PSYC_DATA_MAX_PAYLOAD;
+    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]);
@@ -461,10 +466,11 @@ transmit_data (struct GNUNET_PSYC_TransmitHandle *tmit)
 static void
 transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
 {
-  uint16_t max_data_size, data_size;
+  uint16_t max_data_size = 0;
+  uint16_t data_size = 0;
   char data[GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD] = "";
   struct GNUNET_MessageHeader *msg = (struct GNUNET_MessageHeader *) data;
-  int notify_ret;
+  int notify_ret = GNUNET_YES;
 
   switch (tmit->state)
   {
@@ -472,11 +478,16 @@ transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
   {
     struct GNUNET_PSYC_MessageModifier *mod
       = (struct GNUNET_PSYC_MessageModifier *) msg;
-    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 = tmit->notify_mod (tmit->notify_mod_cls, &data_size, &mod[1],
-                                   &mod->oper, &mod->value_size);
+
+    if (NULL != tmit->notify_mod)
+    {
+      max_data_size = GNUNET_PSYC_MODIFIER_MAX_PAYLOAD;
+      data_size = max_data_size;
+      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) + 1;
     LOG (GNUNET_ERROR_TYPE_DEBUG,
@@ -498,11 +509,16 @@ transmit_mod (struct GNUNET_PSYC_TransmitHandle *tmit)
   }
   case GNUNET_PSYC_MESSAGE_STATE_MOD_CONT:
   {
-    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 = tmit->notify_mod (tmit->notify_mod_cls,
-                                   &data_size, &msg[1], NULL, NULL);
+
+    if (NULL != tmit->notify_mod)
+    {
+      max_data_size = GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD;
+      data_size = max_data_size;
+      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",
@@ -847,7 +863,8 @@ static void
 recv_error (struct GNUNET_PSYC_ReceiveHandle *recv)
 {
   if (NULL != recv->message_part_cb)
-    recv->message_part_cb (recv->cb_cls, recv->message_id, 0, recv->flags, NULL);
+    recv->message_part_cb (recv->cb_cls, NULL, recv->message_id, recv->flags,
+                           0, NULL);
 
   if (NULL != recv->message_cb)
     recv->message_cb (recv->cb_cls, recv->message_id, recv->flags, NULL);
@@ -1066,8 +1083,10 @@ GNUNET_PSYC_receive_message (struct GNUNET_PSYC_ReceiveHandle *recv,
     }
 
     if (NULL != recv->message_part_cb)
-      recv->message_part_cb (recv->cb_cls, recv->message_id, 0, // FIXME: data_offset
-                             recv->flags, pmsg);
+      recv->message_part_cb (recv->cb_cls, &recv->slave_key,
+                             recv->message_id, recv->flags,
+                             0, // FIXME: data_offset
+                             pmsg);
 
     switch (ptype)
     {
@@ -1144,8 +1163,10 @@ struct ParseMessageClosure
 
 
 static void
-parse_message_part_cb (void *cls, uint64_t message_id, uint64_t data_offset,
-                       uint32_t flags, const struct GNUNET_MessageHeader *msg)
+parse_message_part_cb (void *cls,
+                       const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_key,
+                       uint64_t message_id, uint32_t flags, uint64_t data_offset,
+                       const struct GNUNET_MessageHeader *msg)
 {
   struct ParseMessageClosure *pmc = cls;
   if (NULL == msg)
@@ -1230,7 +1251,7 @@ GNUNET_PSYC_message_parse (const struct GNUNET_PSYC_Message *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);
+    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);
diff --git a/src/util/client_manager.c b/src/util/client_manager.c
index d6e5f155b..98cf5b1ad 100644
--- a/src/util/client_manager.c
+++ b/src/util/client_manager.c
@@ -191,6 +191,10 @@ recv_message (void *cls, const struct GNUNET_MessageHeader *msg)
     size = ntohs (msg->size);
     /* FIXME: decrease reconnect_delay gradually after a successful reconnection */
   }
+  else /* disconnected */
+  {
+    mgr->client_tmit = NULL;
+  }
 
   size_t i = 0;
   while (NULL != mgr->handlers[i].callback)