psyc: handling messages from multicast and passing them to clients; pause/resume fixes

3 files changed, 355 insertions, 109 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 77eb82a4c..a3b1b8f82 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -56,7 +56,7 @@ static struct GNUNET_SERVER_NotificationContext *nc;
 static struct GNUNET_PSYCSTORE_Handle *store;
 
 /**
- * channel's pub_key_hash -> struct Channel
+ * Channel's pub_key_hash -> struct Channel
  */
 static struct GNUNET_CONTAINER_MultiHashMap *clients;
 
@@ -70,6 +70,9 @@ struct TransmitMessage
 
   char *buf;
   uint16_t size;
+  /**
+   * enum GNUNET_PSYC_DataStatus
+   */
   uint8_t status;
 };
 
@@ -83,15 +86,17 @@ struct Channel
   struct TransmitMessage *tmit_head;
   struct TransmitMessage *tmit_tail;
 
-  char *tmit_buf;
   GNUNET_SCHEDULER_TaskIdentifier tmit_task;
   uint32_t tmit_mod_count;
   uint32_t tmit_mod_recvd;
-  uint16_t tmit_size;
+  /**
+   * enum GNUNET_PSYC_DataStatus
+   */
   uint8_t tmit_status;
 
   uint8_t in_transmit;
   uint8_t is_master;
+  uint8_t disconnected;
 };
 
 /**
@@ -142,6 +147,10 @@ struct Slave
 };
 
 
+static void
+transmit_message (struct Channel *ch, struct GNUNET_TIME_Relative delay);
+
+
 /**
  * Task run during shutdown.
  *
@@ -163,6 +172,30 @@ shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   }
 }
 
+
+static void
+client_cleanup (struct Channel *ch)
+{
+  if (ch->is_master)
+  {
+    struct Master *mst = (struct Master *) ch;
+    if (NULL != mst->origin)
+      GNUNET_MULTICAST_origin_stop (mst->origin);
+  }
+  else
+  {
+    struct Slave *slv = (struct Slave *) ch;
+    if (NULL != slv->join_req)
+      GNUNET_free (slv->join_req);
+    if (NULL != slv->relays)
+      GNUNET_free (slv->relays);
+    if (NULL != slv->member)
+      GNUNET_MULTICAST_member_part (slv->member);
+  }
+
+  GNUNET_free (ch);
+}
+
 /**
  * Called whenever a client is disconnected.
  * Frees our resources associated with that client.
@@ -188,30 +221,17 @@ client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
     return;
   }
 
-  if (NULL != ch->tmit_buf)
-  {
-    GNUNET_free (ch->tmit_buf);
-    ch->tmit_buf = NULL;
-  }
+  ch->disconnected = GNUNET_YES;
 
-  if (ch->is_master)
+  /* Send pending messages to multicast before cleanup. */
+  if (NULL != ch->tmit_head)
   {
-    struct Master *mst = (struct Master *) ch;
-    if (NULL != mst->origin)
-      GNUNET_MULTICAST_origin_stop (mst->origin);
+    transmit_message (ch, GNUNET_TIME_UNIT_ZERO);
   }
   else
   {
-    struct Slave *slv = (struct Slave *) ch;
-    if (NULL != slv->join_req)
-      GNUNET_free (slv->join_req);
-    if (NULL != slv->relays)
-      GNUNET_free (slv->relays);
-    if (NULL != slv->member)
-      GNUNET_MULTICAST_member_part (slv->member);
+    client_cleanup (ch);
   }
-
-  GNUNET_free (ch);
 }
 
 void
@@ -259,14 +279,98 @@ request_cb (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *member_key,
 
 }
 
