psyc: ipc messages

4 files changed, 774 insertions, 213 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index a3b1b8f82..628c39900 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -71,9 +71,9 @@ struct TransmitMessage
   char *buf;
   uint16_t size;
   /**
-   * enum GNUNET_PSYC_DataStatus
+   * enum MessageState
    */
-  uint8_t status;
+  uint8_t state;
 };
 
 /**
@@ -87,12 +87,21 @@ struct Channel
   struct TransmitMessage *tmit_tail;
 
   GNUNET_SCHEDULER_TaskIdentifier tmit_task;
-  uint32_t tmit_mod_count;
-  uint32_t tmit_mod_recvd;
+
   /**
-   * enum GNUNET_PSYC_DataStatus
+   * Expected value size for the modifier being received from the PSYC service.
    */
-  uint8_t tmit_status;
+  uint32_t tmit_mod_value_size_expected;
+
+  /**
+   * Actual value size for the modifier being received from the PSYC service.
+   */
+  uint32_t tmit_mod_value_size;
+
+  /**
+   * enum MessageState
+   */
+  uint8_t tmit_state;
 
   uint8_t in_transmit;
   uint8_t is_master;
@@ -112,12 +121,27 @@ struct Master
   struct GNUNET_MULTICAST_Origin *origin;
   struct GNUNET_MULTICAST_OriginMessageHandle *tmit_handle;
 
+  /**
+   * Maximum message ID for this channel.
+   *
+   * Incremented before sending a message, thus the message_id in messages sent
+   * starts from 1.
+   */
   uint64_t max_message_id;
+
+  /**
+   * ID of the last message that contains any state operations.
+   * 0 if there is no such message.
+   */
   uint64_t max_state_message_id;
+
+  /**
+   * Maximum group generation for this channel.
+   */
   uint64_t max_group_generation;
 
   /**
-   * enum GNUNET_PSYC_Policy
+   * @see enum GNUNET_PSYC_Policy
    */
   uint32_t policy;
 };
@@ -196,6 +220,7 @@ client_cleanup (struct Channel *ch)
   GNUNET_free (ch);
 }
 
+
 /**
  * Called whenever a client is disconnected.
  * Frees our resources associated with that client.
@@ -234,7 +259,8 @@ client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
   }
 }
 
-void
+
+static void
 join_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
          const struct GNUNET_MessageHeader *join_req,
          struct GNUNET_MULTICAST_JoinHandle *jh)
@@ -242,7 +268,8 @@ join_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
 
 }
 
-void
+
+static void
 membership_test_cb (void *cls,
                     const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
                     uint64_t message_id, uint64_t group_generation,
@@ -251,16 +278,18 @@ membership_test_cb (void *cls,
 
 }
 
-void
+
+static void
 replay_fragment_cb (void *cls,
                     const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
                     uint64_t fragment_id, uint64_t flags,
                     struct GNUNET_MULTICAST_ReplayHandle *rh)
-{
 
+{
 }
 
-void
+
+static void
 replay_message_cb (void *cls,
                    const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
                    uint64_t message_id,
@@ -271,56 +300,30 @@ replay_message_cb (void *cls,
 
 }
 
-void
-request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
-            const struct GNUNET_MessageHeader *req,
-            enum GNUNET_MULTICAST_MessageFlags flags)
-{
-
-}
-
 
-void
+static void
 fragment_store_result (void *cls, int64_t result, const char *err_msg)
 {
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
               "fragment_store() returned %l (%s)\n", result, err_msg);
 }
 
+
 /**
- * Send PSYC messages in an incoming multicast message to a client.
+ * Iterator callback for sending a message to a client.
+ *
+ * @see message_cb()
  */
-int
-send_to_client (void *cls, const struct GNUNET_HashCode *ch_key_hash, void *chan)
+static int
+message_to_client (void *cls, const struct GNUNET_HashCode *chan_key_hash,
+                   void *chan)
 {
-  const struct GNUNET_MULTICAST_MessageHeader *msg = cls;
+  const struct GNUNET_MessageHeader *msg = cls;
   struct Channel *ch = chan;
 
-  uint16_t size = ntohs (msg->header.size);
-  uint16_t pos = 0;
-
-  while (sizeof (*msg) + pos < size)
-  {
-    const struct GNUNET_MessageHeader *pmsg
-      = (const struct GNUNET_MessageHeader *) ((char *) &msg[1] + pos);
-    uint16_t psize = ntohs (pmsg->size);
-    if (sizeof (*msg) + pos + psize > size)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                  "Ignoring message of type %u with invalid size. "
-                  "(%u + %u + %u > %u)\n", ntohs (pmsg->type),
-                  sizeof (*msg), pos, psize, size);
-      break;
-    }
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Sending message of type %u and size %u to client.\n",
-                ntohs (pmsg->type), psize);
+  GNUNET_SERVER_notification_context_add (nc, ch->client);
+  GNUNET_SERVER_notification_context_unicast (nc, ch->client, msg, GNUNET_NO);
 
-    GNUNET_SERVER_notification_context_add (nc, ch->client);
-    GNUNET_SERVER_notification_context_unicast (nc, ch->client, pmsg,
-                                                GNUNET_NO);
-    pos += psize;
-  }
   return GNUNET_YES;
 }
 
