psyc, multicast: join decision, tests

4 files changed, 351 insertions, 162 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 2b9ee7135..cef89b828 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -387,6 +387,11 @@ struct Slave
   struct GNUNET_MessageHeader *join_req;
 
   /**
+   * Join decision received from multicast.
+   */
+  struct SlaveJoinDecision *join_dcsn;
+
+  /**
    * Maximum request ID for this channel.
    */
   uint64_t max_request_id;
@@ -397,6 +402,10 @@ static inline void
 transmit_message (struct Channel *ch);
 
 
+static uint64_t
+message_queue_drop (struct Channel *ch);
+
+
 /**
  * Task run during shutdown.
  *
@@ -413,7 +422,7 @@ shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   }
   if (NULL != stats)
   {
-    GNUNET_STATISTICS_destroy (stats, GNUNET_NO);
+    GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
     stats = NULL;
   }
 }
@@ -471,7 +480,8 @@ cleanup_slave (struct Slave *slv)
 static void
 cleanup_channel (struct Channel *ch)
 {
-  /* FIXME: fragment_cache_clear */
+  message_queue_drop (ch);
+  GNUNET_CONTAINER_multihashmap_remove_all (recv_cache, &ch->pub_key_hash);
 
   if (NULL != ch->store_op)
     GNUNET_PSYCSTORE_operation_cancel (ch->store_op);
@@ -570,7 +580,7 @@ struct JoinMemTestClosure
 
 
 /**
- * Membership test result callback used for join requests.m
+ * Membership test result callback used for join requests.
  */
 static void
 join_mem_test_cb (void *cls, int64_t result, const char *err_msg)
@@ -585,8 +595,7 @@ 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);
-    msg_to_clients (jcls->ch,
-                    (struct GNUNET_MessageHeader *) jcls->master_join_req);
+    msg_to_clients (jcls->ch, &jcls->master_join_req->header);
   }
   else
   {
@@ -602,9 +611,10 @@ join_mem_test_cb (void *cls, int64_t result, const char *err_msg)
  * Incoming join request from multicast.
  */
 static void
-join_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
-         const struct GNUNET_MessageHeader *join_msg,
-         struct GNUNET_MULTICAST_JoinHandle *jh)
+mcast_join_request_cb (void *cls,
+                       const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
+                       const struct GNUNET_MessageHeader *join_msg,
+                       struct GNUNET_MULTICAST_JoinHandle *jh)
 {
   struct Channel *ch = cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%p Got join request.\n", ch);
@@ -643,21 +653,58 @@ join_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
 }
 
 
+/**
+ * Join decision received from multicast.
+ */
+static void
+mcast_join_decision_cb (void *cls, int is_admitted,
+                        const struct GNUNET_PeerIdentity *peer,
+                        uint16_t relay_count,
+                        const struct GNUNET_PeerIdentity *relays,
+                        const struct GNUNET_MessageHeader *join_resp)
+{
+  struct Slave *slv = cls;
+  struct Channel *ch = &slv->ch;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Got join decision: %d\n", slv, is_admitted);
+
+  uint16_t join_resp_size = (NULL != join_resp) ? ntohs (join_resp->size) : 0;
+  struct SlaveJoinDecision *
+    dcsn = slv->join_dcsn = GNUNET_malloc (sizeof (*dcsn) + join_resp_size);
+  dcsn->header.size = htons (sizeof (*dcsn) + join_resp_size);
+  dcsn->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION);
+  dcsn->is_admitted = htonl (is_admitted);
+  if (0 < join_resp_size)
+    memcpy (&dcsn[1], join_resp, join_resp_size);
+
+  msg_to_clients (ch, &dcsn->header);
+
+  if (GNUNET_YES == is_admitted)
+  {
+    ch->ready = GNUNET_YES;
+  }
+  else
+  {
+    slv->member = NULL;
+  }
+}
+
+
 static void
-membership_test_cb (void *cls,
-                    const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
-                    uint64_t message_id, uint64_t group_generation,
-                    struct GNUNET_MULTICAST_MembershipTestHandle *mth)
+mcast_membership_test_cb (void *cls,
+                          const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
+                          uint64_t message_id, uint64_t group_generation,
+                          struct GNUNET_MULTICAST_MembershipTestHandle *mth)
 {
 
 }
 
 
 static void