+
+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.
+ */
+int
+send_to_client (void *cls, const struct GNUNET_HashCode *ch_key_hash, void *chan)
+{
+  const struct GNUNET_MULTICAST_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, pmsg,
+                                                GNUNET_NO);
+    pos += psize;
+  }
+  return GNUNET_YES;
+}
+
+
+/**
+ * Incoming message fragment from multicast.
+ *
+ * Store it using PSYCstore and send it to all clients of the channel.
+ */
 void
 message_cb (void *cls, const struct GNUNET_MessageHeader *msg)
 {
+  uint16_t type = ntohs (msg->type);
+  uint16_t size = ntohs (msg->size);
+
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Received message of type %u from multicast.\n",
-              ntohs (msg->type));
+              "Received message of type %u and size %u from multicast.\n",
+              type, size);
+
+  struct Channel *ch = cls;
+  struct Master *mst = cls;
+  struct Slave *slv = cls;
+
+  struct GNUNET_CRYPTO_EddsaPublicKey *ch_key
+    = ch->is_master ? &mst->pub_key : &slv->chan_key;
+  struct GNUNET_HashCode *ch_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,
+                                     (const struct
+                                      GNUNET_MULTICAST_MessageHeader *) msg,
+                                     0, NULL, NULL);
+    GNUNET_CONTAINER_multihashmap_get_multiple (clients, ch_key_hash,
+                                                send_to_client, (void *) msg);
+    break;
+
+  default:
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "Ignoring unknown message of type %u and size %u.\n",
+                type, size);
+  }
 }
 
+
+/**
+ * Response from PSYCstore with the current counter values for a channel master.
+ */
 void
 master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
                     uint64_t max_message_id, uint64_t max_group_generation,
@@ -299,6 +403,9 @@ master_counters_cb (void *cls, int result, uint64_t max_fragment_id,
 }
 
 
+/**
+ * Response from PSYCstore with the current counter values for a channel slave.
+ */
 void
 slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
                    uint64_t max_message_id, uint64_t max_group_generation,
@@ -332,6 +439,9 @@ slave_counters_cb (void *cls, int result, uint64_t max_fragment_id,
 }
 
 
+/**
+ * Handle a connecting client starting a channel master.
+ */
 static void
 handle_master_start (void *cls, struct GNUNET_SERVER_Client *client,
                      const struct GNUNET_MessageHeader *msg)
@@ -357,6 +467,9 @@ 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,
                    const struct GNUNET_MessageHeader *msg)
@@ -389,13 +502,26 @@ handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
 }
 
 
+/**
+ * Send transmission acknowledgement to a client.
+ *
+ * Sent after the last GNUNET_PSYC_MessageModifier and after each
+ * GNUNET_PSYC_MessageData.
+ *
+ * @param ch The channel struct for the 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 = htons (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE - ch->tmit_size);
+
+  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);
 
   GNUNET_SERVER_notification_context_add (nc, ch->client);
   GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
@@ -404,30 +530,53 @@ send_transmit_ack (struct Channel *ch)
 }
 
 
+/**
+ * Callback for the transmit functions of multicast.
+ */
 static int
 transmit_notify (void *cls, size_t *data_size, void *data)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "transmit_notify()\n");
   struct Channel *ch = cls;
   struct TransmitMessage *msg = ch->tmit_head;
 
-  if (NULL == msg || *data_size < ntohs (msg->size))
+  if (NULL == msg || *data_size < msg->size)
   {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "transmit_notify: nothing to send.\n");
     *data_size = 0;
     return GNUNET_NO;
   }
 
-  *data_size = ntohs (msg->size);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "transmit_notify: sending %u bytes.\n", msg->size);
+
+  *data_size = msg->size;
   memcpy (data, msg->buf, *data_size);
 
-  GNUNET_free (ch->tmit_buf);
-  ch->tmit_buf = NULL;
   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;
+
+  if (0 == ch->tmit_task)
+  {
+    if (NULL != ch->tmit_head)
+    {
+      transmit_message (ch, GNUNET_TIME_UNIT_ZERO);
+    }
+    else if (ch->disconnected)
+    {
+      /* FIXME: handle partial message (when still in_transmit) */
+      client_cleanup (ch);
+    }
+  }
 
-  return (GNUNET_YES == ch->in_transmit) ? GNUNET_NO : GNUNET_YES;
+  return ret;
 }
 
 
+/**
+ * Transmit a message from a channel master to the multicast group.
+ */
 static void
 master_transmit_message (void *cls,
                          const struct GNUNET_SCHEDULER_TaskContext *tc)
@@ -449,6 +598,9 @@ master_transmit_message (void *cls,
 }
 
 
+/**
+ * Transmit a message from a channel slave to the multicast group.
+ */
 static void
 slave_transmit_message (void *cls,
                         const struct GNUNET_SCHEDULER_TaskContext *tc)
@@ -468,50 +620,90 @@ slave_transmit_message (void *cls,
 }
 
 
