towards having sending in transport-testing API (not yet flexible enough)

5 files changed, 613 insertions, 154 deletions

diff --git a/src/transport/Makefile.am b/src/transport/Makefile.am
index 767031e6f5..d6c4e3ccf5 100644
--- a/src/transport/Makefile.am
+++ b/src/transport/Makefile.am
@@ -153,7 +153,8 @@ libgnunettransporttesting_la_SOURCES = \
   transport-testing.c transport-testing.h \
   transport-testing-filenames.c \
   transport-testing-loggers.c \
-  transport-testing-main.c
+  transport-testing-main.c \
+  transport-testing-send.c
 libgnunettransporttesting_la_LIBADD = \
   libgnunettransport.la \
   $(top_builddir)/src/hello/libgnunethello.la \
diff --git a/src/transport/test_transport_api.c b/src/transport/test_transport_api.c
index bc0061363a..4da4f25332 100644
--- a/src/transport/test_transport_api.c
+++ b/src/transport/test_transport_api.c
@@ -36,31 +36,8 @@
  */
 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 30)
 
-/**
- * How long until we give up on transmitting the message?
- */
-#define TIMEOUT_TRANSMIT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
-
-#define TEST_MESSAGE_SIZE 2600
-
-#define TEST_MESSAGE_TYPE 12345
-
-
 static struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext *ccc;
 
