-improve UDP logging
[oweals/gnunet.git] / src / dv / plugin_transport_dv.c
index f6ea3e736ec68f67b8d95777a0c33d7f41d1e2dc..382e575966061c89a424908d0c7854687cda3b55 100644 (file)
@@ -1,6 +1,6 @@
 /*
      This file is part of GNUnet
-     (C) 2002--2013 Christian Grothoff (and other contributing authors)
+     Copyright (C) 2002--2014 Christian Grothoff (and other contributing authors)
 
      GNUnet is free software; you can redistribute it and/or modify
      it under the terms of the GNU General Public License as published
@@ -25,7 +25,6 @@
  * @author Nathan Evans
  * @author Christian Grothoff
  */
-
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_protocols.h"
@@ -38,7 +37,6 @@
 
 #define LOG(kind,...) GNUNET_log_from (kind, "transport-dv",__VA_ARGS__)
 
-#define PLUGIN_NAME "dv"
 
 /**
  * Encapsulation of all of the state of the plugin.
@@ -70,7 +68,7 @@ struct PendingRequest
   GNUNET_TRANSPORT_TransmitContinuation transmit_cont;
 
   /**
-   * Closure for transmit_cont.
+   * Closure for @e transmit_cont.
    */
   void *transmit_cont_cls;
 
@@ -84,6 +82,10 @@ struct PendingRequest
    */
   struct Session *session;
 
+  /**
+   * Number of bytes to transmit.
+   */
+  size_t size;
 };
 
 
@@ -92,12 +94,6 @@ struct PendingRequest
  */
 struct Session
 {
-
-  /**
-   * Mandatory session header.
-   */
-  struct SessionHeader header;
-
   /**
    * Pointer to the global plugin struct.
    */
@@ -113,11 +109,28 @@ struct Session
    */
   struct PendingRequest *pr_tail;
 
+  /**
+   * Address we use for the other peer.
+   */
+  struct GNUNET_HELLO_Address *address;
+
   /**
    * To whom are we talking to.
    */
   struct GNUNET_PeerIdentity sender;
 
+  /**
+   * Number of bytes waiting for transmission to this peer.
+   * FIXME: not set yet.
+   */
+  unsigned long long bytes_in_queue;
+
+  /**
+   * Number of messages waiting for transmission to this peer.
+   * FIXME: not set yet.
+   */
+  unsigned int msgs_in_queue;
+
   /**
    * Current distance to the given peer.
    */
@@ -126,11 +139,11 @@ struct Session
   /**
    * Current network the next hop peer is located in
    */
-  uint32_t network;
+  enum GNUNET_ATS_Network_Type network;
 
   /**
    * Does the transport service know about this session (and we thus
-   * need to call 'session_end' when it is released?)
+   * need to call `session_end` when it is released?)
    */
   int active;
 
@@ -168,9 +181,50 @@ struct Plugin
    */
   struct GNUNET_SERVER_MessageStreamTokenizer *mst;
 
+  /**
+   * Function to call about session status changes.
+   */
+  GNUNET_TRANSPORT_SessionInfoCallback sic;
+
+  /**
+   * Closure for @e sic.
+   */
+  void *sic_cls;
 };
 
 