+/**
+ * Schedule message transmission from a channel to the multicast group.
+ *
+ * @param ch The channel.
+ * @param delay Transmission delay.
+ */
+static void
+transmit_message (struct Channel *ch, struct GNUNET_TIME_Relative delay)
+{
+  if (0 != ch->tmit_task)
+    GNUNET_SCHEDULER_cancel (ch->tmit_task);
+
+  ch->tmit_task
+    = ch->is_master
+    ? GNUNET_SCHEDULER_add_delayed (delay, master_transmit_message, ch)
+    : 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.
+ *
+ * @param ch Channel struct for the client.
+ * @param msg Message from the client.
+ *
+ * @return #GNUNET_OK on success, else #GNUNET_SYSERR.
+ */
 static int
-buffer_message (struct Channel *ch, const struct GNUNET_MessageHeader *msg)
+queue_message (struct Channel *ch, const struct GNUNET_MessageHeader *msg)
 {
   uint16_t size = ntohs (msg->size);
   struct GNUNET_TIME_Relative tmit_delay = GNUNET_TIME_UNIT_ZERO;
+  struct TransmitMessage *tmit_msg = ch->tmit_tail;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Queueing message of type %u and size %u "
+              "for transmission to multicast.\n",
+              ntohs (msg->type), size);
 
   if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < size)
     return GNUNET_SYSERR;
 
-  if (0 == ch->tmit_size)
-  {
-    ch->tmit_buf = GNUNET_malloc (size);
-    memcpy (ch->tmit_buf, msg, size);
-    ch->tmit_size = size;
-  }
-  else if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE <= ch->tmit_size + size)
-  {
-    ch->tmit_buf = GNUNET_realloc (ch->tmit_buf, ch->tmit_size + size);
-    memcpy (ch->tmit_buf + ch->tmit_size, msg, size);
-    ch->tmit_size += size;
-  }
-
-  if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE
-      < ch->tmit_size + sizeof (struct GNUNET_PSYC_MessageData))
+  if (NULL == tmit_msg
+      || tmit_msg->status != GNUNET_PSYC_DATA_CONT
+      || GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < tmit_msg->size + size)
   {
-    struct TransmitMessage *tmit_msg = GNUNET_new (struct TransmitMessage);
-    tmit_msg->buf = (char *) msg;
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Appending message qto 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;
     GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, tmit_msg);
-    tmit_delay = GNUNET_TIME_UNIT_ZERO;
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                "Appending message to existing buffer.\n");
+    /* Append to existing buffer */
+    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;
   }
 
-  if (0 != ch->tmit_task)
-    GNUNET_SCHEDULER_cancel (ch->tmit_task);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "tmit_size: %u\n", tmit_msg->size);
 
-  ch->tmit_task
-    = ch->is_master
-    ? GNUNET_SCHEDULER_add_delayed (tmit_delay, master_transmit_message, ch)
-    : GNUNET_SCHEDULER_add_delayed (tmit_delay, slave_transmit_message, ch);
+  /* 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))
+    tmit_delay = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 2);
+
+  transmit_message (ch, tmit_delay);
 
   return GNUNET_OK;
 }
 
+/**
+ * Incoming method from a client.
+ */
 static void
 handle_transmit_method (void *cls, struct GNUNET_SERVER_Client *client,
                         const struct GNUNET_MessageHeader *msg)