-static struct GNUNET_TRANSPORT_TransmitHandle *th;
-
-
-static void
-custom_shutdown (void *cls)
-{
-  if (NULL != th)
-  {
-    GNUNET_TRANSPORT_notify_transmit_ready_cancel (th);
-    th = NULL;
-  }
-}
-
 
 static void
 notify_receive (void *cls,
@@ -81,8 +58,8 @@ notify_receive (void *cls,
     GNUNET_free (ps);
   }
 
-  if ((TEST_MESSAGE_TYPE == ntohs (message->type)) &&
-      (TEST_MESSAGE_SIZE == ntohs (message->size)))
+  if ((GNUNET_TRANSPORT_TESTING_SIMPLE_MTYPE == ntohs (message->type)) &&
+      (GNUNET_TRANSPORT_TESTING_LARGE_MESSAGE_SIZE == ntohs (message->size)))
   {
     ccc->global_ret = GNUNET_OK;
     GNUNET_SCHEDULER_shutdown ();
@@ -96,90 +73,6 @@ notify_receive (void *cls,
 }
 
 
-static size_t
-notify_ready (void *cls,
-              size_t size,
-              void *buf)
-{
-  struct GNUNET_TRANSPORT_TESTING_PeerContext *p = cls;
-  struct GNUNET_MessageHeader *hdr;
-
-  th = NULL;
-  if (buf == NULL)
-  {
-    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
-                "Timeout occurred while waiting for transmit_ready\n");
-    GNUNET_SCHEDULER_shutdown ();
-    ccc->global_ret = 42;
-    return 0;
-  }
-
-  GNUNET_assert (size >= TEST_MESSAGE_SIZE);
-  if (NULL != buf)
-  {
-    memset (buf, '\0', TEST_MESSAGE_SIZE);
-    hdr = buf;
-    hdr->size = htons (TEST_MESSAGE_SIZE);
-    hdr->type = htons (TEST_MESSAGE_TYPE);
-  }
-
-  {
-    char *ps = GNUNET_strdup (GNUNET_i2s (&ccc->p[1]->id));
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Peer %u (`%4s') sending message with type %u and size %u bytes to peer %u (`%4s')\n",
-                ccc->p[1]->no,
-                ps,
-                ntohs (hdr->type),
-                ntohs (hdr->size),
-                p->no,
-                GNUNET_i2s (&p->id));
-    GNUNET_free (ps);
-  }
-  return TEST_MESSAGE_SIZE;
-}
-
-
-static void
-sendtask (void *cls)
-{
-  {
-    char *receiver_s = GNUNET_strdup (GNUNET_i2s (&ccc->p[0]->id));
-
-    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-                "Sending message from peer %u (`%s') -> peer %u (`%s') !\n",
-                ccc->p[1]->no,
-                GNUNET_i2s (&ccc->p[1]->id),
-                ccc->p[0]->no,
-                receiver_s);
-    GNUNET_free (receiver_s);
-  }
-  ccc->global_ret = GNUNET_SYSERR;
-  th = GNUNET_TRANSPORT_notify_transmit_ready (ccc->p[1]->th,
-                                               &ccc->p[0]->id,
-                                               TEST_MESSAGE_SIZE,
-                                               TIMEOUT_TRANSMIT,
-                                               &notify_ready,
-                                               ccc->p[0]);
-  GNUNET_assert (NULL != th);
-}
-
-
-static void
-notify_disconnect (void *cls,
-                   struct GNUNET_TRANSPORT_TESTING_PeerContext *me,
-                   const struct GNUNET_PeerIdentity *other)
-{
-  GNUNET_TRANSPORT_TESTING_log_disconnect (cls,
-                                           me,
-                                           other);
-  if (NULL != th)
-  {
-    GNUNET_TRANSPORT_notify_transmit_ready_cancel (th);
-    th = NULL;
-  }
-}
-
-
 /**
  * Runs the test.
  *
@@ -192,12 +85,12 @@ test (char *argv[],
       int bi_directional)
 {
   struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext my_ccc = {
-    .connect_continuation = &sendtask,
+    .connect_continuation = &GNUNET_TRANSPORT_TESTING_large_send,
+    .connect_continuation_cls = &my_ccc,
     .config_file = "test_transport_api_data.conf",
     .rec = &notify_receive,
     .nc = &GNUNET_TRANSPORT_TESTING_log_connect,
-    .nd = &notify_disconnect,
-    .shutdown_task = &custom_shutdown,
+    .nd = &GNUNET_TRANSPORT_TESTING_log_disconnect,
     .timeout = TIMEOUT,
     .bi_directional = bi_directional
   };
diff --git a/src/transport/transport-testing-send.c b/src/transport/transport-testing-send.c
new file mode 100644
index 0000000000..20d1f2fbc8
--- /dev/null
+++ b/src/transport/transport-testing-send.c
@@ -0,0 +1,241 @@
+/*
+     This file is part of GNUnet.
+     Copyright (C) 2016 GNUnet e.V.
+
+     GNUnet is free software; you can redistribute it and/or modify
+     it under the terms of the GNU General Public License as published
+     by the Free Software Foundation; either version 3, or (at your
+     option) any later version.
+
+     GNUnet is distributed in the hope that it will be useful, but
+     WITHOUT ANY WARRANTY; without even the implied warranty of
+     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+     General Public License for more details.
+
+     You should have received a copy of the GNU General Public License
+     along with GNUnet; see the file COPYING.  If not, write to the
+     Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
+     Boston, MA 02110-1301, USA.
+*/
+/**
+ * @file transport-testing-send.c
+ * @brief convenience transmission function for tests
+ * @author Christian Grothoff
+ */
+#include "transport-testing.h"
+
+/**
+ * Acceptable transmission delay.
+ */
+#define TIMEOUT_TRANSMIT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5)
+
+
+static size_t
+notify_ready (void *cls,
+              size_t size,
+              void *buf)
+{ 
+  struct TRANSPORT_TESTING_SendJob *sj = cls;
+  struct GNUNET_TRANSPORT_TESTING_PeerContext *sender = sj->sender;
+  struct GNUNET_TRANSPORT_TESTING_PeerContext *receiver = sj->receiver;
+  struct GNUNET_TRANSPORT_TESTING_Handle *tth = sender->tth;
+  uint16_t msize = sj->msize;
+  struct GNUNET_TRANSPORT_TESTING_TestMessage *test;
+
+  sj->th = NULL;
+  GNUNET_CONTAINER_DLL_remove (tth->sj_head,
+			       tth->sj_tail,
+			       sj);
+  if (NULL == buf)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                "Timeout occurred while waiting for transmit_ready\n");
+    GNUNET_SCHEDULER_shutdown ();
+    GNUNET_free (sj);
+    return 0;
+  }
+
+  GNUNET_assert (size >= msize);
+  if (NULL != buf)
+  {
+    memset (buf, '\0', msize);
+    test = buf;
+    test->header.size = htons (msize);
+    test->header.type = htons (sj->mtype);
+    test->num = htonl (sj->num);
+  }
+
+  {
+    char *ps = GNUNET_strdup (GNUNET_i2s (&sender->id));
+
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "Sending message %u from %u (%s) with type %u and size %u bytes to peer %u (%s)\n",
+		(unsigned int) sj->num,
+		sender->no,
+		ps,
+                sj->mtype,
+                msize,
+                receiver->no,
+                GNUNET_i2s (&receiver->id));
+    GNUNET_free (ps);
+  }
+  GNUNET_free (sj);
+  return msize;
+}
+
+
+/**
+ * Return @a cx in @a cls.
+ */
+static void
+find_cr (void *cls,   
+	 struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cx)
+{
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest **cr = cls;
+
+  *cr = cx;
+}
+
+
+/**
+ * Send a test message of type @a mtype and size @a msize from
+ * peer @a sender to peer @a receiver.  The peers should be
+ * connected when this function is called.
+ *
+ * @param sender the sending peer
+ * @param receiver the receiving peer
+ * @param mtype message type to use
+ * @param msize size of the message, at least `sizeof (struct GNUNET_TRANSPORT_TESTING_TestMessage)`
+ * @param num unique message number
+ * @return #GNUNET_OK if message was queued,
+ *         #GNUNET_NO if peers are not connected
+ *         #GNUNET_SYSERR if @a msize is illegal
+ */
+int
+GNUNET_TRANSPORT_TESTING_send (struct GNUNET_TRANSPORT_TESTING_PeerContext *sender,
+			       struct GNUNET_TRANSPORT_TESTING_PeerContext *receiver,
+			       uint16_t mtype,
+			       uint16_t msize,
+			       uint32_t num)
+{
+  struct GNUNET_TRANSPORT_TESTING_Handle *tth = sender->tth;
+  struct TRANSPORT_TESTING_SendJob *sj;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cr;
+
+  if (msize < sizeof (struct GNUNET_TRANSPORT_TESTING_TestMessage))
+  {
+    GNUNET_break (0);
+    return GNUNET_SYSERR;
+  }
+  cr = NULL;
+  GNUNET_TRANSPORT_TESTING_find_connecting_context (sender,
+						    receiver,
+						    &find_cr,
+						    &cr);
+  if (NULL == cr)
+    GNUNET_TRANSPORT_TESTING_find_connecting_context (receiver,
+						      sender,
+						      &find_cr,
+						      &cr);
+  if ( (NULL == cr) ||
+       (GNUNET_YES != cr->connected) )
+  {
+    GNUNET_break (0);
+    return GNUNET_NO;
+  }
+  sj = GNUNET_new (struct TRANSPORT_TESTING_SendJob);
+  sj->num = num;
+  sj->sender = sender;
+  sj->receiver = receiver;
+  sj->mtype = mtype;
+  sj->msize = msize;
+  GNUNET_CONTAINER_DLL_insert (tth->sj_head,
+			       tth->sj_tail,
+			       sj);
+  {
+    char *receiver_s = GNUNET_strdup (GNUNET_i2s (&receiver->id));
+
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "Sending message from peer %u (`%s') -> peer %u (`%s') !\n",
+                sender->no,
+                GNUNET_i2s (&sender->id),
+                receiver->no,
+                receiver_s);
+    GNUNET_free (receiver_s);
+  }
+  sj->th = GNUNET_TRANSPORT_notify_transmit_ready (sender->th,
+						   &receiver->id,
+						   msize,
+						   TIMEOUT_TRANSMIT,
+						   &notify_ready,
+						   sj);
+  GNUNET_assert (NULL != sj->th);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Task that sends a test message from the 
+ * first peer to the second peer.
+ *
+ * @param ccc context which should contain at least two peers, the
+ *        first two of which should be currently connected
+ */
+static void
+do_send (struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext *ccc,
+	 uint16_t size)
+{
+  int ret;
+
+  ccc->global_ret = GNUNET_SYSERR;
+  ret = GNUNET_TRANSPORT_TESTING_send (ccc->p[0],
+				       ccc->p[1],
+				       GNUNET_TRANSPORT_TESTING_SIMPLE_MTYPE,
+				       size,
+				       ccc->send_num_gen++);
+  GNUNET_assert (GNUNET_SYSERR != ret);
+  if (GNUNET_NO == ret)
+  {
+    GNUNET_break (0);
+    ccc->global_ret = GNUNET_SYSERR;
+    GNUNET_SCHEDULER_shutdown ();
+  }
+}
+
+
+/**
+ * Task that sends a minimalistic test message from the 
+ * first peer to the second peer.
+ *
+ * @param cls the `struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext`
+ *        which should contain at least two peers, the first two
+ *        of which should be currently connected
+ */
+void
+GNUNET_TRANSPORT_TESTING_simple_send (void *cls)
+{
+  struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext *ccc = cls;
+
+  do_send (ccc,
+	   sizeof (struct GNUNET_MessageHeader));
+}
+
+
+/**
+ * Task that sends a large test message from the 
+ * first peer to the second peer.
+ *
+ * @param cls the `struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext`
+ *        which should contain at least two peers, the first two
+ *        of which should be currently connected
+ */
+void
+GNUNET_TRANSPORT_TESTING_large_send (void *cls)
+{
+  struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext *ccc = cls;
+
+  do_send (ccc,
+	   2600);
+}
+
+/* end of transport-testing-send.c */
diff --git a/src/transport/transport-testing.c b/src/transport/transport-testing.c
index 3226274591..50c2973891 100644
--- a/src/transport/transport-testing.c
+++ b/src/transport/transport-testing.c
@@ -44,11 +44,21 @@ find_peer_context (struct GNUNET_TRANSPORT_TESTING_Handle *tth,
 }
 
 