@@ -330,7 +333,7 @@ send_to_client (void *cls, const struct GNUNET_HashCode *ch_key_hash, void *chan
  *
  * Store it using PSYCstore and send it to all clients of the channel.
  */
-void
+static void
 message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 {
   uint16_t type = ntohs (msg->type);
@@ -344,34 +347,100 @@ message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
   struct Master *mst = cls;
   struct Slave *slv = cls;
 
-  struct GNUNET_CRYPTO_EddsaPublicKey *ch_key
+  /* const struct GNUNET_MULTICAST_MessageHeader *mmsg
+     = (const struct GNUNET_MULTICAST_MessageHeader *) msg; */
+  struct GNUNET_CRYPTO_EddsaPublicKey *chan_key
     = ch->is_master ? &mst->pub_key : &slv->chan_key;
-  struct GNUNET_HashCode *ch_key_hash
+  struct GNUNET_HashCode *chan_key_hash
     = ch->is_master ? &mst->pub_key_hash : &slv->chan_key_hash;
 
   switch (type)
   {
   case GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE:
-    GNUNET_PSYCSTORE_fragment_store (store, ch_key,
+  {
+    GNUNET_PSYCSTORE_fragment_store (store, chan_key,
                                      (const struct
                                       GNUNET_MULTICAST_MessageHeader *) msg,
                                      0, NULL, NULL);
-    GNUNET_CONTAINER_multihashmap_get_multiple (clients, ch_key_hash,
-                                                send_to_client, (void *) msg);
-    break;
 
+    uint16_t size = ntohs (msg->size);
+    uint16_t psize = 0;
+    uint16_t pos = 0;
+
+    for (pos = 0; sizeof (*msg) + pos < size; pos += psize)
+    {
+      const struct GNUNET_MessageHeader *pmsg
+        = (const struct GNUNET_MessageHeader *) ((char *) &msg[1] + pos);
+      uint16_t psize = ntohs (pmsg->size);
+      if (sizeof (*msg) + pos + psize > size)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "Message received from multicast contains invalid PSYC "
+                    "message. Not sending to clients.\n");
+        return;
+      }
+    }
+
+#if TODO
+    /* FIXME: apply modifiers to state in PSYCstore */
+    GNUNET_PSYCSTORE_state_modify (store, chan_key,
+                                   GNUNET_ntohll (mmsg->message_id),
+                                   meth->mod_count, mods,
+                                   rcb, rcb_cls);
+#endif
+
+    const struct GNUNET_MULTICAST_MessageHeader *mmsg
+      = (const struct GNUNET_MULTICAST_MessageHeader *) msg;
+    struct GNUNET_PSYC_MessageHeader *pmsg;
+
+    psize = sizeof (*pmsg) + size - sizeof (*mmsg);
+    pmsg = GNUNET_malloc (psize);
+    pmsg->header.size = htons (psize);
+    pmsg->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE);
+    pmsg->message_id = mmsg->message_id;
+
+    memcpy (&pmsg[1], &mmsg[1], size - sizeof (*mmsg));
+
+    GNUNET_CONTAINER_multihashmap_get_multiple (clients, chan_key_hash,
+                                                message_to_client,
+                                                (void *) pmsg);
+    GNUNET_free (pmsg);
+    break;
+  }
   default:
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                "Ignoring unknown message of type %u and size %u.\n",
+                "Discarding unknown message of type %u and size %u.\n",
                 type, size);
   }
 }
 
 
 /**
+ * Send a request received from multicast to a client.
+ */
+static int
+request_to_client (void *cls, const struct GNUNET_HashCode *chan_key_hash,
+                   void *chan)
+{
+  /* TODO */
+
+  return GNUNET_YES;
+}
+
+
+static void
+request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
+            const struct GNUNET_MessageHeader *req,
+            enum GNUNET_MULTICAST_MessageFlags flags)
+{
+
+}
+
+
+/**
  * Response from PSYCstore with the current counter values for a channel master.
  */
-void
+static void
 master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
                     uint64_t max_message_id, uint64_t max_group_generation,
                     uint64_t max_state_message_id)
@@ -513,20 +582,13 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
 static void
 send_transmit_ack (struct Channel *ch)
 {
-  struct TransmitAck *res = GNUNET_malloc (sizeof (*res));
-  res->header.size = htons (sizeof (*res));
-  res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK);
-
-  res->buf_avail = GNUNET_MULTICAST_FRAGMENT_MAX_SIZE;
-  struct TransmitMessage *tmit_msg = ch->tmit_tail;
-  if (NULL != tmit_msg && GNUNET_PSYC_DATA_CONT == tmit_msg->status)
-    res->buf_avail -= tmit_msg->size;
-  res->buf_avail = htons (res->buf_avail);
+  struct GNUNET_MessageHeader res;
+  res.size = htons (sizeof (res));
+  res.type = htons (GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK);
 
   GNUNET_SERVER_notification_context_add (nc, ch->client);
-  GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
+  GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res,
                                               GNUNET_NO);
-  GNUNET_free (res);
 }
 
 
@@ -555,7 +617,7 @@ transmit_notify (void *cls, size_t *data_size, void *data)
   GNUNET_CONTAINER_DLL_remove (ch->tmit_head, ch->tmit_tail, msg);
   GNUNET_free (msg);
 
-  int ret = (GNUNET_YES == ch->in_transmit) ? GNUNET_NO : GNUNET_YES;
+  int ret = (MSG_STATE_END < ch->tmit_state) ? GNUNET_NO : GNUNET_YES;
 
   if (0 == ch->tmit_task)
   {
@@ -638,6 +700,7 @@ transmit_message (struct Channel *ch, struct GNUNET_TIME_Relative delay)
     : GNUNET_SCHEDULER_add_delayed (delay, slave_transmit_message, ch);
 }
 
+
 /**
  * Queue incoming message parts from a client for transmission, and send them to
  * the multicast group when the buffer is full or reached the end of message.
@@ -659,21 +722,20 @@ queue_message (struct Channel *ch, const struct GNUNET_MessageHeader *msg)
               "for transmission to multicast.\n",
               ntohs (msg->type), size);
 
-  if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < size)
+  if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < size)
     return GNUNET_SYSERR;
 
   if (NULL == tmit_msg
-      || tmit_msg->status != GNUNET_PSYC_DATA_CONT
-      || GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < tmit_msg->size + size)
+      || GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < tmit_msg->size + size)
   {
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                "Appending message qto new buffer.\n");
+                "Appending message to new buffer.\n");
     /* Start filling up new buffer */
     tmit_msg = GNUNET_new (struct TransmitMessage);
     tmit_msg->buf = GNUNET_malloc (size);
     memcpy (tmit_msg->buf, msg, size);
     tmit_msg->size = size;
-    tmit_msg->status = ch->tmit_status;
+    tmit_msg->state = ch->tmit_state;
     GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, tmit_msg);
   }
   else
@@ -684,23 +746,41 @@ queue_message (struct Channel *ch, const struct GNUNET_MessageHeader *msg)
     tmit_msg->buf = GNUNET_realloc (tmit_msg->buf, tmit_msg->size + size);
     memcpy (tmit_msg->buf + tmit_msg->size, msg, size);
     tmit_msg->size += size;
-    tmit_msg->status = ch->tmit_status;
+    tmit_msg->state = ch->tmit_state;
   }
 
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "tmit_size: %u\n", tmit_msg->size);
 
   /* Wait a bit for the remaining message parts from the client
      if there's still some space left in the buffer. */