@@ -524,18 +716,16 @@ handle_transmit_method (void *cls, struct GNUNET_SERVER_Client *client,
 
   if (GNUNET_NO != ch->in_transmit)
   {
-    // FIXME: already transmitting a message, send back error message.
+    /* FIXME: already transmitting a message, send back error message. */
     return;
   }
 
   ch->in_transmit = GNUNET_YES;
-  ch->tmit_buf = NULL;
-  ch->tmit_size = 0;
   ch->tmit_mod_recvd = 0;
   ch->tmit_mod_count = ntohl (meth->mod_count);
   ch->tmit_status = GNUNET_PSYC_DATA_CONT;
 
-  buffer_message (ch, msg);
+  queue_message (ch, msg);
 
   if (0 == ch->tmit_mod_count)
     send_transmit_ack (ch);
@@ -544,18 +734,23 @@ handle_transmit_method (void *cls, struct GNUNET_SERVER_Client *client,
 };
 
 
+/**
+ * Incoming modifier from a client.
+ */
 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++;
-  buffer_message (ch, msg);
+  queue_message (ch, msg);
 
   if (ch->tmit_mod_recvd == ch->tmit_mod_count)
     send_transmit_ack (ch);
@@ -564,6 +759,9 @@ handle_transmit_modifier (void *cls, struct GNUNET_SERVER_Client *client,
 };
 
 
+/**
+ * Incoming data from a client.
+ */
 static void
 handle_transmit_data (void *cls, struct GNUNET_SERVER_Client *client,
                       const struct GNUNET_MessageHeader *msg)
@@ -575,7 +773,7 @@ handle_transmit_data (void *cls, struct GNUNET_SERVER_Client *client,
   GNUNET_assert (NULL != ch);
 
   ch->tmit_status = ntohs (data->status);
-  buffer_message (ch, msg);
+  queue_message (ch, msg);
   send_transmit_ack (ch);
 
   if (GNUNET_PSYC_DATA_CONT != ch->tmit_status)
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index 18b5920b3..f971de1b8 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -69,12 +69,12 @@ struct GNUNET_PSYC_Channel
   /**
    * Head of operations to transmit.
    */
-  struct OperationHandle *transmit_head;
+  struct OperationHandle *tmit_head;
 
   /**
    * Tail of operations to transmit.
    */
-  struct OperationHandle *transmit_tail;
+  struct OperationHandle *tmit_tail;
 
   /**
    * Message to send on reconnect.
@@ -116,6 +116,16 @@ struct GNUNET_PSYC_Channel
    * Buffer space available for transmitting the next data fragment.
    */
   uint16_t tmit_buf_avail;
+
+  /**
+   * Is transmission paused?
+   */
+  uint8_t tmit_paused;
+
+  /**
+   * Are we still waiting for a PSYC_TRANSMIT_ACK?
+   */
+  uint8_t tmit_ack_pending;
 };
 
 
@@ -243,6 +253,11 @@ static void
 transmit_next (struct GNUNET_PSYC_Channel *ch);
 
 
+/**
+ * Request data from client to transmit.
+ *
+ * @param mst Master handle.
+ */
 static void
 master_transmit_data (struct GNUNET_PSYC_Master *mst)
 {
@@ -268,12 +283,13 @@ master_transmit_data (struct GNUNET_PSYC_Master *mst)
   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.\n");
   }
 
   if ((GNUNET_PSYC_DATA_CONT == mst->tmit->status && 0 == data_size))
   {
     /* Transmission paused, nothing to send. */
+    ch->tmit_paused = GNUNET_YES;
     GNUNET_free (op);
   }
   else
@@ -281,7 +297,8 @@ master_transmit_data (struct GNUNET_PSYC_Master *mst)
     GNUNET_assert (data_size <= ch->tmit_buf_avail);
     pdata->header.size = htons (sizeof (*pdata) + data_size);
     pdata->status = htons (mst->tmit->status);
-    GNUNET_CONTAINER_DLL_insert_tail (ch->transmit_head, ch->transmit_tail, op);
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
+    ch->tmit_ack_pending = GNUNET_YES;
     transmit_next (ch);
   }
 }