-static void
-notify_connecting_context (struct GNUNET_TRANSPORT_TESTING_Handle *tth,
-                           struct GNUNET_TRANSPORT_TESTING_PeerContext *p1,
-                           struct GNUNET_TRANSPORT_TESTING_PeerContext *p2)
+/**
+ * Find any connecting context matching the given pair of peers.
+ *
+ * @param p1 first peer
+ * @param p2 second peer
+ * @param cb function to call 
+ * @param cb_cls closure for @a cb
+ */
+void
+GNUNET_TRANSPORT_TESTING_find_connecting_context (struct GNUNET_TRANSPORT_TESTING_PeerContext *p1,
+						  struct GNUNET_TRANSPORT_TESTING_PeerContext *p2,
+						  GNUNET_TRANSPORT_TESTING_ConnectContextCallback cb,
+						  void *cb_cls)
 {
+  struct GNUNET_TRANSPORT_TESTING_Handle *tth = p1->tth;
   struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc;
   struct GNUNET_TRANSPORT_TESTING_ConnectRequest *ccn;
 
@@ -56,28 +66,55 @@ notify_connecting_context (struct GNUNET_TRANSPORT_TESTING_Handle *tth,
   {
     ccn = cc->next;
     if ( (cc->p1 == p1) &&
-         (cc->p2 == p2) )
-      cc->p1_c = GNUNET_YES;
-    if ( (cc->p1 == p2) &&
-         (cc->p2 == p1) )
-      cc->p2_c = GNUNET_YES;
-    if ( (cc->p1_c == GNUNET_YES) &&
-         (cc->p2_c == GNUNET_YES) )
-    {
-      cc->cb (cc->cb_cls);
-      GNUNET_TRANSPORT_TESTING_connect_peers_cancel (cc);
-    }
+	 (cc->p2 == p2) ) 
+      cb (cb_cls,
+	  cc);
   }
 }
 
 
 static void
