more work on new CADET

12 files changed, 605 insertions, 48 deletions

diff --git a/src/cadet/cadet_protocol.h b/src/cadet/cadet_protocol.h
index c555d6155..ba2044b0d 100644
--- a/src/cadet/cadet_protocol.h
+++ b/src/cadet/cadet_protocol.h
@@ -74,7 +74,7 @@ struct GNUNET_CADET_ConnectionCreate
   /**
    * ID of the connection
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 
   /**
    * path_length structs defining the *whole* path from the origin [0] to the
@@ -102,7 +102,7 @@ struct GNUNET_CADET_ConnectionACK
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 
 };
 
@@ -125,7 +125,7 @@ struct GNUNET_CADET_ConnectionBroken
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 
   /**
    * ID of the endpoint
@@ -157,7 +157,7 @@ struct GNUNET_CADET_ConnectionDestroy
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 };
 
 
@@ -179,7 +179,7 @@ struct GNUNET_CADET_ACK
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 };
 
 
@@ -201,7 +201,7 @@ struct GNUNET_CADET_Poll
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 
 };
 
@@ -247,7 +247,7 @@ struct GNUNET_CADET_KX
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 
   /**
    * Sender's ephemeral public ECC key encoded in a
@@ -283,7 +283,7 @@ struct GNUNET_CADET_Encrypted
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
 
   /**
    * MAC of the encrypted message, used to verify message integrity.
diff --git a/src/cadet/gnunet-service-cadet-new.c b/src/cadet/gnunet-service-cadet-new.c
index 74296a27e..b2f39b3cd 100644
--- a/src/cadet/gnunet-service-cadet-new.c
+++ b/src/cadet/gnunet-service-cadet-new.c
@@ -170,6 +170,11 @@ struct GNUNET_CONTAINER_MultiHashMap *loose_channels;
  */
 struct GNUNET_CONTAINER_MultiPeerMap *peers;
 