+/**
+ * If a session monitor is attached, notify it about the new
+ * session state.
+ *
+ * @param plugin our plugin
+ * @param session session that changed state
+ * @param state new state of the session
+ */
+static void
+notify_session_monitor (struct Plugin *plugin,
+                        struct Session *session,
+                        enum GNUNET_TRANSPORT_SessionState state)
+{
+  struct GNUNET_TRANSPORT_SessionInfo info;
+
+  if (NULL == plugin->sic)
+    return;
+  memset (&info, 0, sizeof (info));
+  info.state = state;
+  info.is_inbound = GNUNET_SYSERR; /* hard to say */
+  info.num_msg_pending = session->msgs_in_queue;
+  info.num_bytes_pending = session->bytes_in_queue;
+  /* info.receive_delay remains zero as this is not supported by DV
+     (cannot selectively not receive from 'core') */
+  info.session_timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
+  info.address = session->address;
+  plugin->sic (plugin->sic_cls,
+               session,
+               &info);
+}
+
+
 /**
  * Notify transport service about the change in distance.
  *
@@ -182,23 +236,22 @@ notify_distance_change (struct Session *session)
   struct Plugin *plugin = session->plugin;
   struct GNUNET_ATS_Information ats;
 
+  if (GNUNET_YES != session->active)
+    return;
   ats.type = htonl ((uint32_t) GNUNET_ATS_QUALITY_NET_DISTANCE);
   ats.value = htonl (session->distance);
   plugin->env->update_address_metrics (plugin->env->cls,
-                                      &session->sender,
-                                      NULL, 0,
-                                      session,
-                                      &ats, 1);
+      session->address, session, &ats, 1);
 }
 
 
 /**
  * Function called by MST on each message from the box.
  *
- * @param cls closure with the 'struct Plugin'
+ * @param cls closure with the `struct Plugin *`
  * @param client identification of the client (with the 'struct Session')
  * @param message the actual message
- * @return GNUNET_OK on success
+ * @return #GNUNET_OK on success
  */
 static int
 unbox_cb (void *cls,
@@ -212,12 +265,16 @@ unbox_cb (void *cls,
   ats.type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
   ats.value = htonl (session->distance);
   session->active = GNUNET_YES;
-  plugin->env->receive (plugin->env->cls,
-                       &session->sender,
-                        message,
-                       session, "", 0);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Delivering message of type %u with %u bytes from peer `%s'\n",
+       ntohs (message->type),
+       ntohs (message->size),
+       GNUNET_i2s (&session->sender));
+
+  plugin->env->receive (plugin->env->cls, session->address, session,
+                        message);
   plugin->env->update_address_metrics (plugin->env->cls,
-               &session->sender, NULL, 0, session, &ats, 1);
+      session->address, session, &ats, 1);
   return GNUNET_OK;
 }
 
@@ -240,11 +297,9 @@ handle_dv_message_received (void *cls,
   struct GNUNET_ATS_Information ats;
   struct Session *session;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received `%s' message for peer `%s': new distance %u\n",
-      "DV_MESSAGE_RECEIVED",
-      GNUNET_i2s (sender), distance);
-
-
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received DV_MESSAGE_RECEIVED message for peer `%s': new distance %u\n",
+       GNUNET_i2s (sender), distance);
   session = GNUNET_CONTAINER_multipeermap_get (plugin->sessions,
                                               sender);
   if (NULL == session)
@@ -255,6 +310,8 @@ handle_dv_message_received (void *cls,
   if (GNUNET_MESSAGE_TYPE_DV_BOX == ntohs (msg->type))
   {
     /* need to unbox using MST */
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Unboxing DV message using MST\n");
     GNUNET_SERVER_mst_receive (plugin->mst,
                               session,
                               (const char *) &msg[1],
@@ -266,18 +323,21 @@ handle_dv_message_received (void *cls,
   ats.type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
   ats.value = htonl (distance);
   session->active = GNUNET_YES;
-  plugin->env->receive (plugin->env->cls, sender,
-                        msg,
-                        session, "", 0);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Delivering message of type %u with %u bytes from peer `%s'\n",
+       ntohs (msg->type),
+       ntohs (msg->size),
+       GNUNET_i2s (sender));
+  plugin->env->receive (plugin->env->cls, session->address, session, msg);
   plugin->env->update_address_metrics (plugin->env->cls,
-                                      sender, "", 0, session, &ats, 1);
+      session->address, session, &ats, 1);
 }
 
 
 /**
  * Function called if DV starts to be able to talk to a peer.
  *
- * @param cls closure with 'struct Plugin'
+ * @param cls closure with `struct Plugin *`
  * @param peer newly connected peer
  * @param distance distance to the peer
  * @param network the network the next hop is located in
@@ -285,21 +345,23 @@ handle_dv_message_received (void *cls,
 static void
 handle_dv_connect (void *cls,
                   const struct GNUNET_PeerIdentity *peer,
-                  uint32_t distance, uint32_t network)
+                  uint32_t distance,
+                   enum GNUNET_ATS_Network_Type network)
 {
   struct Plugin *plugin = cls;
   struct Session *session;
   struct GNUNET_ATS_Information ats[2];
 
+  GNUNET_break (GNUNET_ATS_NET_UNSPECIFIED != network);
   /**
    * This requires transport plugin to be linked to libgnunetats.
    * If you remove it, also remove libgnunetats linkage from Makefile.am
    */
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-      "Received `%s' message for peer `%s' with next hop in network %s \n",
-      "DV_CONNECT",
-      GNUNET_i2s (peer),
-      GNUNET_ATS_print_network_type (network));
+       "Received `%s' message for peer `%s' with next hop in network %s\n",
+       "DV_CONNECT",
+       GNUNET_i2s (peer),
+       GNUNET_ATS_print_network_type (network));
 
   session = GNUNET_CONTAINER_multipeermap_get (plugin->sessions,
                                               peer);