@@ -305,7 +322,6 @@ message_handler (void *cls,
   struct CountersResult *cres;
   struct TransmitAck *tack;
 
-
   if (NULL == msg)
   {
     reschedule_connect (ch);
@@ -317,7 +333,8 @@ message_handler (void *cls,
   uint16_t type = ntohs (msg->type);
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Received message of type %d from PSYC service\n", type);
+       "Received message of type %d and size %u from PSYC service\n",
+       type, size);
 
   switch (type)
   {
@@ -328,10 +345,16 @@ message_handler (void *cls,
   case GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK:
     size_eq = sizeof (struct TransmitAck);
     break;
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD:
+    size_min = sizeof (struct GNUNET_PSYC_MessageMethod);
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+    size_min = sizeof (struct GNUNET_PSYC_MessageModifier);
+  case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+    size_min = sizeof (struct GNUNET_PSYC_MessageData);
   }
 
   if (! ((0 < size_eq && size == size_eq)
-         || (0 < size_min && size >= size_min)))
+         || (0 < size_min && size_min <= size)))
   {
     GNUNET_break (0);
     reschedule_connect (ch);
@@ -370,7 +393,9 @@ message_handler (void *cls,
       else
       {
         ch->tmit_buf_avail = ntohs (tack->buf_avail);
-        master_transmit_data (mst);
+        ch->tmit_ack_pending = GNUNET_NO;
+        if (GNUNET_NO == ch->tmit_paused)
+          master_transmit_data (mst);
       }
     }
     else
@@ -378,6 +403,18 @@ 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:
+
+    break;
   }
 
   GNUNET_CLIENT_receive (ch->client, &message_handler, ch,
@@ -397,9 +434,9 @@ static size_t
 send_next_message (void *cls, size_t size, void *buf)
 {
   struct GNUNET_PSYC_Channel *ch = cls;
-  struct OperationHandle *op = ch->transmit_head;
+  struct OperationHandle *op = ch->tmit_head;
   size_t ret;
-
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "send_next_message()\n");
   ch->th = NULL;
   if (NULL == op->msg)
     return 0;
@@ -409,15 +446,12 @@ send_next_message (void *cls, size_t size, void *buf)
     reschedule_connect (ch);
     return 0;
   }
-  LOG (GNUNET_ERROR_TYPE_DEBUG,
-       "Sending message of type %d to PSYC service\n",
-       ntohs (op->msg->type));
   memcpy (buf, op->msg, ret);
 
-  GNUNET_CONTAINER_DLL_remove (ch->transmit_head, ch->transmit_tail, op);
+  GNUNET_CONTAINER_DLL_remove (ch->tmit_head, ch->tmit_tail, op);
   GNUNET_free (op);
 
-  if (NULL != ch->transmit_head)
+  if (NULL != ch->tmit_head)
     transmit_next (ch);
 
   if (GNUNET_NO == ch->in_receive)
@@ -438,10 +472,11 @@ send_next_message (void *cls, size_t size, void *buf)
 static void
 transmit_next (struct GNUNET_PSYC_Channel *ch)
 {
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "transmit_next()\n");
   if (NULL != ch->th || NULL == ch->client)
     return;
 
-  struct OperationHandle *op = ch->transmit_head;
+  struct OperationHandle *op = ch->tmit_head;
   if (NULL == op)
     return;
 
@@ -472,14 +507,14 @@ reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   ch->client = GNUNET_CLIENT_connect ("psyc", ch->cfg);
   GNUNET_assert (NULL != ch->client);
 
-  if (NULL == ch->transmit_head ||
-      ch->transmit_head->msg->type != ch->reconnect_msg->type)
+  if (NULL == ch->tmit_head ||
+      ch->tmit_head->msg->type != ch->reconnect_msg->type)
   {
     uint16_t reconn_size = ntohs (ch->reconnect_msg->size);
     struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + reconn_size);
     memcpy (&op[1], ch->reconnect_msg, reconn_size);
     op->msg = (struct GNUNET_MessageHeader *) &op[1];
-    GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op);
+    GNUNET_CONTAINER_DLL_insert (ch->tmit_head, ch->tmit_tail, op);
   }
   transmit_next (ch);
 }
@@ -496,7 +531,7 @@ disconnect (void *c)
   struct GNUNET_PSYC_Channel *ch = c;
 
   GNUNET_assert (NULL != ch);
-  if (ch->transmit_head != ch->transmit_tail)
+  if (ch->tmit_head != ch->tmit_tail)
   {
     LOG (GNUNET_ERROR_TYPE_ERROR,
          "Disconnecting while there are still outstanding messages!\n");
@@ -654,7 +689,7 @@ send_modifier (void *cls, struct GNUNET_ENV_Modifier *mod)
   memcpy (&pmod[1], mod->name, name_size);
   memcpy ((char *) &pmod[1] + name_size, mod->value, mod->value_size);
 
-  GNUNET_CONTAINER_DLL_insert_tail (ch->transmit_head, ch->transmit_tail, op);
+  GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
   return GNUNET_YES;
 }
 
@@ -699,7 +734,7 @@ GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
   pmeth->mod_count = GNUNET_ntohll (GNUNET_ENV_environment_get_mod_count (env));
   memcpy (&pmeth[1], method_name, size);
 
-  GNUNET_CONTAINER_DLL_insert_tail (ch->transmit_head, ch->transmit_tail, op);
+  GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, op);
   GNUNET_ENV_environment_iterate (env, send_modifier, master);
   transmit_next (ch);
 