-  if (GNUNET_PSYC_DATA_CONT == tmit_msg->status
-      && (tmit_msg->size + sizeof (struct GNUNET_PSYC_MessageData)
-          < GNUNET_MULTICAST_FRAGMENT_MAX_SIZE))
+  if (tmit_msg->state < MSG_STATE_END
+      && (tmit_msg->size + sizeof (struct GNUNET_MessageHeader)
+          < GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD))
+  {
     tmit_delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2);
+  }
+  else
+  {
+    send_transmit_ack (ch);
+  }
 
   transmit_message (ch, tmit_delay);
 
   return GNUNET_OK;
 }
 
+
+static void
+transmit_error (struct Channel *ch)
+{
+  struct GNUNET_MessageHeader *msg = GNUNET_malloc (sizeof (*msg));
+  msg->size = ntohs (sizeof (*msg));
+  msg->type = ntohs (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL);
+  queue_message (ch, msg);
+
+  GNUNET_SERVER_client_disconnect (ch->client);
+}
+
 /**
  * Incoming method from a client.
  */
@@ -708,28 +788,21 @@ static void
 handle_transmit_method (void *cls, struct GNUNET_SERVER_Client *client,
                         const struct GNUNET_MessageHeader *msg)
 {
-  const struct GNUNET_PSYC_MessageMethod *meth
-    = (const struct GNUNET_PSYC_MessageMethod *) msg;
+  /* const struct GNUNET_PSYC_MessageMethod *meth
+     = (const struct GNUNET_PSYC_MessageMethod *) msg; */
   struct Channel *ch
     = GNUNET_SERVER_client_get_user_context (client, struct Channel);
   GNUNET_assert (NULL != ch);
 
-  if (GNUNET_NO != ch->in_transmit)
+  if (MSG_STATE_START != ch->tmit_state)
   {
-    /* FIXME: already transmitting a message, send back error message. */
+    transmit_error (ch);
     return;
   }
-
-  ch->in_transmit = GNUNET_YES;
-  ch->tmit_mod_recvd = 0;
-  ch->tmit_mod_count = ntohl (meth->mod_count);
-  ch->tmit_status = GNUNET_PSYC_DATA_CONT;
+  ch->tmit_state = MSG_STATE_METHOD;
 
   queue_message (ch, msg);
-
-  if (0 == ch->tmit_mod_count)
-    send_transmit_ack (ch);
-
+  send_transmit_ack (ch);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 };
 
@@ -741,20 +814,52 @@ static void
 handle_transmit_modifier (void *cls, struct GNUNET_SERVER_Client *client,
                           const struct GNUNET_MessageHeader *msg)
 {
-  /*
   const struct GNUNET_PSYC_MessageModifier *mod
     = (const struct GNUNET_PSYC_MessageModifier *) msg;
-  */
+
   struct Channel *ch
     = GNUNET_SERVER_client_get_user_context (client, struct Channel);
   GNUNET_assert (NULL != ch);
 
-  ch->tmit_mod_recvd++;
+  if (MSG_STATE_METHOD != ch->tmit_state
+      || MSG_STATE_MODIFIER != ch->tmit_state
+      || MSG_STATE_MOD_CONT != ch->tmit_state
+      || ch->tmit_mod_value_size_expected != ch->tmit_mod_value_size)
+  {
+    transmit_error (ch);
+    return;
+  }
+  ch->tmit_mod_value_size_expected = ntohl (mod->value_size);
+  ch->tmit_mod_value_size = ntohs (msg->size) - ntohs(mod->name_size) - 1;
+
   queue_message (ch, msg);
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+};
 
-  if (ch->tmit_mod_recvd == ch->tmit_mod_count)
-    send_transmit_ack (ch);
 
+/**
+ * Incoming modifier from a client.
+ */
+static void
+handle_transmit_mod_cont (void *cls, struct GNUNET_SERVER_Client *client,
+                          const struct GNUNET_MessageHeader *msg)
+{
+  struct Channel *ch
+    = GNUNET_SERVER_client_get_user_context (client, struct Channel);
+  GNUNET_assert (NULL != ch);
+
+  ch->tmit_mod_value_size += ntohs (msg->size);
+
+  if (MSG_STATE_MODIFIER != ch->tmit_state
+      || MSG_STATE_MOD_CONT != ch->tmit_state
+      || ch->tmit_mod_value_size_expected < ch->tmit_mod_value_size)
+  {
+    transmit_error (ch);
+    return;
+  }
+  ch->tmit_state = MSG_STATE_MOD_CONT;
+
+  queue_message (ch, msg);
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 };
 
@@ -766,18 +871,25 @@ static void
 handle_transmit_data (void *cls, struct GNUNET_SERVER_Client *client,
                       const struct GNUNET_MessageHeader *msg)
 {
-  const struct GNUNET_PSYC_MessageData *data
-    = (const struct GNUNET_PSYC_MessageData *) msg;
   struct Channel *ch
     = GNUNET_SERVER_client_get_user_context (client, struct Channel);
   GNUNET_assert (NULL != ch);
 
-  ch->tmit_status = ntohs (data->status);
+  if (MSG_STATE_METHOD != ch->tmit_state
+      || MSG_STATE_MODIFIER != ch->tmit_state
+      || MSG_STATE_MOD_CONT != ch->tmit_state
+      || ch->tmit_mod_value_size_expected != ch->tmit_mod_value_size)
+  {
+    transmit_error (ch);
+    return;
+  }
+  ch->tmit_state = MSG_STATE_DATA;
+
   queue_message (ch, msg);
   send_transmit_ack (ch);
 
-  if (GNUNET_PSYC_DATA_CONT != ch->tmit_status)
-    ch->in_transmit = GNUNET_NO;
+  if (MSG_STATE_END <= ch->tmit_state)
+    ch->tmit_state = MSG_STATE_START;
 
   GNUNET_SERVER_receive_done (client, GNUNET_OK);
 };
@@ -800,16 +912,21 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
 
     { &handle_slave_join, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN, 0 },
-
+#if TODO
+    { &handle_psyc_message, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_MESSAGE, 0 },
+#endif
     { &handle_transmit_method, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD, 0 },
 
     { &handle_transmit_modifier, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER, 0 },
 
+    { &handle_transmit_mod_cont, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT, 0 },
+
     { &handle_transmit_data, NULL,
       GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA, 0 },