@@ -307,66 +369,75 @@ handle_dv_connect (void *cls,
   {
     GNUNET_break (0);
     session->distance = distance;
-    if (GNUNET_YES == session->active)
-      notify_distance_change (session);
+    notify_distance_change (session);
     return; /* nothing to do */
   }
 
   session = GNUNET_new (struct Session);
+  session->address = GNUNET_HELLO_address_allocate (peer, "dv",
+      NULL, 0, GNUNET_HELLO_ADDRESS_INFO_NONE);
   session->sender = *peer;
+  session->plugin = plugin;
   session->distance = distance;
   session->network = network;
-  GNUNET_assert(
-      GNUNET_YES == GNUNET_CONTAINER_multipeermap_put (plugin->sessions,
-          &session->sender, session,
-          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
+  GNUNET_assert(GNUNET_YES ==
+                GNUNET_CONTAINER_multipeermap_put (plugin->sessions,
+                                                   &session->sender, session,
+                                                   GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-      "Creating new session %p for peer `%s'\n",
-      session,
-      GNUNET_i2s (peer));
+       "Creating new DV session %p for peer `%s' at distance %u\n",
+       session,
+       GNUNET_i2s (peer),
+       distance);
 
   /* Notify transport and ats about new connection */
   ats[0].type = htonl (GNUNET_ATS_QUALITY_NET_DISTANCE);
   ats[0].value = htonl (distance);
   ats[1].type = htonl (GNUNET_ATS_NETWORK_TYPE);
-  ats[1].value = htonl (network);
-  plugin->env->session_start (plugin->env->cls, peer, PLUGIN_NAME, NULL, 0,
-      session, ats, 2);
+  ats[1].value = htonl ((uint32_t) network);
+  session->active = GNUNET_YES;
+  plugin->env->session_start (plugin->env->cls, session->address,
+                              session, ats, 2);
+  notify_session_monitor (session->plugin,
+                          session,
+                          GNUNET_TRANSPORT_SS_UP);
 }
 
 
 /**
  * Function called if DV distance to a peer is changed.
  *
- * @param cls closure with 'struct Plugin'
+ * @param cls closure with `struct Plugin *`
  * @param peer connected peer
  * @param distance new distance to the peer
+ * @param network network type used for the connection
  */
 static void
 handle_dv_distance_changed (void *cls,
                            const struct GNUNET_PeerIdentity *peer,
-                           uint32_t distance)
+                           uint32_t distance,
+                            enum GNUNET_ATS_Network_Type network)
 {
   struct Plugin *plugin = cls;
   struct Session *session;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received `%s' message for peer `%s': new distance %u\n",
-      "DV_DISTANCE_CHANGED",
-      GNUNET_i2s (peer), distance);
-
+  GNUNET_break (GNUNET_ATS_NET_UNSPECIFIED != network);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received `%s' message for peer `%s': new distance %u\n",
+       "DV_DISTANCE_CHANGED",
+       GNUNET_i2s (peer),
+       distance);
   session = GNUNET_CONTAINER_multipeermap_get (plugin->sessions,
                                               peer);
   if (NULL == session)
   {
     GNUNET_break (0);
-    /* FIXME */
-    handle_dv_connect (plugin, peer, distance, 0);
+    handle_dv_connect (plugin, peer, distance, network);
     return;
   }
   session->distance = distance;