+set_p1c (void *cls,   
+	 struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cx)
+{
+  cx->p1_c = GNUNET_YES;
+}
+
+
+static void
+set_p2c (void *cls,   
+	 struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cx)
+{
+  cx->p2_c = GNUNET_YES;
+}
+
+
+static void
+clear_p1c (void *cls,   
+	   struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cx)
+{
+  cx->p1_c = GNUNET_NO;
+}
+
+
+static void
+clear_p2c (void *cls,   
+	 struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cx)
+{
+  cx->p2_c = GNUNET_NO;
+}
+
+
+static void
 notify_connect (void *cls,
                 const struct GNUNET_PeerIdentity *peer)
 {
   struct GNUNET_TRANSPORT_TESTING_PeerContext *p = cls;
+  struct GNUNET_TRANSPORT_TESTING_Handle *tth = p->tth;
   char *p2_s;
   struct GNUNET_TRANSPORT_TESTING_PeerContext *p2;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *ccn;
 
   p2 = find_peer_context (p->tth,
                           peer);
@@ -100,30 +137,83 @@ notify_connect (void *cls,
        p->no,
        GNUNET_i2s (&p->id));
   GNUNET_free (p2_s);
-  notify_connecting_context (p->tth,
-                             p,
-                             p2);
+  /* update flags in connecting contexts */
+  GNUNET_TRANSPORT_TESTING_find_connecting_context (p,
+						    p2,
+						    &set_p1c,
+						    NULL);
+  GNUNET_TRANSPORT_TESTING_find_connecting_context (p2,
+						    p,
+						    &set_p2c,
+						    NULL);
+  /* update set connected flag for all requests */
+  for (cc = tth->cc_head; NULL != cc; cc = cc->next)
+  {
+    if (GNUNET_YES == cc->connected)
+      continue;
+    if ( (GNUNET_YES == cc->p1_c) &&
+         (GNUNET_YES == cc->p2_c) )
+    {
+      cc->connected = GNUNET_YES;
+      /* stop trying to connect */
+      if (NULL != cc->tct)
+      {
+	GNUNET_SCHEDULER_cancel (cc->tct);
+	cc->tct = NULL;
+      }
+      if (NULL != cc->oh)
+      {
+	GNUNET_TRANSPORT_offer_hello_cancel (cc->oh);
+	cc->oh = NULL;
+      }
+      if (NULL != cc->ats_sh)
+      {
+	GNUNET_ATS_connectivity_suggest_cancel (cc->ats_sh);
+	cc->ats_sh = NULL;
+      }
+    }
+  }
+  /* then notify application */
+  for (cc = tth->cc_head; NULL != cc; cc = ccn)
+  {
+    ccn = cc->next;
+    if ( (GNUNET_YES == cc->connected) &&
+         (NULL != cc->cb) )
+    {
+      cc->cb (cc->cb_cls);
+      cc->cb = NULL; /* only notify once! */
+    }
+  }
 }
 
 