+/**
+ * Map from expanded connection hash codes to `struct CadetConnection` objects.
+ */
+struct GNUNET_CONTAINER_MultiHashMap *connections;
+
 
 
 /**
@@ -317,6 +322,11 @@ shutdown_task (void *cls)
     GNUNET_CONTAINER_multipeermap_destroy (peers);
     peers = NULL;
   }
+  if (NULL != connections)
+  {
+    GNUNET_CONTAINER_multihashmap_destroy (connections);
+    connections = NULL;
+  }
   if (NULL != ats_ch)
   {
     GNUNET_ATS_connectivity_done (ats_ch);
@@ -1232,6 +1242,8 @@ run (void *cls,
                                                          GNUNET_NO);
   peers = GNUNET_CONTAINER_multipeermap_create (16,
                                                 GNUNET_YES);
+  connections = GNUNET_CONTAINER_multihashmap_create (256,
+                                                      GNUNET_YES);
   GCH_init (c);
   GCD_init (c);
   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
diff --git a/src/cadet/gnunet-service-cadet-new.h b/src/cadet/gnunet-service-cadet-new.h
index 9521b6363..416c9d13a 100644
--- a/src/cadet/gnunet-service-cadet-new.h
+++ b/src/cadet/gnunet-service-cadet-new.h
@@ -140,6 +140,11 @@ extern struct GNUNET_CRYPTO_EddsaPrivateKey *my_private_key;
 extern struct GNUNET_CONTAINER_MultiHashMap *open_ports;
 
 /**
+ * Map from expanded connection hash codes to `struct CadetConnection` objects.
+ */
+extern struct GNUNET_CONTAINER_MultiHashMap *connections;
+
+/**
  * Map from ports to channels where the ports were closed at the
  * time we got the inbound connection.
  * Indexed by port, contains `struct CadetChannel`.
diff --git a/src/cadet/gnunet-service-cadet-new_connection.c b/src/cadet/gnunet-service-cadet-new_connection.c
index ea0b2c6e5..1f31331a5 100644
--- a/src/cadet/gnunet-service-cadet-new_connection.c
+++ b/src/cadet/gnunet-service-cadet-new_connection.c
@@ -30,6 +30,45 @@
 #include "gnunet-service-cadet-new_paths.h"
 #include "gnunet-service-cadet-new_peer.h"
 #include "gnunet-service-cadet-new_connection.h"
+#include "gnunet_cadet_service.h"
+#include "cadet_protocol.h"
+
+
+/**
+ * All the states a connection can be in.
+ */
+enum CadetConnectionState
+{
+  /**
+   * Uninitialized status, we have not yet even gotten the message queue.
+   */
+  CADET_CONNECTION_NEW,
+
+  /**
+   * Connection create message in queue, awaiting transmission by CORE.
+   */
+  CADET_CONNECTION_SENDING_CREATE,
+
+  /**
+   * Connection create message sent, waiting for ACK.
+   */
+  CADET_CONNECTION_SENT,
+
+  /**
+   * Connection confirmed, ready to carry traffic.
+   */
+  CADET_CONNECTION_READY,
+
+  /**
+   * Connection to be destroyed, just waiting to empty queues.
+   */
+  CADET_CONNECTION_DESTROYED,
+
+  /**
+   * Connection to be destroyed because of a distant peer, same as DESTROYED.
+   */
+  CADET_CONNECTION_BROKEN
+};
 
 
 /**
@@ -37,6 +76,12 @@
  */
 struct CadetConnection
 {
+
+  /**
+   * ID of the connection.
+   */
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
+
   /**
    * To which peer does this connection go?
    */
@@ -48,6 +93,26 @@ struct CadetConnection
   struct CadetPeerPath *path;
 
   /**
+   * Pending message, NULL if we are ready to transmit.
+   */
+  struct GNUNET_MQ_Envelope *env;
+
+  /**
+   * Message queue to the first hop, or NULL if we have no connection yet.
+   */
+  struct GNUNET_MQ_Handle *mq;
+
+  /**
+   * Handle for calling #GCP_request_mq_cancel() once we are finished.
+   */
+  struct GCP_MessageQueueManager *mq_man;
+
+  /**
+   * Task for connection maintenance.
+   */
+  struct GNUNET_SCHEDULER_Task *task;
+
+  /**
    * Function to call once we are ready to transmit.
    */
   GNUNET_SCHEDULER_TaskCallback ready_cb;
@@ -58,6 +123,16 @@ struct CadetConnection
   void *ready_cb_cls;
 
   /**
+   * How long do we wait before we try again with a CREATE message?
+   */
+  struct GNUNET_TIME_Relative retry_delay;
+
+  /**
+   * State of the connection.
+   */
+  enum CadetConnectionState state;
+
+  /**
    * Offset of our @e destination in @e path.
    */
   unsigned int off;
@@ -74,8 +149,9 @@ struct CadetConnection
 int
 GCC_is_ready (struct CadetConnection *cc)
 {
-  GNUNET_break (0);
-  return GNUNET_NO;
+  return ( (NULL != cc->mq) &&
+           (CADET_CONNECTION_READY == cc->state) &&
+           (NULL == cc->env) ) ? GNUNET_YES : GNUNET_NO;
 }
 
 
@@ -87,15 +163,193 @@ GCC_is_ready (struct CadetConnection *cc)
 void
 GCC_destroy (struct CadetConnection *cc)
 {
+  if (NULL != cc->env)
+  {
+    if (NULL != cc->mq)
+      GNUNET_MQ_send_cancel (cc->env);
+    else
+      GNUNET_MQ_discard (cc->env);
+    cc->env = NULL;
+  }
+  if ( (NULL != cc->mq) &&
+       (CADET_CONNECTION_SENDING_CREATE != cc->state) )
+  {
+    /* Need to notify next hop that we are down. */
+    struct GNUNET_MQ_Envelope *env;
+    struct GNUNET_CADET_ConnectionDestroy *destroy_msg;
+
+    env = GNUNET_MQ_msg (destroy_msg,
+                         GNUNET_MESSAGE_TYPE_CADET_CONNECTION_DESTROY);
+    destroy_msg->cid = cc->cid;
+    GNUNET_MQ_send (cc->mq,
+                    env);
+  }
+  cc->mq = NULL;
+  GCP_request_mq_cancel (cc->mq_man);
+  cc->mq_man = NULL;
   GCPP_del_connection (cc->path,
                        cc->off,
                        cc);
-  GNUNET_assert (0); // FIXME: incomplete implementation!
   GNUNET_free (cc);
 }
 
 
 /**
+ * Expand the shorter CADET hash to a full GNUnet hash.
+ *
+ * @param id hash to expand
+ * @return expanded hash
+ */
+static const struct GNUNET_HashCode *
+h2hc (const struct GNUNET_CADET_Hash *id)
+{
+  static struct GNUNET_HashCode hc;
+  char *ptr = (char *) &hc;
+
+  GNUNET_assert (sizeof (hc) == 2 * sizeof (*id));
+  GNUNET_memcpy (ptr,
+                 id,
+                 sizeof (*id));
+  GNUNET_memcpy (&ptr[sizeof (*id)],
+                 id,
+                 sizeof (*id));
+  return &hc;
+}
+
+
+/**
+ * Get the connection ID as a full hash.
+ *
+ * @param cc Connection to get the ID from.
+ * @return full hash ID of the connection.
+ */
+const struct GNUNET_HashCode *
+GCC_get_h (const struct CadetConnection *cc)
+{
+  return h2hc (&cc->cid.connection_of_tunnel);
+}
+
+
+/**
+ * An ACK was received for this connection, process it.
+ *
+ * @param cc the connection that got the ACK.
+ */
+void
+GCC_handle_ack (struct CadetConnection *cc)
+{
+  GNUNET_SCHEDULER_cancel (cc->task);
+#if FIXME
+  cc->task = GNUNET_SCHEDULER_add_delayed (cc->keepalive_period,
+                                           &send_keepalive,
+                                           cc);
+#endif
+  cc->state = CADET_CONNECTION_READY;
+  cc->ready_cb (cc->ready_cb_cls);
+}
+
+
+/**
+ * Send a CREATE message to the first hop.
+ *
+ * @param cls the `struct CadetConnection` to initiate
+ */
+static void
+send_create (void *cls);
+
+
+/**
+ * We finished transmission of the create message, now wait for
+ * ACK or retransmit.
+ *
+ * @param cls the `struct CadetConnection` that sent the create message
+ */
+static void
+transmit_create_done_cb (void *cls)
+{
+  struct CadetConnection *cc = cls;
+
+  cc->state = CADET_CONNECTION_SENT;
+  cc->env = NULL;
+  /* FIXME: at some point, we need to reset the delay back to 0! */
+  cc->retry_delay = GNUNET_TIME_STD_BACKOFF (cc->retry_delay);
+  cc->task = GNUNET_SCHEDULER_add_delayed (cc->retry_delay,
+                                           &send_create,
+                                           cc);
+}
+
+
+/**
+ * Send a CREATE message to the first hop.
+ *
+ * @param cls the `struct CadetConnection` to initiate
+ */
+static void
+send_create (void *cls)
+{
+  struct CadetConnection *cc = cls;
+  struct GNUNET_CADET_ConnectionCreate *create_msg;
+  struct GNUNET_PeerIdentity *pids;
+  struct GNUNET_MQ_Envelope *env;
+  unsigned int path_length;
+
+  cc->task = NULL;
+  GNUNET_assert (NULL != cc->mq);
+  path_length = GCPP_get_length (cc->path);
+  env = GNUNET_MQ_msg_extra (create_msg,
+                             path_length * sizeof (struct GNUNET_PeerIdentity),
+                             GNUNET_MESSAGE_TYPE_CADET_CONNECTION_CREATE);
+  create_msg->cid = cc->cid;
+  pids = (struct GNUNET_PeerIdentity *) &create_msg[1];
+  for (unsigned int i=0;i<path_length;i++)
+    pids[i] = *GCP_get_id (GCPP_get_peer_at_offset (cc->path,
+                                                    i));
+  cc->env = env;
+  GNUNET_MQ_notify_sent (env,
+                         &transmit_create_done_cb,
+                         cc);
+  GNUNET_MQ_send (cc->mq,
+                  env);
+}
+
+
+/**
+ * There has been a change in the message queue existence for our
+ * peer at the first hop.  Adjust accordingly.
+ *
+ * @param cls the `struct CadetConnection`
+ * @param mq NULL if the CORE connection was lost, non-NULL if
+ *           it became available
+ */
+static void
+manage_first_hop_mq (void *cls,
+                     struct GNUNET_MQ_Handle *mq)
+{
+  struct CadetConnection *cc = cls;
+
+  if (NULL == mq)
+  {
+    /* Connection is down, for now... */
+    cc->mq = NULL;
+    if (NULL != cc->task)
+    {
+      GNUNET_SCHEDULER_cancel (cc->task);
+      cc->task = NULL;
+    }
+    return;
+  }
+
+  cc->mq = mq;
+  cc->state = CADET_CONNECTION_SENDING_CREATE;
+
+  /* Now repeat sending connection creation messages
+     down the path, until we get an ACK! */
+  cc->task = GNUNET_SCHEDULER_add_now (&send_create,
+                                       cc);
+}
+
+
+/**
  * Create a connection to @a destination via @a path and
  * notify @a cb whenever we are ready for more data.
  *
@@ -103,6 +357,7 @@ GCC_destroy (struct CadetConnection *cc)
  * @param path which path to take (may not be the full path)
  * @param ready_cb function to call when ready to transmit
  * @param ready_cb_cls closure for @a cb
+ * @return handle to the connection
  */
 struct CadetConnection *
 GCC_create (struct CadetPeer *destination,
@@ -111,44 +366,82 @@ GCC_create (struct CadetPeer *destination,
             void *ready_cb_cls)
 {
   struct CadetConnection *cc;
+  struct CadetPeer *first_hop;
   unsigned int off;
 
   off = GCPP_find_peer (path,
                         destination);
   GNUNET_assert (UINT_MAX > off);
-
-  GNUNET_assert (0); // fIXME: unfinished
-
   cc = GNUNET_new (struct CadetConnection);
+  GNUNET_CRYPTO_random_block (GNUNET_CRYPTO_QUALITY_NONCE,
+                              &cc->cid,
+                              sizeof (cc->cid));
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_put (connections,
+                                                    GCC_get_h (cc),
+                                                    cc,
+                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+  cc->ready_cb = ready_cb;
+  cc->ready_cb_cls = ready_cb_cls;
   cc->path = path;
   cc->off = off;
   GCPP_add_connection (path,
                        off,
                        cc);
   for (unsigned int i=0;i<off;i++)
-  {
-    // FIXME: remember existence of this connection with
-    // ALL peers on the path!
-    // (and remove on destruction of connection!)
-  }
+    GCP_add_connection (GCPP_get_peer_at_offset (path,
+                                                 off),
+                        cc);
+
+  first_hop = GCPP_get_peer_at_offset (path,
+                                       0);
+  cc->mq_man = GCP_request_mq (first_hop,
+                               &manage_first_hop_mq,
+                               cc);
   return cc;
 }
 
 
 /**
+ * We finished transmission of a message, if we are still ready, tell
+ * the tunnel!
+ *
+ * @param cls our `struct CadetConnection`
+ */
+static void
+transmit_done_cb (void *cls)
+{
+  struct CadetConnection *cc = cls;
+
+  cc->env = NULL;
+  if ( (NULL != cc->mq) &&
+       (CADET_CONNECTION_READY == cc->state) )
+    cc->ready_cb (cc->ready_cb_cls);
+}
+
+
+/**
  * Transmit message @a msg via connection @a cc.  Must only be called
  * (once) after the connection has signalled that it is ready via the
  * `ready_cb`.  Clients can also use #GCC_is_ready() to check if the
  * connection is right now ready for transmission.
  *
  * @param cc connection identification
- * @param msg message to transmit
+ * @param env envelope with message to transmit
  */
 void
 GCC_transmit (struct CadetConnection *cc,
-              const struct GNUNET_MessageHeader *msg)
+              struct GNUNET_MQ_Envelope *env)
 {
-  GNUNET_assert (0); // FIXME
+  GNUNET_assert (NULL == cc->env);
+  cc->env = env;
+  GNUNET_MQ_notify_sent (env,
+                         &transmit_done_cb,
+                         cc);
+  if ( (NULL != cc->mq) &&
+       (CADET_CONNECTION_READY == cc->state) )
+    GNUNET_MQ_send (cc->mq,
+                    env);
 }
 
 
@@ -174,8 +467,7 @@ GCC_get_path (struct CadetConnection *cc)
 const struct GNUNET_CADET_ConnectionTunnelIdentifier *
 GCC_get_id (struct CadetConnection *cc)
 {
-  GNUNET_assert (0); // FIXME
-  return NULL;
+  return &cc->cid;
 }
 
 
@@ -189,4 +481,7 @@ void
 GCC_debug (struct CadetConnection *cc,
            enum GNUNET_ErrorType level)
 {
+  GNUNET_break (0); // FIXME: implement...
 }
+
+/* end of gnunet-service-cadet-new_connection.c */
diff --git a/src/cadet/gnunet-service-cadet-new_connection.h b/src/cadet/gnunet-service-cadet-new_connection.h
index c11e34024..c9738d86e 100644
--- a/src/cadet/gnunet-service-cadet-new_connection.h
+++ b/src/cadet/gnunet-service-cadet-new_connection.h
@@ -60,6 +60,7 @@ GCC_destroy (struct CadetConnection *cc);
  * @param path which path to take (may not be the full path)
  * @param ready_cb function to call when ready to transmit
  * @param ready_cb_cls closure for @a cb
+ * @return handle to the connection
  */
 struct CadetConnection *
 GCC_create (struct CadetPeer *destination,
@@ -75,11 +76,14 @@ GCC_create (struct CadetPeer *destination,
  * connection is right now ready for transmission.
  *
  * @param cc connection identification
- * @param msg message to transmit
+ * @param env envelope with message to transmit;
+ *            the #GNUNET_MQ_notify_send() must not have yet been used
+ *            for the envelope.  Also, the message better match the
+ *            connection identifier of this connection...
  */
 void
 GCC_transmit (struct CadetConnection *cc,
-              const struct GNUNET_MessageHeader *msg);
+              struct GNUNET_MQ_Envelope *env);
 
 
 /**
@@ -103,6 +107,17 @@ GCC_get_id (struct CadetConnection *cc);
 
 
 /**
+ * Get the connection ID as a full hash.
+ *
+ * @param cc Connection to get the ID from.
+ * @return full hash ID of the connection.
+ * @deprecated try to replace use of full hash codes eventually...
+ */
+const struct GNUNET_HashCode *
+GCC_get_h (const struct CadetConnection *cc);
+
+
+/**
  * Log connection info.
  *
  * @param cc connection
diff --git a/src/cadet/gnunet-service-cadet-new_peer.c b/src/cadet/gnunet-service-cadet-new_peer.c
index 9878c540e..ed3b24641 100644
--- a/src/cadet/gnunet-service-cadet-new_peer.c
+++ b/src/cadet/gnunet-service-cadet-new_peer.c
@@ -35,6 +35,7 @@
 #include "cadet_protocol.h"
 #include "cadet_path.h"
 #include "gnunet-service-cadet-new.h"
+#include "gnunet-service-cadet-new_connection.h"
 #include "gnunet-service-cadet-new_dht.h"
 #include "gnunet-service-cadet-new_peer.h"
 #include "gnunet-service-cadet-new_paths.h"
@@ -51,6 +52,43 @@
 #define IDLE_PATH_TIMEOUT GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_MINUTES, 2)
 
 
+
+
+/**
+ * Data structure used to track whom we have to notify about changes
+ * to our message queue.
+ */
+struct GCP_MessageQueueManager
+{
+
+  /**
+   * Kept in a DLL.
+   */
+  struct GCP_MessageQueueManager *next;
+
+  /**
+   * Kept in a DLL.
+   */
+  struct GCP_MessageQueueManager *prev;
+
+  /**
+   * Function to call with updated message queue object.
+   */
+  GCP_MessageQueueNotificationCallback cb;
+
+  /**
+   * Closure for @e cb.
+   */
+  void *cb_cls;
+
+  /**
+   * The peer this is for.
+   */
+  struct CadetPeer *cp;
+
+};
+
+
 /**
  * Struct containing all information regarding a given peer
  */
@@ -79,6 +117,16 @@ struct CadetPeer
   struct CadetPeerPathEntry **path_tails;
 
   /**
+   * Notifications to call when @e core_mq changes.
+   */
+  struct GCP_MessageQueueManager *mqm_head;
+
+  /**
+   * Notifications to call when @e core_mq changes.
+   */
+  struct GCP_MessageQueueManager *mqm_tail;
+
+  /**
    * MIN-heap of paths owned by this peer (they also end at this
    * peer).  Ordered by desirability.
    */
@@ -207,11 +255,35 @@ destroy_peer (void *cls)
   GNUNET_CONTAINER_multihashmap_destroy (cp->connections);
   GNUNET_CONTAINER_heap_destroy (cp->path_heap);
   GNUNET_free_non_null (cp->hello);
+  /* Peer should not be freed if paths exist; if there are no paths,
+     there ought to be no connections, and without connections, no
+     notifications. Thus we can assert that mqm_head is empty at this
+     point. */
+  GNUNET_assert (NULL == cp->mqm_head);
   GNUNET_free (cp);
 }
 
 
 /**
+ * Set the message queue to @a mq for peer @a cp and notify watchers.
+ *
+ * @param cp peer to modify
+ * @param mq message queue to set (can be NULL)
+ */
+void
+GCP_set_mq (struct CadetPeer *cp,
+            struct GNUNET_MQ_Handle *mq)
+{
+  cp->core_mq = mq;
+  for (struct GCP_MessageQueueManager *mqm = cp->mqm_head;
+       NULL != mqm;
+       mqm = mqm->next)
+    mqm->cb (mqm->cb_cls,
+             mq);
+}
+
+
+/**
  * Function called to destroy a peer now.
  *
  * @param cls NULL
@@ -465,6 +537,41 @@ GCP_detach_path (struct CadetPeer *cp,
 
 
 /**
+ * Add a @a connection to this @a cp.
+ *
+ * @param cp peer via which the @a connection goes
+ * @param cc the connection to add
+ */
+void
+GCP_add_connection (struct CadetPeer *cp,
+                    struct CadetConnection *cc)
+{
+  GNUNET_assert (GNUNET_OK ==
+                 GNUNET_CONTAINER_multihashmap_put (cp->connections,
+                                                    GCC_get_h (cc),
+                                                    cc,
+                                                    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+}
+
+
+/**
+ * Remove a @a connection that went via this @a cp.
+ *
+ * @param cp peer via which the @a connection went
+ * @param cc the connection to remove
+ */
+void
+GCP_remove_connection (struct CadetPeer *cp,
+                       struct CadetConnection *cc)
+{
+  GNUNET_assert (GNUNET_YES ==
+                 GNUNET_CONTAINER_multihashmap_remove (cp->connections,
+                                                       GCC_get_h (cc),
+                                                       cc));
+}
+
+
+/**
  * This peer is now on more "active" duty, activate processes related to it.
  *
  * @param cp the more-active peer
@@ -728,4 +835,51 @@ GCP_drop_tunnel (struct CadetPeer *peer,
 }
 
 
+/**
+ * Start message queue change notifications.
+ *
+ * @param cp peer to notify for
+ * @param cb function to call if mq becomes available or unavailable
+ * @param cb_cls closure for @a cb
+ * @return handle to cancel request
+ */
+struct GCP_MessageQueueManager *
+GCP_request_mq (struct CadetPeer *cp,
+                GCP_MessageQueueNotificationCallback cb,
+                void *cb_cls)
+{
+  struct GCP_MessageQueueManager *mqm;
+
+  mqm = GNUNET_new (struct GCP_MessageQueueManager);
+  mqm->cb = cb;
+  mqm->cb_cls = cb_cls;
+  mqm->cp = cp;
+  GNUNET_CONTAINER_DLL_insert (cp->mqm_head,
+                               cp->mqm_tail,
+                               mqm);
+  if (NULL != cp->core_mq)
+    cb (cb_cls,
+        cp->core_mq);
+  return mqm;
+}
+
+
+/**
+ * Stops message queue change notifications.
+ *
+ * @param mqm handle matching request to cancel
+ */
+void
+GCP_request_mq_cancel (struct GCP_MessageQueueManager *mqm)
+{
+  struct CadetPeer *cp = mqm->cp;
+
+  GNUNET_CONTAINER_DLL_remove (cp->mqm_head,
+                               cp->mqm_tail,
+                               mqm);
+  GNUNET_free (mqm);
+}
+
+
+
 /* end of gnunet-service-cadet-new_peer.c */
diff --git a/src/cadet/gnunet-service-cadet-new_peer.h b/src/cadet/gnunet-service-cadet-new_peer.h
index 780640674..74988981f 100644
--- a/src/cadet/gnunet-service-cadet-new_peer.h
+++ b/src/cadet/gnunet-service-cadet-new_peer.h
@@ -218,6 +218,28 @@ GCP_detach_path (struct CadetPeer *cp,
 
 
 /**
+ * Add a @a connection to this @a cp.
+ *
+ * @param cp peer via which the @a connection goes
+ * @param cc the connection to add
+ */
+void
+GCP_add_connection (struct CadetPeer *cp,
+                    struct CadetConnection *cc);
+
+
+/**
+ * Remove a @a connection that went via this @a cp.
+ *
+ * @param cp peer via which the @a connection went
+ * @param cc the connection to remove
+ */
+void
+GCP_remove_connection (struct CadetPeer *cp,
+                       struct CadetConnection *cc);
+
+
+/**
  * We got a HELLO for a @a cp, remember it, and possibly
  * trigger adequate actions (like trying to connect).
  *
@@ -238,4 +260,56 @@ void
 GCP_destroy_all_peers (void);
 
 
+/**
+ * Data structure used to track whom we have to notify about changes
+ * to our message queue.
+ */
+struct GCP_MessageQueueManager;
+
+
+/**
+ * Function to call with updated message queue object.
+ *
+ * @param cls closure
+ * @param mq NULL if MQ is gone, otherwise an active message queue
+ */
+typedef void
+(*GCP_MessageQueueNotificationCallback)(void *cls,
+                                        struct GNUNET_MQ_Handle *mq);
+
+
+/**
+ * Start message queue change notifications.
+ *
+ * @param cp peer to notify for
+ * @param cb function to call if mq becomes available or unavailable
+ * @param cb_cls closure for @a cb
+ * @return handle to cancel request
+ */
+struct GCP_MessageQueueManager *
+GCP_request_mq (struct CadetPeer *cp,
+                GCP_MessageQueueNotificationCallback cb,
+                void *cb_cls);
+
+
+/**
+ * Stops message queue change notifications.
+ *
+ * @param mqm handle matching request to cancel
+ */
+void
+GCP_request_mq_cancel (struct GCP_MessageQueueManager *mqm);
+
+
+/**
+ * Set the message queue to @a mq for peer @a cp and notify watchers.
+ *
+ * @param cp peer to modify
+ * @param mq message queue to set (can be NULL)
+ */
+void
+GCP_set_mq (struct CadetPeer *cp,
+            struct GNUNET_MQ_Handle *mq);
+
+
 #endif
diff --git a/src/cadet/gnunet-service-cadet-new_tunnels.c b/src/cadet/gnunet-service-cadet-new_tunnels.c
index 1a07140f2..18c469b67 100644
--- a/src/cadet/gnunet-service-cadet-new_tunnels.c
+++ b/src/cadet/gnunet-service-cadet-new_tunnels.c
@@ -268,9 +268,9 @@ struct CadetTunnelQueueEntry
   void *cont_cls;
 
   /**
-   * (Encrypted) message to send follows.
+   * Envelope of message to send follows.
    */
-  /* struct GNUNET_MessageHeader *msg; */
+  struct GNUNET_MQ_Envelope *env;
 };
 
 
@@ -533,7 +533,7 @@ destroy_tunnel (void *cls)
     GNUNET_CONTAINER_DLL_remove (t->tq_head,
                                  t->tq_tail,
                                  tqe);
-    // FIXME: implement!
+    GNUNET_MQ_discard (tqe->env);
     GNUNET_free (tqe);
   }
   GCP_drop_tunnel (t->destination,
@@ -570,7 +570,7 @@ connection_ready_cb (void *cls)
                                t->tq_tail,
                                tq);
   GCC_transmit (ct->cc,
-                (const struct GNUNET_MessageHeader *) &tq[1]);
+                tq->env);
   tq->cont (tq->cont_cls);
   GNUNET_free (tq);
 }
@@ -693,12 +693,13 @@ consider_path_cb (void *cls,
   ct->created = GNUNET_TIME_absolute_get ();
   ct->t = t;
   ct->cc = GCC_create (t->destination,
-                      path,
-                      &connection_ready_cb,
-                      t);
+                       path,
+                       &connection_ready_cb,
+                       t);
   /* FIXME: schedule job to kill connection (and path?)  if it takes
      too long to get ready! (And track performance data on how long
-     other connections took with the tunnel!) */
+     other connections took with the tunnel!)
+     => Note: to be done within 'connection'-logic! */
   GNUNET_CONTAINER_DLL_insert (t->connection_head,
                                t->connection_tail,
                                ct);