-
     { NULL, NULL, 0, 0 }
   };
 
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index 85d10858e..90c07480a 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -29,6 +29,20 @@
 
 #include "gnunet_common.h"
 
+
+enum MessageState
+{
+  MSG_STATE_START = 0,
+  MSG_STATE_HEADER = 1,
+  MSG_STATE_METHOD = 2,
+  MSG_STATE_MODIFIER = 3,
+  MSG_STATE_MOD_CONT = 4,
+  MSG_STATE_DATA = 5,
+  MSG_STATE_END = 6,
+  MSG_STATE_CANCEL = 7,
+};
+
+
 GNUNET_NETWORK_STRUCT_BEGIN
 
 /**** service -> library ****/
@@ -53,8 +67,7 @@ struct OperationResult
    */
   int64_t result_code GNUNET_PACKED;
 
-  /* followed by 0-terminated error message (on error) */
-
+  /* followed by NUL-terminated error message (on error) */
 };
 
 
@@ -74,6 +87,7 @@ struct CountersResult
 };
 
 
+#if REMOVE
 /**
  * Transmit acknowledgment.
  *
@@ -95,6 +109,7 @@ struct TransmitAck
    */
   uint16_t buf_avail;
 };
+#endif
 
 
 /**** library -> service ****/
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index 290f3e375..a5a01fa92 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -30,15 +30,17 @@
  * @author Gabor X Toth
  */
 
+#include <inttypes.h>
+
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_env_lib.h"
+#include "gnunet_multicast_service.h"
 #include "gnunet_psyc_service.h"
 #include "psyc.h"
 
 #define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
 
-
 struct OperationHandle
 {
   struct OperationHandle *prev;
@@ -91,31 +93,55 @@ struct GNUNET_PSYC_Channel
    */
   struct GNUNET_TIME_Relative reconnect_delay;
 
-  GNUNET_PSYC_Method method_cb;
+  /**
+   * Message part callback.
+   */
+  GNUNET_PSYC_MessageCallback message_cb;
+
+  /**
+   * Message part callback for historic message.
+   */
+  GNUNET_PSYC_MessageCallback hist_message_cb;
 
+  /**
+   * Join handler callback.
+   */
   GNUNET_PSYC_JoinCallback join_cb;
 
+  /**
+   * Closure for @a message_cb and @a join_cb.
+   */
   void *cb_cls;
 
   /**
-   * Are we polling for incoming messages right now?
+   * ID of the message being received from the PSYC service.
    */
-  int in_receive;
+  uint64_t recv_message_id;
 
   /**
-   * Are we currently transmitting a message?
+   * State of the currently being received message from the PSYC service.
    */
-  int in_transmit;
+  enum MessageState recv_state;
 
   /**
-   * Is this a master or slave channel?
+   * Flags for the currently being received message from the PSYC service.
+   */
+  enum GNUNET_PSYC_MessageFlags recv_flags;
+
+  /**
+   * Expected value size for the modifier being received from the PSYC service.
+   */
+  uint32_t recv_mod_value_size_expected;
+
+  /**
+   * Actual value size for the modifier being received from the PSYC service.
    */
-  int is_master;
+  uint32_t recv_mod_value_size;
 
   /**
    * Buffer space available for transmitting the next data fragment.
    */
-  uint16_t tmit_buf_avail;
+  uint16_t tmit_size; // FIXME
 
   /**
    * Is transmission paused?
@@ -125,7 +151,22 @@ struct GNUNET_PSYC_Channel
   /**
    * Are we still waiting for a PSYC_TRANSMIT_ACK?
    */
-  uint8_t tmit_ack_pending;
+  uint8_t tmit_ack_pending; // FIXME
+
+  /**
+   * Are we polling for incoming messages right now?
+   */
+  uint8_t in_receive;
+
+  /**
+   * Are we currently transmitting a message?
+   */
+  uint8_t in_transmit;
+
+  /**
+   * Is this a master or slave channel?
+   */
+  uint8_t is_master;
 };
 
 
@@ -135,9 +176,10 @@ struct GNUNET_PSYC_Channel
 struct GNUNET_PSYC_MasterTransmitHandle
 {
   struct GNUNET_PSYC_Master *master;
-  GNUNET_PSYC_MasterTransmitNotify notify;
+  GNUNET_PSYC_MasterTransmitNotify notify_mod;
+  GNUNET_PSYC_MasterTransmitNotify notify_data;
   void *notify_cls;
-  enum GNUNET_PSYC_DataStatus status;
+  enum MessageState state;
 };
 
 
@@ -254,52 +296,383 @@ transmit_next (struct GNUNET_PSYC_Channel *ch);
 
 
 /**
- * Request data from client to transmit.
+ * Reset data stored related to the last received message.
+ */
+static void
+recv_reset (struct GNUNET_PSYC_Channel *ch)
+{
+  ch->recv_state = MSG_STATE_START;
+  ch->recv_flags = 0;
+  ch->recv_message_id = 0;
+  ch->recv_mod_value_size =0;
+  ch->recv_mod_value_size_expected = 0;
+}
+
+
+static void
+recv_error (struct GNUNET_PSYC_Channel *ch)
+{
+  recv_reset (ch);
+
+  GNUNET_PSYC_MessageCallback message_cb
+    = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC
+    ? ch->hist_message_cb
+    : ch->message_cb;
+
+  if (NULL != message_cb)
+    message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, NULL);
+}
+
+/**
+ * Request a modifier from a client to transmit.
  *
  * @param mst Master handle.
  */
 static void
-master_transmit_data (struct GNUNET_PSYC_Master *mst)
+master_transmit_mod (struct GNUNET_PSYC_Master *mst)
 {
   struct GNUNET_PSYC_Channel *ch = &mst->ch;
-  size_t data_size = ch->tmit_buf_avail;
-  struct GNUNET_PSYC_MessageData *pdata;
-  struct OperationHandle *op
-    = GNUNET_malloc (sizeof (*op) + sizeof (*pdata) + data_size);
-  pdata = (struct GNUNET_PSYC_MessageData *) &op[1];
-  op->msg = (struct GNUNET_MessageHeader *) pdata;
-  pdata->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA);
+  uint16_t max_data_size
+    = ch->tmit_size > sizeof (struct GNUNET_MessageHeader)
+    ? GNUNET_PSYC_MODIFIER_MAX_PAYLOAD - ch->tmit_size
+    : GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD - ch->tmit_size;
+  uint16_t data_size = max_data_size;
 