-replay_fragment_cb (void *cls,
-                    const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
-                    uint64_t fragment_id, uint64_t flags,
-                    struct GNUNET_MULTICAST_ReplayHandle *rh)
+mcast_replay_fragment_cb (void *cls,
+                          const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
+                          uint64_t fragment_id, uint64_t flags,
+                          struct GNUNET_MULTICAST_ReplayHandle *rh)
 
 {
 
@@ -665,12 +712,12 @@ replay_fragment_cb (void *cls,
 
 
 static void
-replay_message_cb (void *cls,
-                   const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
-                   uint64_t message_id,
-                   uint64_t fragment_offset,
-                   uint64_t flags,
-                   struct GNUNET_MULTICAST_ReplayHandle *rh)
+mcast_replay_message_cb (void *cls,
+                         const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
+                         uint64_t message_id,
+                         uint64_t fragment_offset,
+                         uint64_t flags,
+                         struct GNUNET_MULTICAST_ReplayHandle *rh)
 {
 
 }
@@ -744,7 +791,7 @@ mmsg_to_clients (struct Channel *ch,
   pmsg->message_id = mmsg->message_id;
 
   memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
-  msg_to_clients (ch, (const struct GNUNET_MessageHeader *) pmsg);
+  msg_to_clients (ch, &pmsg->header);
   GNUNET_free (pmsg);
 }
 
@@ -988,6 +1035,7 @@ fragment_queue_run (struct Channel *ch, uint64_t msg_id,
  *   has already been delivered to the client.
  *
  * @param ch  Channel.
+ *
  * @return Number of messages removed from queue and sent to client.
  */
 static uint64_t
@@ -1061,6 +1109,43 @@ message_queue_run (struct Channel *ch)
 
 
 /**
+ * Drop message queue of a channel.
+ *
+ * Remove all messages in queue without sending it to clients.
+ *
+ * @param ch  Channel.
+ *
+ * @return Number of messages removed from queue.
+ */
+static uint64_t
+message_queue_drop (struct Channel *ch)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "%p Dropping message queue.\n", ch);
+  uint64_t n = 0;
+  uint64_t msg_id;
+  while (GNUNET_YES == GNUNET_CONTAINER_heap_peek2 (ch->recv_msgs, NULL,
+                                                    &msg_id))
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "%p Dropping message %" PRIu64 " from queue.\n", ch, msg_id);
+    struct GNUNET_HashCode msg_id_hash;
+    hash_key_from_hll (&msg_id_hash, msg_id);
+
+    struct FragmentQueue *
+      fragq = GNUNET_CONTAINER_multihashmap_get (ch->recv_frags, &msg_id_hash);
+
+    fragment_queue_run (ch, msg_id, fragq, GNUNET_YES);
+    GNUNET_CONTAINER_heap_remove_root (ch->recv_msgs);
+    n++;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Removed %" PRIu64 " messages from queue.\n", ch, n);
+  return n;
+}
+
+
+/**
  * Handle incoming message from multicast.
  *
  * @param ch   Channel.
@@ -1069,7 +1154,7 @@ message_queue_run (struct Channel *ch)
  * @return #GNUNET_OK or #GNUNET_SYSERR
  */
 static int