+/**
+ * Offer the current HELLO of P2 to P1.
+ *
+ * @param cls our `struct GNUNET_TRANSPORT_TESTING_ConnectRequest`
+ */
+static void
+offer_hello (void *cls);
+
+
 static void
 notify_disconnect (void *cls,
                    const struct GNUNET_PeerIdentity *peer)
 {
   struct GNUNET_TRANSPORT_TESTING_PeerContext *p = cls;
+  struct GNUNET_TRANSPORT_TESTING_Handle *tth = p->tth;
   char *p2_s;
   /* Find PeerContext */
   int no = 0;
   struct GNUNET_TRANSPORT_TESTING_PeerContext *p2 = NULL;
-
-  if (NULL != p)
-  {
-    p2 = find_peer_context (p->tth,
-                            peer);
-    no = p->no;
-  }
-
-  if (p2 != NULL)
+  struct TRANSPORT_TESTING_SendJob *sj;
+  struct TRANSPORT_TESTING_SendJob *sjn;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc;
+  
+  p2 = find_peer_context (p->tth,
+			  peer);
+  no = p->no;
+  if (NULL != p2)
     GNUNET_asprintf (&p2_s,
                      "%u (`%s')",
                      p2->no,
@@ -138,12 +228,57 @@ notify_disconnect (void *cls,
        no,
        GNUNET_i2s (&p->id));
   GNUNET_free (p2_s);
-
-  if (NULL == p)
-    return;
+  /* notify about disconnect */
   if (NULL != p->nd)
     p->nd (p->cb_cls,
            peer);
+  if (NULL == p2)
+    return;
+  /* abort all transmissions this disconnected pair is involved in */
+  for (sj = tth->sj_head; NULL != sj; sj = sjn)
+  {
+    sjn = sj->next;
+    if ( ( (sj->sender == p2) &&
+	   (sj->receiver == p) ) ||
+	 ( (sj->receiver == p2) &&
+	   (sj->sender == p) ) )
+    {
+      GNUNET_CONTAINER_DLL_remove (tth->sj_head,
+				   tth->sj_tail,
+				   sj);
+      GNUNET_TRANSPORT_notify_transmit_ready_cancel (sj->th);
+      GNUNET_free (sj);
+    }
+  }
+  /* update set connected flags for all requests */
+  GNUNET_TRANSPORT_TESTING_find_connecting_context (p,
+						    p2,
+						    &clear_p1c,
+						    NULL);
+  GNUNET_TRANSPORT_TESTING_find_connecting_context (p2,
+						    p,
+						    &clear_p2c,
+						    NULL);
+  /* resume connectivity requests as necessary */
+  for (cc = tth->cc_head; NULL != cc; cc = cc->next)
+  {
+    if (GNUNET_NO == cc->connected)
+      continue;
+    if ( (GNUNET_YES != cc->p1_c) ||
+         (GNUNET_YES != cc->p2_c) )
+    {
+      cc->connected = GNUNET_NO;
+      /* start trying to connect */
+      if ( (NULL == cc->tct) &&
+	   (NULL == cc->oh) )
+	cc->tct = GNUNET_SCHEDULER_add_now (&offer_hello,
+					    cc);
+      if (NULL == cc->ats_sh)
+	cc->ats_sh = GNUNET_ATS_connectivity_suggest (cc->p1->ats,
+						      &p2->id,
+						      1);
+    }
+  }
 }
 
 
@@ -352,6 +487,9 @@ GNUNET_TRANSPORT_TESTING_restart_peer (struct GNUNET_TRANSPORT_TESTING_PeerConte
                                        GNUNET_TRANSPORT_TESTING_StartCallback restart_cb,
                                        void *restart_cb_cls)
 {
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *ccn;
+ 
   /* shutdown */
   LOG (GNUNET_ERROR_TYPE_DEBUG,
        "Stopping peer %u (`%s')\n",
@@ -367,6 +505,13 @@ GNUNET_TRANSPORT_TESTING_restart_peer (struct GNUNET_TRANSPORT_TESTING_PeerConte
     GNUNET_TRANSPORT_disconnect (p->th);
     p->th = NULL;
   }
+  for (cc = p->tth->cc_head; NULL != cc; cc = ccn)
+  {
+    ccn = cc->next;
+    if ( (cc->p1 == p) ||
+         (cc->p2 == p) )
+      GNUNET_TRANSPORT_TESTING_connect_peers_cancel (cc);
+  }
   if (NULL != p->ats)
   {
     GNUNET_ATS_connectivity_done (p->ats);
@@ -427,7 +572,16 @@ void
 GNUNET_TRANSPORT_TESTING_stop_peer (struct GNUNET_TRANSPORT_TESTING_PeerContext *p)
 {
   struct GNUNET_TRANSPORT_TESTING_Handle *tth = p->tth;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc;
+  struct GNUNET_TRANSPORT_TESTING_ConnectRequest *ccn;
 
+  for (cc = tth->cc_head; NULL != cc; cc = ccn)
+  {
+    ccn = cc->next;
+    if ( (cc->p1 == p) ||
+         (cc->p2 == p) )
+      GNUNET_TRANSPORT_TESTING_connect_peers_cancel (cc);
+  }
   if (NULL != p->ghh)
   {
     GNUNET_TRANSPORT_get_hello_cancel (p->ghh);
@@ -478,15 +632,6 @@ GNUNET_TRANSPORT_TESTING_stop_peer (struct GNUNET_TRANSPORT_TESTING_PeerContext
 
 
 /**
- * Offer the current HELLO of P2 to P1.
- *
- * @param cls our `struct GNUNET_TRANSPORT_TESTING_ConnectRequest`
- */
-static void
-offer_hello (void *cls);
-
-
-/**
  * Function called after the HELLO was passed to the
  * transport service.
  */
diff --git a/src/transport/transport-testing.h b/src/transport/transport-testing.h
index 4be451d352..e61621b2ee 100644
--- a/src/transport/transport-testing.h
+++ b/src/transport/transport-testing.h
@@ -207,8 +207,69 @@ struct GNUNET_TRANSPORT_TESTING_ConnectRequest
    */
   void *cb_cls;
 
-  int p1_c; // dead?
-  int p2_c; // dead?
+  /** 
+   * Set if peer1 says the connection is up to peer2.
+   */
+  int p1_c;
+
+  /** 
+   * Set if peer2 says the connection is up to peer1.
+   */
+  int p2_c;
+
+  /**
+   * #GNUNET_YES if both @e p1_c and @e p2_c are #GNUNET_YES.
+   */
+  int connected;
+};
+
+
+/**
+ * Information we keep for active transmission jobs.
+ */
+struct TRANSPORT_TESTING_SendJob
+{
+
+  /**
+   * Kept in a DLL.
+   */ 
+  struct TRANSPORT_TESTING_SendJob *next;
+
+  /**
+   * Kept in a DLL.
+   */ 
+  struct TRANSPORT_TESTING_SendJob *prev;
+
+  /**
+   * Sender of the message.
+   */
+  struct GNUNET_TRANSPORT_TESTING_PeerContext *sender;
+
+  /** 
+   * Receiver of the message.
+   */ 
+  struct GNUNET_TRANSPORT_TESTING_PeerContext *receiver;
+
+  /**
+   * Operation handle.
+   */
+  struct GNUNET_TRANSPORT_TransmitHandle *th;
+
+  /**
+   * Number of the message.
+   */ 
+  uint32_t num;
+  
+  /**
+   * Type of message to send.
+   */ 
+  uint16_t mtype;
+
+  /**
+   * Length of the message.
+   */
+  uint16_t msize;
+  
 };
 
 
@@ -233,6 +294,16 @@ struct GNUNET_TRANSPORT_TESTING_Handle
   struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc_tail;
 
   /**
+   * Kept in a DLL.
+   */ 
+  struct TRANSPORT_TESTING_SendJob *sj_head;
+
+  /**
+   * Kept in a DLL.
+   */ 
+  struct TRANSPORT_TESTING_SendJob *sj_tail;
+  
+  /**
    * head DLL of peers
    */
   struct GNUNET_TRANSPORT_TESTING_PeerContext *p_head;
@@ -341,6 +412,32 @@ void
 GNUNET_TRANSPORT_TESTING_connect_peers_cancel (struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc);
 
 
+/**
+ * Function called on matching connect requests.
+ *
+ * @param cls closure
+ * @param cc request matching the query
+ */
+typedef void
+(*GNUNET_TRANSPORT_TESTING_ConnectContextCallback)(void *cls,
+						   struct GNUNET_TRANSPORT_TESTING_ConnectRequest *cc);
+
+
+/**
+ * Find any connecting context matching the given pair of peers.
+ *
+ * @param p1 first peer
+ * @param p2 second peer
+ * @param cb function to call 
+ * @param cb_cls closure for @a cb
+ */
+void
+GNUNET_TRANSPORT_TESTING_find_connecting_context (struct GNUNET_TRANSPORT_TESTING_PeerContext *p1,
+						  struct GNUNET_TRANSPORT_TESTING_PeerContext *p2,
+						  GNUNET_TRANSPORT_TESTING_ConnectContextCallback cb,
+						  void *cb_cls);
+
+
 /* ********************** high-level process functions *************** */
 
 
@@ -531,6 +628,14 @@ struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext
    */
   int global_ret;
 
+  /**
+   * Generator for the `num` field in test messages.  Incremented each
+   * time #GNUNET_TRANSPORT_TESTING_simple_send or
+   * #GNUNET_TRANSPORT_TESTING_large_send are used to transmit a
+   * message.
+   */
+  uint32_t send_num_gen;
+  
   /* ******* internal state, clients should not mess with this **** */
 
   /**
@@ -652,6 +757,80 @@ GNUNET_TRANSPORT_TESTING_main_ (const char *argv0,
 #define GNUNET_TRANSPORT_TESTING_main(num_peers,check,check_cls) \
   GNUNET_TRANSPORT_TESTING_main_ (argv[0], __FILE__, num_peers, check, check_cls)
 
+/* ***************** Convenience functions for sending ********* */
+
+
+/**
+ * Send a test message of type @a mtype and size @a msize from
+ * peer @a sender to peer @a receiver.  The peers should be
+ * connected when this function is called.
+ *
+ * @param sender the sending peer
+ * @param receiver the receiving peer
+ * @param mtype message type to use
+ * @param msize size of the message, at least `sizeof (struct GNUNET_TRANSPORT_TESTING_TestMessage)`
+ * @param num unique message number
+ * @return #GNUNET_OK if message was queued,
+ *         #GNUNET_NO if peers are not connected
+ *         #GNUNET_SYSERR if @a msize is illegal
+ */
+int
+GNUNET_TRANSPORT_TESTING_send (struct GNUNET_TRANSPORT_TESTING_PeerContext *sender,
+			       struct GNUNET_TRANSPORT_TESTING_PeerContext *receiver,
+			       uint16_t mtype,
+			       uint16_t msize,
+			       uint32_t num);
+
+
+/**
+ * Message type used by #GNUNET_TRANSPORT_TESTING_simple_send().
+ */
+#define GNUNET_TRANSPORT_TESTING_SIMPLE_MTYPE 12345
+
+GNUNET_NETWORK_STRUCT_BEGIN
+struct GNUNET_TRANSPORT_TESTING_TestMessage
+{
+  /**
+   * Type is #GNUNET_TRANSPORT_TESTING_SIMPLE_MTYPE.
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Monotonically increasing counter throughout the test.
+   */
+  uint32_t num GNUNET_PACKED;
+};
+GNUNET_NETWORK_STRUCT_END
+
+
+/**
+ * Task that sends a minimalistic test message from the 
+ * first peer to the second peer.
+ *
+ * @param cls the `struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext`
+ *        which should contain at least two peers, the first two
+ *        of which should be currently connected
+ */
+void
+GNUNET_TRANSPORT_TESTING_simple_send (void *cls);
+
+/**
+ * Size of a message sent with 
+ * #GNUNET_TRANSPORT_TESTING_large_send().  Big enough
+ * to usually force defragmentation.
+ */
+#define GNUNET_TRANSPORT_TESTING_LARGE_MESSAGE_SIZE 2600
+
+/**
+ * Task that sends a large test message from the 
+ * first peer to the second peer.
+ *
+ * @param cls the `struct GNUNET_TRANSPORT_TESTING_ConnectCheckContext`
+ *        which should contain at least two peers, the first two
+ *        of which should be currently connected
+ */
+void
+GNUNET_TRANSPORT_TESTING_large_send (void *cls);
 
 
 /* ********************** log-only convenience functions ************* */