-  switch (mst->tmit->notify (mst->tmit->notify_cls, &data_size, &pdata[1]))
+  struct GNUNET_MessageHeader *msg;
+  struct OperationHandle *op
+    = GNUNET_malloc (sizeof (*op) + sizeof (*msg) + data_size);
+  op->msg = msg = (struct GNUNET_MessageHeader *) &op[1];
+  msg->type
+    = MSG_STATE_MODIFIER == mst->tmit->state
+    ? htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER)
+    : htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT);
+
+  int notify_ret = mst->tmit->notify_data (mst->tmit->notify_cls,
+                                           &data_size, &msg[1]);
+  switch (notify_ret)
   {
   case GNUNET_NO:
-    mst->tmit->status = GNUNET_PSYC_DATA_CONT;
+    if (0 != data_size)
+      mst->tmit->state = MSG_STATE_MOD_CONT;
     break;
 
   case GNUNET_YES:
-    mst->tmit->status = GNUNET_PSYC_DATA_END;
+    mst->tmit->state = (0 == data_size) ? MSG_STATE_DATA : MSG_STATE_MODIFIER;
     break;
 
   default:
-    mst->tmit->status = GNUNET_PSYC_DATA_CANCEL;
-    data_size = 0;
-    LOG (GNUNET_ERROR_TYPE_ERROR, "MasterTransmitNotify returned error.\n");
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+         "MasterTransmitNotify returned error when requesting a modifier.\n");
+
+    mst->tmit->state = MSG_STATE_START;
+    msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL);
+    msg->size = htons (sizeof (*msg));
+
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
+    transmit_next (ch);
+    return;
   }
 
-  if ((GNUNET_PSYC_DATA_CONT == mst->tmit->status && 0 == data_size))
+  if ((GNUNET_NO == notify_ret && 0 == data_size))
   {
     /* Transmission paused, nothing to send. */
     ch->tmit_paused = GNUNET_YES;
     GNUNET_free (op);
   }
-  else
+
+  if (0 < data_size)
+  {
+    GNUNET_assert (data_size <= GNUNET_PSYC_DATA_MAX_PAYLOAD);
+    msg->size = htons (sizeof (*msg) + data_size);
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
+  }
+
+  /* End of message. */
+  if (GNUNET_YES == notify_ret)
+  {
+    op = GNUNET_malloc (sizeof *(op) + sizeof (*msg));
+    op->msg = msg = (struct GNUNET_MessageHeader *) &op[1];
+    msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END);
+    msg->size = htons (sizeof (*msg));
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
+  }
+
+  transmit_next (ch);
+}
+
+
+/**
+ * Request data from a client to transmit.
+ *
+ * @param mst Master handle.
+ */
+static void
+master_transmit_data (struct GNUNET_PSYC_Master *mst)
+{
+  struct GNUNET_PSYC_Channel *ch = &mst->ch;
+  struct GNUNET_MessageHeader *msg;
+  uint16_t data_size = GNUNET_PSYC_DATA_MAX_PAYLOAD;
+  struct OperationHandle *op
+    = GNUNET_malloc (sizeof (*op) + sizeof (*msg) + data_size);
+  op->msg = msg = (struct GNUNET_MessageHeader *) &op[1];
+  msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA);
+
+  int notify_ret = mst->tmit->notify_data (mst->tmit->notify_cls,
+                                           &data_size, &msg[1]);
+  switch (notify_ret)
   {
-    GNUNET_assert (data_size <= ch->tmit_buf_avail);
-    pdata->header.size = htons (sizeof (*pdata) + data_size);
-    pdata->status = htons (mst->tmit->status);
+  case GNUNET_NO:
+    if (0 == data_size)
+    {
+      /* Transmission paused, nothing to send. */
+      ch->tmit_paused = GNUNET_YES;
+      GNUNET_free (op);
+    }
+    break;
+
+  case GNUNET_YES:
+    mst->tmit->state = MSG_STATE_START;
+    break;
+
+  default:
+    LOG (GNUNET_ERROR_TYPE_ERROR,
+         "MasterTransmitNotify returned error when requesting data.\n");
+
+    mst->tmit->state = MSG_STATE_START;
+    msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL);
+    msg->size = htons (sizeof (*msg));
+
     GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
-    ch->tmit_ack_pending = GNUNET_YES;
     transmit_next (ch);