-  if (GNUNET_YES == session->active)
-    notify_distance_change (session);
+  notify_distance_change (session);
 }
 
 
@@ -387,14 +458,19 @@ free_session (struct Session *session)
                                                       session));
 
   LOG (GNUNET_ERROR_TYPE_DEBUG,
-      "Freeing session %p for peer `%s'\n",
-      session,
-      GNUNET_i2s (&session->sender));
-
+       "Freeing session %p for peer `%s'\n",
+       session,
+       GNUNET_i2s (&session->sender));
   if (GNUNET_YES == session->active)
+  {
+    notify_session_monitor (session->plugin,
+                           session,
+                           GNUNET_TRANSPORT_SS_DONE);
     plugin->env->session_end (plugin->env->cls,
-                             &session->sender,
+                             session->address,
                              session);
+    session->active = GNUNET_NO;
+  }
   while (NULL != (pr = session->pr_head))
   {
     GNUNET_CONTAINER_DLL_remove (session->pr_head,
@@ -405,9 +481,11 @@ free_session (struct Session *session)
     if (NULL != pr->transmit_cont)
       pr->transmit_cont (pr->transmit_cont_cls,
                         &session->sender,
-                        GNUNET_SYSERR, 0, 0);
+                        GNUNET_SYSERR,
+                         pr->size, 0);
     GNUNET_free (pr);
   }
+  GNUNET_HELLO_address_free (session->address);
   GNUNET_free (session);
 }
 
@@ -415,25 +493,24 @@ free_session (struct Session *session)
 /**
  * Function called if DV is no longer able to talk to a peer.
  *
- * @param cls closure with 'struct Plugin'
+ * @param cls closure with `struct Plugin *`
  * @param peer peer that disconnected
  */
 static void
 handle_dv_disconnect (void *cls,
-                     const struct GNUNET_PeerIdentity *peer)
+                      const struct GNUNET_PeerIdentity *peer)
 {
   struct Plugin *plugin = cls;
   struct Session *session;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received `%s' message for peer `%s'\n",
-      "DV_DISCONNECT",
-      GNUNET_i2s (peer));
-
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received `%s' message for peer `%s'\n",
+       "DV_DISCONNECT",
+       GNUNET_i2s (peer));
   session = GNUNET_CONTAINER_multipeermap_get (plugin->sessions,
                                               peer);
   if (NULL == session)
     return; /* nothing to do */
-
   free_session (session);
 }
 