diff --git a/src/cadet/gnunet-service-cadet_connection.c b/src/cadet/gnunet-service-cadet_connection.c
index e727ef199..ffd98bf34 100644
--- a/src/cadet/gnunet-service-cadet_connection.c
+++ b/src/cadet/gnunet-service-cadet_connection.c
@@ -218,7 +218,7 @@ struct CadetConnection
   /**
    * ID of the connection.
    */
-  struct GNUNET_CADET_Hash id;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier id;
 
   /**
    * Path being used for the tunnel. At the origin of the connection
@@ -1100,7 +1100,7 @@ send_broken (struct CadetConnection *c,
  * @param neighbor Peer to notify (neighbor who sent the connection).
  */
 static void
-send_broken_unknown (const struct GNUNET_CADET_Hash *connection_id,
+send_broken_unknown (const struct GNUNET_CADET_ConnectionTunnelIdentifier *connection_id,
                      const struct GNUNET_PeerIdentity *id1,
                      const struct GNUNET_PeerIdentity *id2,
                      struct CadetPeer *neighbor)
@@ -1836,7 +1836,7 @@ add_to_peer (struct CadetConnection *c,
 static void
 log_message (const struct GNUNET_MessageHeader *message,
              const struct CadetPeer *peer,
-             const struct GNUNET_CADET_Hash *conn_id)
+             const struct GNUNET_CADET_ConnectionTunnelIdentifier *conn_id)
 {
   uint16_t size;
   uint16_t type;
@@ -1874,7 +1874,7 @@ void
 GCC_handle_create (struct CadetPeer *peer,
                    const struct GNUNET_CADET_ConnectionCreate *msg)
 {
-  const struct GNUNET_CADET_Hash *cid;
+  const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid;
   struct GNUNET_PeerIdentity *id;
   struct CadetPeerPath *path;
   struct CadetPeer *dest_peer;
@@ -2409,7 +2409,7 @@ GCC_handle_poll (struct CadetPeer *peer,
  */
 static int
 check_message (const struct GNUNET_MessageHeader *message,
-               const struct GNUNET_CADET_Hash* cid,
+               const struct GNUNET_CADET_ConnectionTunnelIdentifier* cid,
                struct CadetConnection *c,
                struct CadetPeer *sender,
                uint32_t pid)
@@ -2534,7 +2534,7 @@ void
 GCC_handle_kx (struct CadetPeer *peer,
                const struct GNUNET_CADET_KX *msg)
 {
-  const struct GNUNET_CADET_Hash* cid;
+  const struct GNUNET_CADET_ConnectionTunnelIdentifier* cid;
   struct CadetConnection *c;
   int fwd;
 
@@ -2591,7 +2591,7 @@ void
 GCC_handle_encrypted (struct CadetPeer *peer,
                       const struct GNUNET_CADET_Encrypted *msg)
 {
-  const struct GNUNET_CADET_Hash* cid;
+  const struct GNUNET_CADET_ConnectionTunnelIdentifier* cid;
   struct CadetConnection *c;
   uint32_t pid;
   int fwd;
@@ -2734,7 +2734,7 @@ GCC_shutdown (void)
  *         NULL in case of error: own id not in path, wrong neighbors, ...
 */
 struct CadetConnection *
-GCC_new (const struct GNUNET_CADET_Hash *cid,
+GCC_new (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
          struct CadetTunnel *t,
          struct CadetPeerPath *path,
          unsigned int own_pos)
@@ -2871,7 +2871,7 @@ GCC_destroy (struct CadetConnection *c)
  *
  * @return ID of the connection.
  */
-const struct GNUNET_CADET_Hash *
+const struct GNUNET_CADET_ConnectionTunnelIdentifier *
 GCC_get_id (const struct CadetConnection *c)
 {
   return &c->id;
diff --git a/src/cadet/gnunet-service-cadet_connection.h b/src/cadet/gnunet-service-cadet_connection.h
index 18f33ce7c..f1eb4d59c 100644
--- a/src/cadet/gnunet-service-cadet_connection.h
+++ b/src/cadet/gnunet-service-cadet_connection.h
@@ -277,7 +277,7 @@ GCC_shutdown (void);
  *         NULL in case of error: own id not in path, wrong neighbors, ...
  */
 struct CadetConnection *
-GCC_new (const struct GNUNET_CADET_Hash *cid,
+GCC_new (const struct GNUNET_CADET_ConnectionTunnelIdentifier *cid,
          struct CadetTunnel *t,
          struct CadetPeerPath *path,
          unsigned int own_pos);
@@ -300,7 +300,7 @@ GCC_destroy (struct CadetConnection *c);
  *
  * @return ID of the connection.
  */
-const struct GNUNET_CADET_Hash *
+const struct GNUNET_CADET_ConnectionTunnelIdentifier *
 GCC_get_id (const struct CadetConnection *c);
 
 
diff --git a/src/cadet/gnunet-service-cadet_local.c b/src/cadet/gnunet-service-cadet_local.c
index 743fbf054..6e6313902 100644
--- a/src/cadet/gnunet-service-cadet_local.c
+++ b/src/cadet/gnunet-service-cadet_local.c
@@ -965,8 +965,9 @@ static void
 iter_connection (void *cls, struct CadetConnection *c)
 {
   struct GNUNET_CADET_LocalInfoTunnel *msg = cls;
-  struct GNUNET_CADET_Hash *h = (struct GNUNET_CADET_Hash *) &msg[1];
+  struct GNUNET_CADET_ConnectionTunnelIdentifier *h;
 
+  h = (struct GNUNET_CADET_ConnectionTunnelIdentifier *) &msg[1];
   h[msg->connections] = *(GCC_get_id (c));
   msg->connections++;
 }
diff --git a/src/cadet/gnunet-service-cadet_tunnel.c b/src/cadet/gnunet-service-cadet_tunnel.c
index 2e17774d4..29087dba4 100644
--- a/src/cadet/gnunet-service-cadet_tunnel.c
+++ b/src/cadet/gnunet-service-cadet_tunnel.c
@@ -1406,7 +1406,7 @@ send_prebuilt_message (const struct GNUNET_MessageHeader *message,
        "Sending message of type %s with PID %u and CID %s\n",
        GC_m2s (type),
        htonl (ax_msg->pid),
-       GC_h2s (&ax_msg->cid));
+       GC_h2s (&ax_msg->cid.connection_of_tunnel));
 
   if (NULL == cont)
   {
@@ -2744,7 +2744,7 @@ struct CadetConnection *
 GCT_use_path (struct CadetTunnel *t, struct CadetPeerPath *path)
 {
   struct CadetConnection *c;
-  struct GNUNET_CADET_Hash cid;
+  struct GNUNET_CADET_ConnectionTunnelIdentifier cid;
   unsigned int own_pos;
 
   if (NULL == t || NULL == path)