+    return;
+  }
+
+  if (0 < data_size)
+  {
+    GNUNET_assert (data_size <= GNUNET_PSYC_DATA_MAX_PAYLOAD);
+    msg->size = htons (sizeof (*msg) + data_size);
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
+  }
+
+  /* End of message. */
+  if (GNUNET_YES == notify_ret)
+  {
+    op = GNUNET_malloc (sizeof *(op) + sizeof (*msg));
+    op->msg = msg = (struct GNUNET_MessageHeader *) &op[1];
+    msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END);
+    msg->size = htons (sizeof (*msg));
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
+  }
+
+  transmit_next (ch);
+}
+
+
+/**
+ * Handle incoming message from the PSYC service.
+ *
+ * @param ch The channel the message is sent to.
+ * @param pmsg The message.
+ */
+static void
+handle_psyc_message (struct GNUNET_PSYC_Channel *ch,
+                     const struct GNUNET_PSYC_MessageHeader *pmsg)
+{
+  const struct GNUNET_MessageHeader *msg;
+  uint16_t msize = ntohs (pmsg->header.size);
+  uint16_t pos = 0;
+  uint16_t size = 0;
+  uint16_t type, size_eq, size_min;
+
+  if (MSG_STATE_START == ch->recv_state)
+  {
+    ch->recv_message_id = GNUNET_ntohll (pmsg->message_id);
+    ch->recv_flags = ntohl (pmsg->flags);
+  }
+  else if (GNUNET_ntohll (pmsg->message_id) != ch->recv_message_id)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "Unexpected message ID. Got: %" PRIu64 ", expected: %" PRIu64 "\n",
+         GNUNET_ntohll (pmsg->message_id), ch->recv_message_id);
+    GNUNET_break_op (0);
+    recv_error (ch);
+  }
+  else if (ntohl (pmsg->flags) != ch->recv_flags)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "Unexpected message flags. Got: %lu, expected: %lu\n",
+         ntohl (pmsg->flags), ch->recv_flags);
+    GNUNET_break_op (0);
+    recv_error (ch);
+  }
+
+  for (pos = 0; sizeof (*pmsg) + pos < msize; pos += size)
+  {
+    msg = (const struct GNUNET_MessageHeader *) ((char *) &msg[1] + pos);
+    size = ntohs (msg->size);
+    type = ntohs (msg->type);
+    size_eq = size_min = 0;
+
+    if (msize < sizeof (*pmsg) + pos + size)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Discarding message of type %u with invalid size. "
+                  "(%u < %u + %u + %u)\n", ntohs (msg->type),
+                  msize, sizeof (*msg), pos, size);
+      break;
+    }
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Received message part of type %u and size %u from PSYC.\n",
+                ntohs (msg->type), size);
+
+
+    switch (type)
+    {
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+      size_min = sizeof (struct GNUNET_PSYC_MessageMethod);
+      break;
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+      size_min = sizeof (struct GNUNET_PSYC_MessageModifier);
+      break;
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+      size_min = sizeof (struct GNUNET_MessageHeader);
+      break;
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+      size_eq = sizeof (struct GNUNET_MessageHeader);
+      break;
+    }
+
+    if (! ((0 < size_eq && size == size_eq)
+           || (0 < size_min && size_min <= size)))
+    {
+      GNUNET_break (0);
+      reschedule_connect (ch);
+      return;
+    }
+
+    switch (type)
+    {
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+    {
+      struct GNUNET_PSYC_MessageMethod *meth
+        = (struct GNUNET_PSYC_MessageMethod *) msg;
+
+      if (MSG_STATE_HEADER != ch->recv_state)
+      {
+        LOG (GNUNET_ERROR_TYPE_WARNING,
+             "Discarding out of order message method.\n");
+        /* It is normal to receive an incomplete message right after connecting,
+         * but should not happen later.
+         * FIXME: add a check for this condition.
+         */
+        GNUNET_break_op (0);
+        recv_error (ch);
+        break;
+      }
+
+      if ('\0' != (char *) meth + msg->size - 1)
+      {
+        LOG (GNUNET_ERROR_TYPE_WARNING,
+             "Discarding message with malformed method. "
+             "Message ID: %" PRIu64 "\n", ch->recv_message_id);
+        GNUNET_break_op (0);
+        recv_error (ch);
+        break;
+      }
+      GNUNET_PSYC_MessageCallback message_cb
+        = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC
+        ? ch->hist_message_cb
+        : ch->message_cb;
+
+      if (NULL != message_cb)
+        message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, msg);
+
+      ch->recv_state = MSG_STATE_METHOD;
+      break;
+    }
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+    {
+      if (MSG_STATE_MODIFIER != ch->recv_state)
+      {
+        LOG (GNUNET_ERROR_TYPE_WARNING,
+             "Discarding out of order message modifier.\n");
+        GNUNET_break_op (0);
+        recv_error (ch);
+        break;
+      }
+
+      struct GNUNET_PSYC_MessageModifier *mod
+        = (struct GNUNET_PSYC_MessageModifier *) msg;
+
+      uint16_t name_size = ntohs (mod->name_size);
+      ch->recv_mod_value_size_expected = ntohs (mod->value_size);
+      ch->recv_mod_value_size = size - sizeof (*mod) - name_size - 1;
+
+      if (size < sizeof (*mod) + name_size + 1
+          || '\0' != (char *) &mod[1] + mod->name_size
+          || ch->recv_mod_value_size_expected < ch->recv_mod_value_size)
+      {
+        LOG (GNUNET_ERROR_TYPE_WARNING, "Discarding malformed modifier.\n");
+        GNUNET_break_op (0);
+        break;
+      }
+
+      ch->recv_state = MSG_STATE_MODIFIER;
+
+      GNUNET_PSYC_MessageCallback message_cb
+        = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC
+        ? ch->hist_message_cb
+        : ch->message_cb;
+
+      if (NULL != message_cb)
+        message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, msg);
+
+      break;
+    }
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+    {
+      ch->recv_mod_value_size += size - sizeof (*msg);
+
+      if (MSG_STATE_MODIFIER != ch->recv_state
+          || ch->recv_mod_value_size_expected < ch->recv_mod_value_size)
+      {
+        LOG (GNUNET_ERROR_TYPE_WARNING,
+             "Discarding out of order message modifier continuation.\n");
+        GNUNET_break_op (0);
+        recv_reset (ch);
+        break;
+      }
+
+      GNUNET_PSYC_MessageCallback message_cb
+        = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC
+        ? ch->hist_message_cb
+        : ch->message_cb;
+
+      if (NULL != message_cb)
+        message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, msg);
+      break;
+    }
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+    {
+      if (ch->recv_state < MSG_STATE_METHOD
+          || ch->recv_mod_value_size_expected != ch->recv_mod_value_size)
+      {
+        LOG (GNUNET_ERROR_TYPE_WARNING,
+             "Discarding out of order message data fragment.\n");
+        GNUNET_break_op (0);
+        recv_reset (ch);
+        break;
+      }
+
+      ch->recv_state = MSG_STATE_DATA;
+      break;
+    }
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+    case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+      recv_reset (ch);
+      break;
+    }
   }
 }
 