@@ -443,7 +520,7 @@ handle_dv_disconnect (void *cls,
  * Clean up the pending request, and call continuations.
  *
  * @param cls closure
- * @param ok GNUNET_OK on success, GNUNET_SYSERR on error
+ * @param ok #GNUNET_OK on success, #GNUNET_SYSERR on error
  */
 static void
 send_finished (void *cls,
@@ -459,7 +536,8 @@ send_finished (void *cls,
   if (NULL != pr->transmit_cont)
     pr->transmit_cont (pr->transmit_cont_cls,
                       &session->sender,
-                      ok, 0, 0);
+                      ok,
+                       pr->size, 0);
   GNUNET_free (pr);
 }
 
@@ -478,7 +556,7 @@ send_finished (void *cls,
  *        been transmitted (or if the transport is ready
  *        for the next transmission call; or if the
  *        peer disconnected...)
- * @param cont_cls closure for cont
+ * @param cont_cls closure for @a cont
  * @return number of bytes used (on the physical network, with overheads);
  *         -1 on hard errors (i.e. address invalid); 0 is a legal value
  *         and does NOT mean that the message was not transmitted (DV)
@@ -486,10 +564,13 @@ send_finished (void *cls,
 static ssize_t
 dv_plugin_send (void *cls,
                struct Session *session,
-                const char *msgbuf, size_t msgbuf_size, unsigned int priority,
+                const char *msgbuf,
+                size_t msgbuf_size,
+                unsigned int priority,
                 struct GNUNET_TIME_Relative timeout,
                 GNUNET_TRANSPORT_TransmitContinuation cont, void *cont_cls)
 {
+  struct Plugin *plugin = cls;
   struct PendingRequest *pr;
   const struct GNUNET_MessageHeader *msg;
   struct GNUNET_MessageHeader *box;
@@ -499,6 +580,8 @@ dv_plugin_send (void *cls,
   if (ntohs (msg->size) != msgbuf_size)
   {
     /* need to box */
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Boxing DV message\n");
     box = GNUNET_malloc (sizeof (struct GNUNET_MessageHeader) + msgbuf_size);
     box->type = htons (GNUNET_MESSAGE_TYPE_DV_BOX);
     box->size = htons (sizeof (struct GNUNET_MessageHeader) + msgbuf_size);
@@ -509,10 +592,12 @@ dv_plugin_send (void *cls,
   pr->transmit_cont = cont;
   pr->transmit_cont_cls = cont_cls;
   pr->session = session;
+  pr->size = msgbuf_size;
   GNUNET_CONTAINER_DLL_insert_tail (session->pr_head,
                                    session->pr_tail,
                                    pr);
-  pr->th = GNUNET_DV_send (session->plugin->dvh,
+
+  pr->th = GNUNET_DV_send (plugin->dvh,
                           &session->sender,
                           msg,
                           &send_finished,
@@ -527,11 +612,12 @@ dv_plugin_send (void *cls,
  * from the given peer and cancel all previous transmissions
  * (and their continuations).
  *
- * @param cls closure
+ * @param cls closure with the `struct Plugin *`
  * @param target peer from which to disconnect
  */
 static void
-dv_plugin_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
+dv_plugin_disconnect_peer (void *cls,
+                           const struct GNUNET_PeerIdentity *target)
 {
   struct Plugin *plugin = cls;
   struct Session *session;
@@ -551,10 +637,45 @@ dv_plugin_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
     if (NULL != pr->transmit_cont)
       pr->transmit_cont (pr->transmit_cont_cls,
                         &session->sender,
-                        GNUNET_SYSERR, 0, 0);
+                        GNUNET_SYSERR,
+                         pr->size, 0);
+    GNUNET_free (pr);
+  }
+  session->active = GNUNET_NO;
+}
+
+
+/**
+ * Function that can be used to force the plugin to disconnect
+ * from the given peer and cancel all previous transmissions
+ * (and their continuations).
+ *
+ * @param cls closure with the `struct Plugin *`
+ * @param session which session to disconnect
+ * @return #GNUNET_OK
+ */
+static int
+dv_plugin_disconnect_session (void *cls,
+                              struct Session *session)
+{
+  struct PendingRequest *pr;
+
+  while (NULL != (pr = session->pr_head))
+  {
+    GNUNET_CONTAINER_DLL_remove (session->pr_head,
+                                session->pr_tail,
+                                pr);
+    GNUNET_DV_send_cancel (pr->th);
+    pr->th = NULL;
+    if (NULL != pr->transmit_cont)
+      pr->transmit_cont (pr->transmit_cont_cls,
+                        &session->sender,
+                        GNUNET_SYSERR,
+                         pr->size, 0);
     GNUNET_free (pr);
   }
   session->active = GNUNET_NO;
+  return GNUNET_OK;
 }
 
 
@@ -570,10 +691,11 @@ dv_plugin_disconnect (void *cls, const struct GNUNET_PeerIdentity *target)
  * @param numeric should (IP) addresses be displayed in numeric form?
  * @param timeout after how long should we give up?
  * @param asc function to call on each string
- * @param asc_cls closure for asc
+ * @param asc_cls closure for @a asc
  */
 static void
-dv_plugin_address_pretty_printer (void *cls, const char *type, const void *addr,
+dv_plugin_address_pretty_printer (void *cls, const char *type,
+                                  const void *addr,
                                   size_t addrlen, int numeric,
                                   struct GNUNET_TIME_Relative timeout,
                                   GNUNET_TRANSPORT_AddressStringCallback asc,
@@ -581,8 +703,16 @@ dv_plugin_address_pretty_printer (void *cls, const char *type, const void *addr,
 {
   if ( (0 == addrlen) &&
        (0 == strcmp (type, "dv")) )
-    asc (asc_cls, "dv");
-  asc (asc_cls, NULL);
+    asc (asc_cls,
+         "dv",
+         GNUNET_OK);
+  else
+    asc (asc_cls,
+         NULL,
+         GNUNET_SYSERR);
+  asc (asc_cls,
+       NULL,
+       GNUNET_OK);
 }
 
 
@@ -591,12 +721,13 @@ dv_plugin_address_pretty_printer (void *cls, const char *type, const void *addr,
  *
  * @param cls closure
  * @param addr the (hopefully) DV address
- * @param addrlen the length of the address
- *
+ * @param addrlen the length of the @a addr
  * @return string representing the DV address
  */
 static const char *
-dv_plugin_address_to_string (void *cls, const void *addr, size_t addrlen)
+dv_plugin_address_to_string (void *cls,
+                             const void *addr,
+                             size_t addrlen)
 {
   if (0 != addrlen)
   {
@@ -618,13 +749,15 @@ dv_plugin_address_to_string (void *cls, const void *addr, size_t addrlen)
  *
  * @param cls closure
  * @param addr pointer to the address
- * @param addrlen length of addr
- * @return GNUNET_OK if this is a plausible address for this peer
- *         and transport, GNUNET_SYSERR if not
+ * @param addrlen length of @a addr
+ * @return #GNUNET_OK if this is a plausible address for this peer
+ *         and transport, #GNUNET_SYSERR if not
  *
  */
 static int
-dv_plugin_check_address (void *cls, const void *addr, size_t addrlen)
+dv_plugin_check_address (void *cls,
+                         const void *addr,
+                         size_t addrlen)
 {
   if (0 != addrlen)
     return GNUNET_SYSERR;
@@ -665,11 +798,11 @@ dv_get_session (void *cls,
  *
  * @param cls closure ('struct Plugin*')
  * @param addr string address
- * @param addrlen length of the address including \0 termination
+ * @param addrlen length of the @a addr including \0 termination
  * @param buf location to store the buffer
- *        If the function returns GNUNET_SYSERR, its contents are undefined.
+ *        If the function returns #GNUNET_SYSERR, its contents are undefined.
  * @param added length of created address
- * @return GNUNET_OK on success, GNUNET_SYSERR on failure
+ * @return #GNUNET_OK on success, #GNUNET_SYSERR on failure
  */
 static int
 dv_plugin_string_to_address (void *cls,
@@ -688,6 +821,25 @@ dv_plugin_string_to_address (void *cls,
 }
 
 
+/**
+ * Function that will be called whenever the transport service wants to
+ * notify the plugin that a session is still active and in use and
+ * therefore the session timeout for this session has to be updated
+ *
+ * @param cls closure (`struct Plugin *`)
+ * @param peer which peer was the session for
+ * @param session which session is being updated
+ */
+static void
+dv_plugin_update_session_timeout (void *cls,
+                                  const struct GNUNET_PeerIdentity *peer,
+                                  struct Session *session)
+{
+  /* DV currently doesn't time out like "normal" plugins,
+     so it should be safe to do nothing, right?
+     (or should we add an internal timeout?) */
+}
+
 
 /**
  * Function to obtain the network type for a session
@@ -695,7 +847,7 @@ dv_plugin_string_to_address (void *cls,
  * used by the next hop here.  Or find some other way
  * to properly allow ATS-DV resource allocation.
  *
- * @param cls closure ('struct Plugin*')
+ * @param cls closure (`struct Plugin *`)
  * @param session the session
  * @return the network type
  */
@@ -708,6 +860,79 @@ dv_get_network (void *cls,
 }
 
 
+/**
+ * Function that is called to get the keepalive factor.
+ * #GNUNET_CONSTANTS_IDLE_CONNECTION_TIMEOUT is divided by this number to
+ * calculate the interval between keepalive packets.
+ *
+ * @param cls closure with the `struct Plugin`
+ * @return keepalive factor
+ */
+static unsigned int
+dv_plugin_query_keepalive_factor (void *cls)
+{
+  return 3;
+}
+
+
+/**
+ * Return information about the given session to the
+ * monitor callback.
+ *
+ * @param cls the `struct Plugin` with the monitor callback (`sic`)
+ * @param peer peer we send information about
+ * @param value our `struct Session` to send information about
+ * @return #GNUNET_OK (continue to iterate)
+ */
+static int
+send_session_info_iter (void *cls,
+                        const struct GNUNET_PeerIdentity *peer,
+                        void *value)
+{
+  struct Plugin *plugin = cls;
+  struct Session *session = value;
+
+  if (GNUNET_YES != session->active)
+    return GNUNET_OK;
+  notify_session_monitor (plugin,
+                          session,
+                          GNUNET_TRANSPORT_SS_UP);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Begin monitoring sessions of a plugin.  There can only
+ * be one active monitor per plugin (i.e. if there are
+ * multiple monitors, the transport service needs to
+ * multiplex the generated events over all of them).
+ *
+ * @param cls closure of the plugin
+ * @param sic callback to invoke, NULL to disable monitor;
+ *            plugin will being by iterating over all active
+ *            sessions immediately and then enter monitor mode
+ * @param sic_cls closure for @a sic
+ */
+static void
+dv_plugin_setup_monitor (void *cls,
+                          GNUNET_TRANSPORT_SessionInfoCallback sic,
+                          void *sic_cls)
+{
+  struct Plugin *plugin = cls;
+
+  plugin->sic = sic;
+  plugin->sic_cls = sic_cls;
+  if (NULL != sic)
+  {
+    GNUNET_CONTAINER_multipeermap_iterate (plugin->sessions,
+                                           &send_session_info_iter,
+                                           plugin);
+    /* signal end of first iteration */
+    sic (sic_cls, NULL, NULL);
+  }
+}
+
+
 /**
  * Entry point for the plugin.
  *
@@ -742,13 +967,17 @@ libgnunet_plugin_transport_dv_init (void *cls)
   api = GNUNET_new (struct GNUNET_TRANSPORT_PluginFunctions);
   api->cls = plugin;
   api->send = &dv_plugin_send;
-  api->disconnect = &dv_plugin_disconnect;
+  api->disconnect_peer = &dv_plugin_disconnect_peer;
+  api->disconnect_session = &dv_plugin_disconnect_session;
   api->address_pretty_printer = &dv_plugin_address_pretty_printer;
   api->check_address = &dv_plugin_check_address;
   api->address_to_string = &dv_plugin_address_to_string;
   api->string_to_address = &dv_plugin_string_to_address;
+  api->query_keepalive_factor = &dv_plugin_query_keepalive_factor;
   api->get_session = &dv_get_session;
   api->get_network = &dv_get_network;
+  api->update_session_timeout = &dv_plugin_update_session_timeout;
+  api->setup_monitor = &dv_plugin_setup_monitor;
   return api;
 }
 
@@ -759,7 +988,7 @@ libgnunet_plugin_transport_dv_init (void *cls)
  * @param cls NULL
  * @param key unused
  * @param value session to free
- * @return GNUNET_OK (continue to iterate)
+ * @return #GNUNET_OK (continue to iterate)
  */
 static int
 free_session_iterator (void *cls,