@@ -720,7 +755,12 @@ GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
 void
 GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th)
 {
-  master_transmit_data (th->master);
+  struct GNUNET_PSYC_Channel *ch = &th->master->ch;
+  if (GNUNET_NO == ch->tmit_ack_pending)
+  {
+    ch->tmit_paused = GNUNET_NO;
+    master_transmit_data (th->master);
+  }
 }
 
 
@@ -938,8 +978,8 @@ GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel,
   slvadd->header.size = htons (sizeof (*slvadd));
   slvadd->announced_at = GNUNET_htonll (announced_at);
   slvadd->effective_since = GNUNET_htonll (effective_since);
-  GNUNET_CONTAINER_DLL_insert_tail (channel->transmit_head,
-                                    channel->transmit_tail,
+  GNUNET_CONTAINER_DLL_insert_tail (channel->tmit_head,
+                                    channel->tmit_tail,
                                     op);
   transmit_next (channel);
 }
@@ -979,8 +1019,8 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
   slvrm->header.type = GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM;
   slvrm->header.size = htons (sizeof (*slvrm));
   slvrm->announced_at = GNUNET_htonll (announced_at);
-  GNUNET_CONTAINER_DLL_insert_tail (channel->transmit_head,
-                                    channel->transmit_tail,
+  GNUNET_CONTAINER_DLL_insert_tail (channel->tmit_head,
+                                    channel->tmit_tail,
                                     op);
   transmit_next (channel);
 }
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index 41c4ca8e4..2986fdf6a 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -144,10 +144,12 @@ join (void *cls, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key,
   return GNUNET_OK;
 }
 
+
 struct TransmitClosure
 {
   struct GNUNET_PSYC_MasterTransmitHandle *handle;
   uint8_t n;
+  uint8_t paused;
   uint8_t fragment_count;
   char *fragments[16];
   uint16_t fragment_sizes[16];
@@ -157,8 +159,9 @@ struct TransmitClosure
 static void
 transmit_resume (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Transmit resume\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission resumed.\n");
   struct TransmitClosure *tmit = cls;
+  tmit->paused = GNUNET_NO;
   GNUNET_PSYC_master_transmit_resume (tmit->handle);
 }
 
@@ -167,33 +170,36 @@ static int
 transmit_notify (void *cls, size_t *data_size, void *data)
 {
   struct TransmitClosure *tmit = cls;
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-              "Transmit notify: %lu bytes\n", *data_size);
-
-  if (tmit->fragment_count <= tmit->n)
-    return GNUNET_YES;
-
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Transmit notify: %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);
 
-  *data_size = tmit->fragment_sizes[tmit->n];
-  memcpy (data, tmit->fragments[tmit->n], *data_size);
-  tmit->n++;
-
-  if (tmit->n == tmit->fragment_count - 1)
+  if (GNUNET_YES == tmit->paused && tmit->n == tmit->fragment_count - 1)
   {
     /* Send last fragment later. */
-    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS, &transmit_resume,
-                                  tmit);
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Transmission paused.\n");
+    tmit->paused = GNUNET_YES;
+    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
+                                  (GNUNET_TIME_UNIT_SECONDS, 3),
+                                  &transmit_resume, tmit);
     *data_size = 0;
     return GNUNET_NO;
   }
-  return tmit->n <= tmit->fragment_count ? GNUNET_NO : GNUNET_YES;
+
+  GNUNET_assert (tmit->fragment_sizes[tmit->n] <= *data_size);
+  *data_size = tmit->fragment_sizes[tmit->n];
+  memcpy (data, tmit->fragments[tmit->n], *data_size);
+
+  return ++tmit->n < tmit->fragment_count ? GNUNET_NO : GNUNET_YES;
 }
 
 void
 master_started (void *cls, uint64_t max_message_id)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Master started: %lu\n", max_message_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Master started: %lu\n", max_message_id);
 
   struct GNUNET_ENV_Environment *env = GNUNET_ENV_environment_create ();
   GNUNET_ENV_environment_add_mod (env, GNUNET_ENV_OP_ASSIGN,
@@ -202,11 +208,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 = 2;
-  tmit->fragments[0] = "foo bar";
-  tmit->fragment_sizes[0] = 7;
-  tmit->fragments[1] = "baz!";
-  tmit->fragment_sizes[1] = 4;
+  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->handle
     = GNUNET_PSYC_master_transmit (mst, "_test", env, transmit_notify, tmit,
                                    GNUNET_PSYC_MASTER_TRANSMIT_INC_GROUP_GEN);