@@ -319,11 +692,10 @@ message_handler (void *cls,
   struct GNUNET_PSYC_Channel *ch = cls;
   struct GNUNET_PSYC_Master *mst = cls;
   struct GNUNET_PSYC_Slave *slv = cls;
-  struct CountersResult *cres;
-  struct TransmitAck *tack;
 
   if (NULL == msg)
   {
+    GNUNET_break (0);
     reschedule_connect (ch);
     return;
   }
@@ -342,8 +714,8 @@ message_handler (void *cls,
   case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
     size_eq = sizeof (struct CountersResult);
     break;
-  case GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK:
-    size_eq = sizeof (struct TransmitAck);
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE:
+    size_min = sizeof (struct GNUNET_PSYC_MessageHeader);
     break;
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
     size_min = sizeof (struct GNUNET_PSYC_MessageMethod);
@@ -352,11 +724,13 @@ message_handler (void *cls,
     size_min = sizeof (struct GNUNET_PSYC_MessageModifier);
     break;
   case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
-    size_min = sizeof (struct GNUNET_PSYC_MessageData);
+    size_min = sizeof (struct GNUNET_MessageHeader);
+    break;
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+  case GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK:
+    size_eq = sizeof (struct GNUNET_MessageHeader);
     break;
-  default:
-    GNUNET_break_op (0);
-    return;
   }
 
   if (! ((0 < size_eq && size == size_eq)
@@ -370,38 +744,63 @@ message_handler (void *cls,
   switch (type)
   {
   case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK:
-    cres = (struct CountersResult *) msg;
+  {
+    struct CountersResult *cres = (struct CountersResult *) msg;
     mst->max_message_id = GNUNET_ntohll (cres->max_message_id);
     if (NULL != mst->start_cb)
       mst->start_cb (ch->cb_cls, mst->max_message_id);
     break;
-
+  }
   case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
-    cres = (struct CountersResult *) msg;
+  {
 #if TODO
+    struct CountersResult *cres = (struct CountersResult *) msg;
     slv->max_message_id = GNUNET_ntohll (cres->max_message_id);
     if (NULL != slv->join_ack_cb)
       mst->join_ack_cb (ch->cb_cls, mst->max_message_id);
 #endif
     break;
-
+  }
   case GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK:
-    tack = (struct TransmitAck *) msg;
+  {
+    ch->tmit_ack_pending = GNUNET_NO;
+
     if (ch->is_master)
     {
       GNUNET_assert (NULL != mst->tmit);
-      if (GNUNET_PSYC_DATA_CONT != mst->tmit->status
-          || NULL == mst->tmit->notify)
-      {
-        GNUNET_free (mst->tmit);
-        mst->tmit = NULL;
-      }
-      else
+      switch (mst->tmit->state)
       {
-        ch->tmit_buf_avail = ntohs (tack->buf_avail);
-        ch->tmit_ack_pending = GNUNET_NO;
+      case MSG_STATE_MODIFIER:
+        if (GNUNET_NO == ch->tmit_paused)
+          master_transmit_mod (mst);
+        break;
+
+      case MSG_STATE_MOD_CONT:
+        if (GNUNET_NO == ch->tmit_paused)
+          master_transmit_mod (mst);
+        break;
+
+      case MSG_STATE_DATA:
         if (GNUNET_NO == ch->tmit_paused)
           master_transmit_data (mst);
+        break;
+
+      case MSG_STATE_END:
+      case MSG_STATE_CANCEL:
+        if (NULL != mst->tmit)
+        {
+          GNUNET_free (mst->tmit);
+          mst->tmit = NULL;
+        }
+        else
+        {
+          LOG (GNUNET_ERROR_TYPE_WARNING,
+               "Ignoring transmit ack, there's no transmission going on.\n");
+        }
+        break;
+      default:
+        LOG (GNUNET_ERROR_TYPE_WARNING,
+             "Ignoring unexpected transmit ack.\n");
       }
     }
     else
@@ -409,17 +808,10 @@ message_handler (void *cls,
       /* TODO: slave */
     }
     break;
+  }
 
-  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
-
-    break;
-
-  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
-
-    break;
-
-  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
-
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE:
+    handle_psyc_message(ch, (const struct GNUNET_PSYC_MessageHeader *) msg);
     break;
   }
 
@@ -506,6 +898,7 @@ reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct GNUNET_PSYC_Channel *ch = cls;
 
+  recv_reset (ch);
   ch->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Connecting to PSYC service.\n");
@@ -588,7 +981,7 @@ disconnect (void *c)
  *        one in the future.
  * @param policy Channel policy specifying join and history restrictions.
  *        Used to automate join decisions.
- * @param method Function to invoke on messages received from slaves.
+ * @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.
@@ -598,7 +991,7 @@ 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_Method method,
+                          GNUNET_PSYC_MessageCallback message_cb,
                           GNUNET_PSYC_JoinCallback join_cb,
                           GNUNET_PSYC_MasterStartCallback master_started_cb,
                           void *cls)
@@ -618,7 +1011,7 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
   ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
   ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, mst);
 
-  ch->method_cb = method;
+  ch->message_cb = message_cb;
   ch->join_cb = join_cb;
   ch->cb_cls = cls;
   mst->start_cb = master_started_cb;
@@ -705,19 +1098,17 @@ send_modifier (void *cls, struct GNUNET_ENV_Modifier *mod)
  *
  * @param master Handle to the PSYC channel.
  * @param method_name Which method should be invoked.
- * @param env Environment containing state operations and transient variables
- *            for the message, or NULL.
- * @param notify Function to call to obtain the arguments.
- * @param notify_cls Closure for @a notify.
+ * @param notify_mod Function to call to obtain modifiers.
+ * @param notify_data Function to call to obtain fragments of the data.
+ * @param notify_cls Closure for @a notify_mod and @a notify_data.
  * @param flags Flags for the message being transmitted.
- * @return Transmission handle, NULL on error (i.e. more than one request
- *         queued).
+ * @return Transmission handle, NULL on error (i.e. more than one request queued).
  */
 struct GNUNET_PSYC_MasterTransmitHandle *
 GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
                              const char *method_name,
-                             const struct GNUNET_ENV_Environment *env,
-                             GNUNET_PSYC_MasterTransmitNotify notify,
+                             GNUNET_PSYC_MasterTransmitNotify notify_mod,
+                             GNUNET_PSYC_MasterTransmitNotify notify_data,
                              void *notify_cls,
                              enum GNUNET_PSYC_MasterTransmitFlags flags)
 {
@@ -737,18 +1128,17 @@ GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
   pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD);
   pmeth->header.size = htons (sizeof (*pmeth) + size);
   pmeth->flags = htonl (flags);
-  pmeth->mod_count = GNUNET_ntohll (GNUNET_ENV_environment_get_mod_count (env));
   memcpy (&pmeth[1], method_name, size);
 
   GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
-  GNUNET_ENV_environment_iterate (env, send_modifier, master);
   transmit_next (ch);
 
   master->tmit = GNUNET_malloc (sizeof (*master->tmit));
   master->tmit->master = master;
-  master->tmit->notify = notify;
+  master->tmit->notify_mod = notify_mod;
+  master->tmit->notify_data = notify_data;
   master->tmit->notify_cls = notify_cls;
-  master->tmit->status = GNUNET_PSYC_DATA_CONT;
+  master->tmit->state = MSG_STATE_START; // FIXME
   return master->tmit;
 }
 
@@ -804,12 +1194,13 @@ GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th)
  * @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 method Function to invoke on messages received from the channel,