-handle_multicast_message (struct Channel *ch,
+client_multicast_message (struct Channel *ch,
                           const struct GNUNET_MULTICAST_MessageHeader *mmsg)
 {
   GNUNET_PSYCSTORE_fragment_store (store, &ch->pub_key, mmsg, 0, NULL, NULL);
@@ -1106,7 +1191,7 @@ handle_multicast_message (struct Channel *ch,
  * Store it using PSYCstore and send it to the client of the channel.
  */
 static void
-message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
+mcast_message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 {
   struct Channel *ch = cls;
   uint16_t type = ntohs (msg->type);
@@ -1120,7 +1205,7 @@ message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
   {
   case GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE:
   {
-    handle_multicast_message (ch, (const struct
+    client_multicast_message (ch, (const struct
                                    GNUNET_MULTICAST_MessageHeader *) msg);
     break;
   }
@@ -1141,9 +1226,10 @@ message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
  * @param flags         Request flags.
  */
 static void
-request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
-            const struct GNUNET_MessageHeader *msg,
-            enum GNUNET_MULTICAST_MessageFlags flags)
+mcast_request_cb (void *cls,
+                  const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
+                  const struct GNUNET_MessageHeader *msg,
+                  enum GNUNET_MULTICAST_MessageFlags flags)
 {
   struct Master *mst = cls;
   struct Channel *ch = &mst->ch;
@@ -1183,7 +1269,7 @@ request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
     pmsg->flags = htonl (GNUNET_PSYC_MESSAGE_REQUEST);
 
     memcpy (&pmsg[1], &req[1], size - sizeof (*req));
-    msg_to_clients (ch, (const struct GNUNET_MessageHeader *) pmsg);
+    msg_to_clients (ch, &pmsg->header);
     GNUNET_free (pmsg);
     break;
   }
@@ -1221,11 +1307,13 @@ master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
     ch->max_state_message_id = max_state_message_id;
     mst->max_group_generation = max_group_generation;
     mst->origin
-      = GNUNET_MULTICAST_origin_start (cfg, &mst->priv_key,
-                                       max_fragment_id,
-                                       join_cb, membership_test_cb,
-                                       replay_fragment_cb, replay_message_cb,
-                                       request_cb, message_cb, ch);
+      = GNUNET_MULTICAST_origin_start (cfg, &mst->priv_key, max_fragment_id,
+                                       &mcast_join_request_cb,
+                                       &mcast_membership_test_cb,
+                                       &mcast_replay_fragment_cb,
+                                       &mcast_replay_message_cb,
+                                       &mcast_request_cb,
+                                       &mcast_message_cb, ch);
     ch->ready = GNUNET_YES;
   }
   else
@@ -1266,11 +1354,13 @@ slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
       = GNUNET_MULTICAST_member_join (cfg, &ch->pub_key, &slv->priv_key,
                                       &slv->origin,
                                       slv->relay_count, slv->relays,
-                                      slv->join_req, join_cb,
-                                      membership_test_cb,
-                                      replay_fragment_cb, replay_message_cb,
-                                      message_cb, ch);
-    ch->ready = GNUNET_YES;
+                                      slv->join_req,
+                                      &mcast_join_request_cb,
+                                      &mcast_join_decision_cb,
+                                      &mcast_membership_test_cb,
+                                      &mcast_replay_fragment_cb,
+                                      &mcast_replay_message_cb,
+                                      &mcast_message_cb, ch);
   }
   else
   {
@@ -1297,7 +1387,7 @@ channel_init (struct Channel *ch)
  * Handle a connecting client starting a channel master.
  */
 static void
-handle_master_start (void *cls, struct GNUNET_SERVER_Client *client,
+client_master_start (void *cls, struct GNUNET_SERVER_Client *client,
                      const struct GNUNET_MessageHeader *msg)
 {
   const struct MasterStartRequest *req
@@ -1363,7 +1453,7 @@ handle_master_start (void *cls, struct GNUNET_SERVER_Client *client,
  * Handle a connecting client joining as a channel slave.
  */
 static void
-handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
+client_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
                    const struct GNUNET_MessageHeader *msg)
 {
   const struct SlaveJoinRequest *req
@@ -1389,6 +1479,8 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
   {
     slv = GNUNET_new (struct Slave);
     slv->priv_key = req->slave_key;
+    slv->pub_key = slv_pub_key;
+    slv->pub_key_hash = slv_pub_key_hash;
     slv->origin = req->origin;
     slv->relay_count = ntohl (req->relay_count);
     if (0 < slv->relay_count)
@@ -1411,10 +1503,10 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
     if (NULL == ch_slv)
     {
       ch_slv = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
-      GNUNET_CONTAINER_multihashmap_put (channel_slaves, &pub_key_hash, ch_slv,
+      GNUNET_CONTAINER_multihashmap_put (channel_slaves, &ch->pub_key_hash, ch_slv,
                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
     }
-    GNUNET_CONTAINER_multihashmap_put (ch_slv, &slv_pub_key_hash, ch,
+    GNUNET_CONTAINER_multihashmap_put (ch_slv, &slv->pub_key_hash, ch,
                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
     GNUNET_CONTAINER_multihashmap_put (slaves, &ch->pub_key_hash, ch,
                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
@@ -1434,6 +1526,29 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
     GNUNET_SERVER_notification_context_add (nc, client);
     GNUNET_SERVER_notification_context_unicast (nc, client, &res.header,
                                                 GNUNET_NO);
+
+    if (NULL == slv->member)
+    {
+      slv->member
+        = GNUNET_MULTICAST_member_join (cfg, &ch->pub_key, &slv->priv_key,
+                                        &slv->origin,
+                                        slv->relay_count, slv->relays,
+                                        slv->join_req,
+                                        &mcast_join_request_cb,
+                                        &mcast_join_decision_cb,
+                                        &mcast_membership_test_cb,
+                                        &mcast_replay_fragment_cb,
+                                        &mcast_replay_message_cb,
+                                        &mcast_message_cb, ch);
+
+    }
+    else if (NULL != slv->join_dcsn)
+    {
+      GNUNET_SERVER_notification_context_add (nc, client);
+      GNUNET_SERVER_notification_context_unicast (nc, client,
+                                                  &slv->join_dcsn->header,
+                                                  GNUNET_NO);
+    }
   }
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
@@ -1451,7 +1566,7 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
 
 struct JoinDecisionClosure
 {
-  uint8_t is_admitted;
+  int32_t is_admitted;
   struct GNUNET_MessageHeader *msg;
 };
 
@@ -1460,8 +1575,8 @@ struct JoinDecisionClosure
  * Iterator callback for responding to join requests of a slave.
  */
 static int
-join_decision_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
-                  void *jh)
+send_join_decision_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
+                       void *jh)
 {
   struct JoinDecisionClosure *jcls = cls;
   // FIXME: add relays
@@ -1474,7 +1589,7 @@ join_decision_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
  * Join decision from client.
  */
 static void
-handle_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
+client_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
                       const struct GNUNET_MessageHeader *msg)
 {
   struct Channel *
@@ -1484,7 +1599,7 @@ handle_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
 
   struct MasterJoinDecision *dcsn = (struct MasterJoinDecision *) msg;
   struct JoinDecisionClosure jcls;
-  jcls.is_admitted = dcsn->is_admitted;
+  jcls.is_admitted = ntohl (dcsn->is_admitted);
   jcls.msg
     = (sizeof (*dcsn) + sizeof (struct GNUNET_PSYC_MessageHeader)
        <= ntohs (msg->size))
@@ -1494,8 +1609,17 @@ handle_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
   struct GNUNET_HashCode slave_key_hash;
   GNUNET_CRYPTO_hash (&dcsn->slave_key, sizeof (dcsn->slave_key),
                       &slave_key_hash);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p Got join decision (%d) from client for channel %s..\n",
+              mst, jcls.is_admitted, GNUNET_h2s (&ch->pub_key_hash));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%p ..and slave %s.\n",
+              mst, GNUNET_h2s (&slave_key_hash));
+
   GNUNET_CONTAINER_multihashmap_get_multiple (mst->join_reqs, &slave_key_hash,
-                                              &join_decision_cb, &jcls);
+                                              &send_join_decision_cb, &jcls);
+  GNUNET_CONTAINER_multihashmap_remove_all (mst->join_reqs, &slave_key_hash);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 }
 
@@ -1758,10 +1882,10 @@ transmit_cancel (struct Channel *ch, struct GNUNET_SERVER_Client *client)
 
 
 /**
- * Incoming message from a client.
+ * Incoming message from a master or slave client.
  */
 static void
-handle_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
+client_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
                      const struct GNUNET_MessageHeader *msg)
 {
   struct Channel *
@@ -1775,8 +1899,7 @@ handle_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
   if (GNUNET_YES != ch->ready)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                "%p Dropping message from client, channel is not ready yet.\n",
-                ch);
+                "%p Channel is not ready, dropping message from client.\n", ch);
     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
     return;
   }
@@ -1784,7 +1907,7 @@ handle_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", ch);
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "%p Message payload too large.\n", ch);
     GNUNET_break (0);
     transmit_cancel (ch, client);
     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
@@ -1817,7 +1940,7 @@ handle_psyc_message (void *cls, struct GNUNET_SERVER_Client *client,
  * Client requests to add a slave to the membership database.
  */
 static void
-handle_slave_add (void *cls, struct GNUNET_SERVER_Client *client,
+client_slave_add (void *cls, struct GNUNET_SERVER_Client *client,
                   const struct GNUNET_MessageHeader *msg)
 {
 
@@ -1828,7 +1951,7 @@ handle_slave_add (void *cls, struct GNUNET_SERVER_Client *client,
  * Client requests to remove a slave from the membership database.
  */
 static void
-handle_slave_remove (void *cls, struct GNUNET_SERVER_Client *client,
+client_slave_remove (void *cls, struct GNUNET_SERVER_Client *client,
                      const struct GNUNET_MessageHeader *msg)
 {
 
@@ -1839,7 +1962,7 @@ handle_slave_remove (void *cls, struct GNUNET_SERVER_Client *client,
  * Client requests channel history from PSYCstore.
  */
 static void
-handle_story_request (void *cls, struct GNUNET_SERVER_Client *client,
+client_story_request (void *cls, struct GNUNET_SERVER_Client *client,
                       const struct GNUNET_MessageHeader *msg)
 {
 
@@ -1850,7 +1973,7 @@ handle_story_request (void *cls, struct GNUNET_SERVER_Client *client,
  * Client requests best matching state variable from PSYCstore.
  */
 static void
-handle_state_get (void *cls, struct GNUNET_SERVER_Client *client,
+client_state_get (void *cls, struct GNUNET_SERVER_Client *client,
                   const struct GNUNET_MessageHeader *msg)
 {
 
@@ -1861,7 +1984,7 @@ handle_state_get (void *cls, struct GNUNET_SERVER_Client *client,
  * Client requests state variables with a given prefix from PSYCstore.
  */
 static void
-handle_state_get_prefix (void *cls, struct GNUNET_SERVER_Client *client,
+client_state_get_prefix (void *cls, struct GNUNET_SERVER_Client *client,
                          const struct GNUNET_MessageHeader *msg)
 {
 
@@ -1880,31 +2003,31 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
      const struct GNUNET_CONFIGURATION_Handle *c)
 {
   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
-    { &handle_master_start, NULL,
+    { &client_master_start, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_MASTER_START, 0 },
 
-    { &handle_slave_join, NULL,
+    { &client_slave_join, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN, 0 },
 
-    { &handle_join_decision, NULL,
+    { &client_join_decision, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION, 0 },
 
-    { &handle_psyc_message, NULL,
+    { &client_psyc_message, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_MESSAGE, 0 },
 
-    { &handle_slave_add, NULL,
+    { &client_slave_add, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_ADD, 0 },
 
-    { &handle_slave_remove, NULL,
+    { &client_slave_remove, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM, 0 },
 
-    { &handle_story_request, NULL,
+    { &client_story_request, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_STORY_REQUEST, 0 },
 
-    { &handle_state_get, NULL,
+    { &client_state_get, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_STATE_GET, 0 },
 
-    { &handle_state_get_prefix, NULL,
+    { &client_state_get_prefix, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX, 0 }
   };
 
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index 47f2a0122..66c8de898 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -250,19 +250,36 @@ struct MasterJoinDecision
   struct GNUNET_MessageHeader header;
 
   /**
+   * #GNUNET_YES if the slave was admitted.
+   */
+  int32_t is_admitted;
+
+  /**
    * Public key of the joining slave.
    */
   struct GNUNET_CRYPTO_EddsaPublicKey slave_key;
 
+  /* Followed by struct GNUNET_MessageHeader join_response */
+};
+
+
+struct SlaveJoinDecision
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION
+   */
+  struct GNUNET_MessageHeader header;
+
   /**
    * #GNUNET_YES if the slave was admitted.
    */
-  uint8_t is_admitted;
+  int32_t is_admitted;
 
   /* Followed by struct GNUNET_MessageHeader join_response */
 };
 
 
+
 GNUNET_NETWORK_STRUCT_END
 
 #endif
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index ee49a584f..7ec9d21b7 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -198,9 +198,9 @@ struct GNUNET_PSYC_Master
   GNUNET_PSYC_MasterStartCallback start_cb;
 
   /**
-   * Join handler callback.
+   * Join request callback.
    */
-  GNUNET_PSYC_JoinCallback join_cb;
+  GNUNET_PSYC_JoinRequestCallback join_req_cb;
 };
 
 
@@ -211,14 +211,16 @@ struct GNUNET_PSYC_Slave
 {
   struct GNUNET_PSYC_Channel ch;
 
-  GNUNET_PSYC_SlaveJoinCallback join_cb;
+  GNUNET_PSYC_SlaveConnectCallback connect_cb;
+
+  GNUNET_PSYC_JoinDecisionCallback join_dcsn_cb;
 };
 
 
 /**
  * Handle that identifies a join request.
  *
- * Used to match calls to #GNUNET_PSYC_JoinCallback to the
+ * Used to match calls to #GNUNET_PSYC_JoinRequestCallback to the
  * corresponding calls to GNUNET_PSYC_join_decision().
  */
 struct GNUNET_PSYC_JoinHandle
@@ -922,7 +924,22 @@ handle_psyc_join_request (struct GNUNET_PSYC_Master *mst,
   jh->mst = mst;
   jh->slave_key = req->slave_key;
 
-  mst->join_cb (mst->ch.cb_cls, &req->slave_key, msg, jh);
+  if (NULL != mst->join_req_cb)
+    mst->join_req_cb (mst->ch.cb_cls, &req->slave_key, msg, jh);
+}
+
+
+static void
+handle_psyc_join_decision (struct GNUNET_PSYC_Slave *slv,
+                           const struct SlaveJoinDecision *dcsn)
+{
+  struct GNUNET_PSYC_MessageHeader *msg = NULL;
+  if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*msg))
+    msg = (struct GNUNET_PSYC_MessageHeader *) &dcsn[1];
+
+  struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh));
+  if (NULL != slv->join_dcsn_cb)
+    slv->join_dcsn_cb (slv->ch.cb_cls, ntohl (dcsn->is_admitted), msg);
 }
 
 
@@ -971,6 +988,9 @@ message_handler (void *cls,
   case GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST:
     size_min = sizeof (struct MasterJoinRequest);
     break;
+  case GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION:
+    size_min = sizeof (struct SlaveJoinDecision);
+    break;
   default:
     GNUNET_break_op (0);
     return;
@@ -995,8 +1015,8 @@ message_handler (void *cls,
   case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
   {
     struct CountersResult *cres = (struct CountersResult *) msg;
-    if (NULL != slv->join_cb)
-      slv->join_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id));
+    if (NULL != slv->connect_cb)
+      slv->connect_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id));
     break;
   }
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK:
@@ -1013,6 +1033,11 @@ message_handler (void *cls,
     handle_psyc_join_request ((struct GNUNET_PSYC_Master *) ch,
                               (const struct MasterJoinRequest *) msg);
     break;
+
+  case GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION:
+    handle_psyc_join_decision ((struct GNUNET_PSYC_Slave *) ch,
+                               (const struct SlaveJoinDecision *) msg);
+    break;
   }
 
   if (NULL != ch->client)
@@ -1175,20 +1200,20 @@ disconnect (void *c)
  * or part messages, the respective methods must call other PSYC functions to
  * inform PSYC about the meaning of the respective events.
  *
- * @param cfg Configuration to use (to connect to PSYC service).
- * @param channel_key ECC key that will be used to sign messages for this
+ * @param cfg  Configuration to use (to connect to PSYC service).
+ * @param channel_key  ECC key that will be used to sign messages for this
  *        PSYC session. The public key is used to identify the PSYC channel.
  *        Note that end-users will usually not use the private key directly, but
  *        rather look it up in GNS for places managed by other users, or select
  *        a file with the private key(s) when setting up their own channels
  *        FIXME: we'll likely want to use NOT the p521 curve here, but a cheaper
  *        one in the future.
- * @param policy Channel policy specifying join and history restrictions.
+ * @param policy  Channel policy specifying join and history restrictions.
  *        Used to automate join decisions.
- * @param message_cb Function to invoke on message parts received from slaves.
- * @param join_cb Function to invoke when a peer wants to join.
- * @param master_started_cb Function to invoke after the channel master started.
- * @param cls Closure for @a master_started_cb and @a join_cb.
+ * @param message_cb  Function to invoke on message parts received from slaves.
+ * @param join_request_cb  Function to invoke when a slave wants to join.
+ * @param master_start_cb  Function to invoke after the channel master started.
+ * @param cls  Closure for @a method and @a join_cb.
  *
  * @return Handle for the channel master, NULL on error.
  */
@@ -1196,9 +1221,9 @@ struct GNUNET_PSYC_Master *
 GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
                           const struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key,
                           enum GNUNET_PSYC_Policy policy,
+                          GNUNET_PSYC_MasterStartCallback start_cb,
+                          GNUNET_PSYC_JoinRequestCallback join_request_cb,
                           GNUNET_PSYC_MessageCallback message_cb,
-                          GNUNET_PSYC_JoinCallback join_cb,
-                          GNUNET_PSYC_MasterStartCallback master_started_cb,
                           void *cls)
 {
   struct GNUNET_PSYC_Master *mst = GNUNET_malloc (sizeof (*mst));
@@ -1210,8 +1235,8 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
   req->channel_key = *channel_key;
   req->policy = policy;
 
-  mst->start_cb = master_started_cb;
-  mst->join_cb = join_cb;
+  mst->start_cb = start_cb;
+  mst->join_req_cb = join_request_cb;
   ch->message_cb = message_cb;
   ch->cb_cls = cls;
   ch->cfg = cfg;
@@ -1244,8 +1269,9 @@ GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master)
  * #GNUNET_PSYC_JoinCallback.
  *
  * @param jh Join request handle.
- * @param is_admitted #GNUNET_YES if joining is approved,
- *        #GNUNET_NO if it is disapproved.
+ * @param is_admitted  #GNUNET_YES    if the join is approved,
+ *                     #GNUNET_NO     if it is disapproved,
+ *                     #GNUNET_SYSERR if we cannot answer the request.
  * @param relay_count Number of relays given.
  * @param relays Array of suggested peers that might be useful relays to use
  *        when joining the multicast group (essentially a list of peers that
@@ -1254,48 +1280,42 @@ GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master)
  *        be the multicast origin) is a good candidate for building the
  *        multicast tree.  Note that it is unnecessary to specify our own
  *        peer identity in this array.
- * @param method_name Method name for the message transmitted with the response.
- * @param env Environment containing transient variables for the message, or NULL.
- * @param data Data of the message.
- * @param data_size Size of @a data.
+ * @param join_resp  Application-dependent join response message.
+ *
+ * @return #GNUNET_OK on success,
+ *         #GNUNET_SYSERR if the message is too large.
  */
-void
+int
 GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
                            int is_admitted,
                            uint32_t relay_count,
                            const struct GNUNET_PeerIdentity *relays,
-                           const char *method_name,
-                           const struct GNUNET_ENV_Environment *env,
-                           const void *data,
-                           size_t data_size)
+                           const struct GNUNET_PSYC_MessageHeader *join_resp)
 {
   struct GNUNET_PSYC_Channel *ch = &jh->mst->ch;
-
   struct MasterJoinDecision *dcsn;
-  struct GNUNET_PSYC_MessageHeader *pmsg = NULL;
-  uint16_t pmsg_size = 0;
-/* FIXME:
-  sizeof (*pmsg)
-    + sizeof (struct GNUNET_PSYC_MessageMethod)
-    + vars_size
-    + sizeof (struct GNUNET_MessageHeader) + data_size
-    + sizeof (struct GNUNET_MessageHeader);
-*/
+  uint16_t join_resp_size
+    = (NULL != join_resp) ? ntohs (join_resp->header.size) : 0;
   uint16_t relay_size = relay_count * sizeof (*relays);
-  struct MessageQueue *
-    mq = GNUNET_malloc (sizeof (*mq) + sizeof (*dcsn) + relay_size + pmsg_size);
+
+  if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD
+      < sizeof (*dcsn) + relay_size + join_resp_size)
+    return GNUNET_SYSERR;
+
+  struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + sizeof (*dcsn)
+                                           + relay_size + join_resp_size);
   dcsn = (struct MasterJoinDecision *) &mq[1];
-  dcsn->header.size = htons (sizeof (*dcsn) + relay_size + pmsg_size);
+  dcsn->header.size = htons (sizeof (*dcsn) + relay_size + join_resp_size);
   dcsn->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION);
-  dcsn->is_admitted = (GNUNET_YES == is_admitted) ? GNUNET_YES : GNUNET_NO;
+  dcsn->is_admitted = htonl (is_admitted);
   dcsn->slave_key = jh->slave_key;
 
-  /* FIXME: add message parts to pmsg */
-  if (0 < pmsg_size)
-    memcpy (&dcsn[1], pmsg, pmsg_size);
+  if (0 < join_resp_size)
+    memcpy (&dcsn[1], join_resp, join_resp_size);
 
   GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, mq);
   transmit_next (ch);
+  return GNUNET_OK;
 }
 
 
@@ -1359,24 +1379,27 @@ GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th)
  * notification on failure (as the channel may simply take days to approve,
  * and disapproval is simply being ignored).
  *
- * @param cfg Configuration to use.
- * @param channel_key ECC public key that identifies the channel we wish to join.
- * @param slave_key ECC private-public key pair that identifies the slave, and
+ * @param cfg  Configuration to use.
+ * @param channel_key  ECC public key that identifies the channel we wish to join.
+ * @param slave_key  ECC private-public key pair that identifies the slave, and
  *        used by multicast to sign the join request and subsequent unicast
  *        requests sent to the master.
- * @param origin Peer identity of the origin.
- * @param relay_count Number of peers in the @a relays array.
- * @param relays Peer identities of members of the multicast group, which serve
+ * @param origin  Peer identity of the origin.
+ * @param relay_count  Number of peers in the @a relays array.
+ * @param relays  Peer identities of members of the multicast group, which serve
  *        as relays and used to join the group at.
- * @param message_cb Function to invoke on message parts received from the
+ * @param message_cb  Function to invoke on message parts received from the
  *        channel, typically at least contains method handlers for @e join and
  *        @e part.
- * @param slave_joined_cb Function invoked once we have joined the channel.
- * @param cls Closure for @a message_cb and @a slave_joined_cb.
- * @param method_name Method name for the join request.
- * @param env Environment containing transient variables for the request, or NULL.
- * @param data Payload for the join message.
- * @param data_size Number of bytes in @a data.
+ * @param slave_connect_cb  Function invoked once we have connected to the
+ *        PSYC service.
+ * @param join_decision_cb  Function invoked once we have received a join
+ *        decision.
+ * @param cls  Closure for @a message_cb and @a slave_joined_cb.
+ * @param method_name  Method name for the join request.
+ * @param env  Environment containing transient variables for the request, or NULL.
+ * @param data  Payload for the join message.
+ * @param data_size  Number of bytes in @a data.
  *
  * @return Handle for the slave, NULL on error.
  */
@@ -1388,7 +1411,8 @@ 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_SlaveJoinCallback slave_joined_cb,
+                        GNUNET_PSYC_SlaveConnectCallback connect_cb,
+                        GNUNET_PSYC_JoinDecisionCallback join_decision_cb,
                         void *cls,
                         const char *method_name,
                         const struct GNUNET_ENV_Environment *env,
@@ -1408,7 +1432,8 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   req->relay_count = htonl (relay_count);
   memcpy (&req[1], relays, relay_count * sizeof (*relays));
 
-  slv->join_cb = slave_joined_cb;
+  slv->connect_cb = connect_cb;
+  slv->join_dcsn_cb = join_decision_cb;
   ch->message_cb = message_cb;
   ch->cb_cls = cls;
 
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 4195e464b..8b4aad773 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -35,9 +35,9 @@
 #include "gnunet_env_lib.h"
 #include "gnunet_psyc_service.h"
 
-#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10)
+#define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
 
-#define DEBUG_SERVICE 0
+#define DEBUG_SERVICE 1
 
 
 /**
@@ -79,6 +79,7 @@ struct TransmitClosure
 
 struct TransmitClosure *tmit;
 
+static int join_req_count;
 
 enum
 {
@@ -167,8 +168,8 @@ end ()
 
 
 static void
-master_message (void *cls, uint64_t message_id, uint32_t flags,
-                const struct GNUNET_MessageHeader *msg)
+master_message_cb (void *cls, uint64_t message_id, uint32_t flags,
+                   const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
   {
@@ -211,8 +212,8 @@ master_message (void *cls, uint64_t message_id, uint32_t flags,
 
 
 static void
-slave_message (void *cls, uint64_t message_id, uint32_t flags,
-               const struct GNUNET_MessageHeader *msg)
+slave_message_cb (void *cls, uint64_t message_id, uint32_t flags,
+                  const struct GNUNET_MessageHeader *msg)
 {
   if (NULL == msg)
   {
@@ -243,23 +244,6 @@ slave_message (void *cls, uint64_t message_id, uint32_t flags,
 
 
 static void
-join_request (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
-              const struct GNUNET_PSYC_MessageHeader *msg,
-              struct GNUNET_PSYC_JoinHandle *jh)
-{
-  struct GNUNET_HashCode slave_key_hash;
-  GNUNET_CRYPTO_hash (slave_key, sizeof (*slave_key), &slave_key_hash);
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Got join request from %s.\n",
-              GNUNET_h2s (&slave_key_hash));
-
-  GNUNET_PSYC_join_decision (jh, GNUNET_YES, 0, NULL, "_notice_join", NULL,
-                             "you're in", 9);
-  // FIXME: also test refusing entry
-}
-
-
-static void
 transmit_resume (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission resumed.\n");
@@ -392,9 +376,23 @@ tmit_notify_data (void *cls, uint16_t *data_size, void *data)
 
 
 static void
-slave_joined (void *cls, uint64_t max_message_id)
+slave_join ();
+
+
+static void
+join_decision_cb (void *cls, int is_admitted,
+                  const struct GNUNET_PSYC_MessageHeader *join_msg)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave joined: %lu\n", max_message_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Slave got join decision: %d\n", is_admitted);
+
+  if (GNUNET_YES != is_admitted)
+  { /* First join request is refused, retry. */
+    //GNUNET_assert (1 == join_req_count);
+    slave_join ();
+    return;
+  }
+
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave sending request to master.\n");
 
   test = TEST_SLAVE_TRANSMIT;
@@ -414,20 +412,46 @@ slave_joined (void *cls, uint64_t max_message_id)
                                   GNUNET_PSYC_SLAVE_TRANSMIT_NONE);
 }
 
+
+static void
+join_request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
+                 const struct GNUNET_PSYC_MessageHeader *msg,
+                 struct GNUNET_PSYC_JoinHandle *jh)
+{
+  struct GNUNET_HashCode slave_key_hash;
+  GNUNET_CRYPTO_hash (slave_key, sizeof (*slave_key), &slave_key_hash);
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Got join request from %s.\n",
+              GNUNET_h2s (&slave_key_hash));
+
+  /* Reject first request */
+  int is_admitted = (0 < join_req_count++) ? GNUNET_YES : GNUNET_NO;
+  GNUNET_PSYC_join_decision (jh, is_admitted, 0, NULL, NULL);
+}
+
+
+static void
+slave_connect_cb (void *cls, uint64_t max_message_id)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+              "Slave connected: %lu\n", max_message_id);
+}
+
+
 static void
 slave_join ()
 {
   GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Joining slave.\n");
 
-  struct GNUNET_PeerIdentity origin;
-  struct GNUNET_PeerIdentity relays[16];
+  struct GNUNET_PeerIdentity origin; // FIXME: this peer
   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, &slave_joined, NULL,
+                                0, NULL, &slave_message_cb,
+                                &slave_connect_cb, &join_decision_cb, NULL,
                                 "_request_join", env, "some data", 9);
   GNUNET_ENV_environment_destroy (env);
 }
@@ -485,7 +509,7 @@ master_transmit ()
 
 
 static void
-master_started (void *cls, uint64_t max_message_id)
+master_start_cb (void *cls, uint64_t max_message_id)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "Master started: %" PRIu64 "\n", max_message_id);
@@ -521,8 +545,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_message, &join_request,
-                                  &master_started, NULL);
+                                  &master_start_cb, &join_request_cb,
+                                  &master_message_cb, NULL);
 }