- *        typically at least contains functions for @e join and @e part.
+ * @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 join_cb function invoked once we have joined with the current
  *        message ID of the channel
  * @param slave_joined_cb Function to invoke when a peer wants to join.
- * @param cls Closure for @a method_cb and @a slave_joined_cb.
+ * @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.
@@ -823,7 +1214,7 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
                         const struct GNUNET_PeerIdentity *origin,
                         uint32_t relay_count,
                         const struct GNUNET_PeerIdentity *relays,
-                        GNUNET_PSYC_Method method,
+                        GNUNET_PSYC_MessageCallback message_cb,
                         GNUNET_PSYC_JoinCallback join_cb,
                         GNUNET_PSYC_SlaveJoinCallback slave_joined_cb,
                         void *cls,
@@ -845,6 +1236,10 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
   req->relay_count = relay_count;
   memcpy (&req[1], relays, relay_count * sizeof (*relays));
 
+  ch->message_cb = message_cb;
+  ch->join_cb = join_cb;
+  ch->cb_cls = cls;
+
   ch->cfg = cfg;
   ch->is_master = GNUNET_NO;
   ch->reconnect_msg = (struct GNUNET_MessageHeader *) req;
@@ -1043,7 +1438,7 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
  * @param channel Which channel should be replayed?
  * @param start_message_id Earliest interesting point in history.
  * @param end_message_id Last (exclusive) interesting point in history.
- * @param method Function to invoke on messages received from the story.
+ * @param message_cb Function to invoke on message parts received from the story.
  * @param finish_cb Function to call when the requested story has been fully
  *        told (counting message IDs might not suffice, as some messages
  *        might be secret and thus the listener would not know the story is
@@ -1057,8 +1452,8 @@ struct GNUNET_PSYC_Story *
 GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *channel,
                                 uint64_t start_message_id,
                                 uint64_t end_message_id,
-                                GNUNET_PSYC_Method method,
-                                GNUNET_PSYC_FinishCallback *finish_cb,
+                                GNUNET_PSYC_MessageCallback message_cb,
+                                GNUNET_PSYC_FinishCallback finish_cb,
                                 void *cls)
 {
   return NULL;
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 2986fdf6a..704819c50 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -148,11 +148,16 @@ join (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
 struct TransmitClosure
 {
   struct GNUNET_PSYC_MasterTransmitHandle *handle;
-  uint8_t n;
+
+  char *mod_names[16];
+  char *mod_values[16];
+  char *data[16];
+
+  uint8_t mod_count;
+  uint8_t data_count;
+
   uint8_t paused;
-  uint8_t fragment_count;
-  char *fragments[16];
-  uint16_t fragment_sizes[16];
+  uint8_t n;
 };
 
 
@@ -167,16 +172,47 @@ transmit_resume (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 
 
 static int
-transmit_notify (void *cls, size_t *data_size, void *data)
+tmit_notify_mod (void *cls, size_t *data_size, void *data)
 {
   struct TransmitClosure *tmit = cls;
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Transmit notify: %lu bytes available, "
+              "Transmit notify modifier: %lu bytes available, "
+              "processing modifier %u/%u.\n",
+              *data_size, tmit->n + 1, tmit->fragment_count);
+  /* FIXME: continuation */
+  uint16_t name_size = strlen (tmit->mod_names[tmit->n]);
+  uint16_t value_size = strlen (tmit->mod_values[tmit->n]);
+  if (name_size + 1 + value_size <= *data_size)
+    return GNUNET_NO;
+
+  *data_size = name_size + 1 + value_size;
+  memcpy (data, tmit->fragments[tmit->n], *data_size);
+
+  if (++tmit->n < tmit->mod_count)
+  {
+    return GNUNET_NO;
+  }
+  else
+  {
+    tmit->n = 0;
+    return GNUNET_YES;
+  }
+}
+
+
+static int
+tmit_notify_data (void *cls, size_t *data_size, void *data)
+{
+  struct TransmitClosure *tmit = cls;
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Transmit notify data: %lu bytes available, "
               "processing fragment %u/%u.\n",
               *data_size, tmit->n + 1, tmit->fragment_count);
-  GNUNET_assert (tmit->fragment_sizes[tmit->n] <= *data_size);
+  uint16_t size = strlen (tmit->data[tmit->n]);
+  if (size <= *data_size)
+    return GNUNET_NO;
 
-  if (GNUNET_YES == tmit->paused && tmit->n == tmit->fragment_count - 1)
+  if (GNUNET_YES == tmit->paused && tmit->n == tmit->data_count - 1)
   {
     /* Send last fragment later. */
     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission paused.\n");
@@ -188,13 +224,13 @@ transmit_notify (void *cls, size_t *data_size, void *data)
     return GNUNET_NO;
   }
 
-  GNUNET_assert (tmit->fragment_sizes[tmit->n] <= *data_size);
-  *data_size = tmit->fragment_sizes[tmit->n];
-  memcpy (data, tmit->fragments[tmit->n], *data_size);
+  *data_size = size;
+  memcpy (data, tmit->data[tmit->n], size);
 
-  return ++tmit->n < tmit->fragment_count ? GNUNET_NO : GNUNET_YES;
+  return ++tmit->n < tmit->data_count ? GNUNET_NO : GNUNET_YES;
 }
 
+
 void
 master_started (void *cls, uint64_t max_message_id)
 {
@@ -208,15 +244,13 @@ master_started (void *cls, uint64_t max_message_id)
                                   "_foo_bar", "foo bar baz", 11);
 
   struct TransmitClosure *tmit = GNUNET_new (struct TransmitClosure);
-  tmit->fragment_count = 3;
-  tmit->fragments[0] = "foo";
-  tmit->fragment_sizes[0] = 4;
-  tmit->fragments[1] = "foo bar";
-  tmit->fragment_sizes[1] = 7;
-  tmit->fragments[2] = "foo bar baz";
-  tmit->fragment_sizes[2] = 11;
+  tmit->data[0] = "foo";
+  tmit->data[1] = "foo bar";
+  tmit->data[2] = "foo bar baz";
+  tmit->data_count = 3;
   tmit->handle
-    = GNUNET_PSYC_master_transmit (mst, "_test", env, transmit_notify, tmit,
+    = GNUNET_PSYC_master_transmit (mst, "_test", tmit_notify_mod,
+                                   tmit_notify_data, tmit,
                                    GNUNET_PSYC_MASTER_TRANSMIT_INC_GROUP_GEN);
 }