/**
 * Iterate over all nodes in the tree.
 *
 * @param tree Tree to use..
 * @param cb Callback to call over each child.
 * @param cb_cls Closure for @c cb.
 *
 * TODO: recursive implementation? (s/heap/stack/g)
 */
void
tree_iterate_all (struct MeshTunnelTree *tree,
                  MeshWholeTreeCallback cb,
                  void *cb_cls)
{
  struct MeshTunnelTreeNode *parent;
  struct MeshTunnelTreeNode *n;
  struct MeshTreePendingNode *head;
  struct MeshTreePendingNode *tail;
  struct MeshTreePendingNode *pending;

  cb (cb_cls, tree->root->peer, 0);
  pending = GNUNET_malloc (sizeof (struct MeshTreePendingNode));
  pending->node = tree->root;
  head = tail = NULL;
  GNUNET_CONTAINER_DLL_insert (head, tail, pending);

  while (NULL != head)
  {
    pending = head;
    parent = pending->node;
    GNUNET_CONTAINER_DLL_remove (head, tail, pending);
    GNUNET_free (pending);
    for (n = parent->children_head; NULL != n; n = n->next)
    {
      cb (cb_cls, n->peer, parent->peer);
      pending = GNUNET_malloc (sizeof (struct MeshTreePendingNode));
      pending->node = n;
      /* Insert_tail: breadth first, Insert: depth first */
      GNUNET_CONTAINER_DLL_insert (head, tail, pending);
    }
  }
}

/**
 * Notifies a tree that a connection it might be using is broken.
 * Marks all peers down the paths as disconnected and notifies the client.
 *
 * @param t Tree to use.
 * @param p1 Short id of one of the peers (order unimportant)
 * @param p2 Short id of one of the peers (order unimportant)
 * @param cb Function to call for every peer that is marked as disconnected.
 * @param cbcls Closure for cb.
 *
 * @return Short ID of the first disconnected peer in the tree.
 */
GNUNET_PEER_Id
tree_notify_connection_broken (struct MeshTunnelTree *t, GNUNET_PEER_Id p1,
                               GNUNET_PEER_Id p2, MeshTreeCallback cb,
                               void *cbcls)
{
  struct MeshTunnelTreeNode *n;
  struct MeshTunnelTreeNode *c;

  n = tree_find_peer (t, p1);
  if (NULL == n)
    return 0;
  if (NULL != n->parent && n->parent->peer == p2)
  {
    tree_mark_peers_disconnected (t, n, cb, cbcls);
    GNUNET_CONTAINER_DLL_remove (n->parent->children_head,
                                 n->parent->children_tail, n);
    GNUNET_CONTAINER_DLL_insert (t->disconnected_head, t->disconnected_tail, n);
    return p1;
  }
  for (c = n->children_head; NULL != c; c = c->next)
  {
    if (c->peer == p2)
    {
      tree_mark_peers_disconnected (t, c, cb, cbcls);
      GNUNET_CONTAINER_DLL_remove (n->children_head, n->children_tail, c);
      GNUNET_CONTAINER_DLL_insert (t->disconnected_head, t->disconnected_tail,
                                   c);
      return p2;
    }
  }
  return 0;
}

static void
nat_add_address (void *cls, int add_remove, const struct sockaddr *addr,
                 socklen_t addrlen)
{
  struct Plugin *plugin = cls;
  struct IPv4HttpAddressWrapper *w_t4 = NULL;
  struct IPv6HttpAddressWrapper *w_t6 = NULL;
  int af;

  af = addr->sa_family;
  switch (af)
  {
  case AF_INET:
    w_t4 = find_address (plugin, addr, addrlen);
    if (w_t4 == NULL)
    {
      struct sockaddr_in *a4 = (struct sockaddr_in *) addr;
      w_t4 = GNUNET_malloc (sizeof (struct IPv4HttpAddressWrapper));
      memcpy (&w_t4->addr.ipv4_addr, &a4->sin_addr, sizeof (struct in_addr));
      w_t4->addr.u4_port = a4->sin_port;

      GNUNET_CONTAINER_DLL_insert (plugin->ipv4_addr_head,
                                   plugin->ipv4_addr_tail, w_t4);

    GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, plugin->name,
                     "Notifying transport to add IPv4 address `%s'\n",
                     http_plugin_address_to_string (NULL, &w_t4->addr,
                                                    sizeof (struct
                                                            IPv4HttpAddress)));
    plugin->env->notify_address (plugin->env->cls, add_remove, &w_t4->addr,
                                 sizeof (struct IPv4HttpAddress));
    }
    break;
  case AF_INET6:
    w_t6 = find_address (plugin, addr, addrlen);
    if (w_t6 == NULL)
    {
      w_t6 = GNUNET_malloc (sizeof (struct IPv6HttpAddressWrapper));
      struct sockaddr_in6 *a6 = (struct sockaddr_in6 *) addr;
      memcpy (&w_t6->addr6.ipv6_addr, &a6->sin6_addr, sizeof (struct in6_addr));
      w_t6->addr6.u6_port = a6->sin6_port;

      GNUNET_CONTAINER_DLL_insert (plugin->ipv6_addr_head,
                                   plugin->ipv6_addr_tail, w_t6);

      GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, plugin->name,
                     "Notifying transport to add IPv6 address `%s'\n",
                     http_plugin_address_to_string (NULL, &w_t6->addr6,
                                                    sizeof (struct
                                                            IPv6HttpAddress)));
      plugin->env->notify_address (plugin->env->cls, add_remove, &w_t6->addr6,
                                 sizeof (struct IPv6HttpAddress));
    }
    break;
  default:
    return;
  }

}

/**
 * Start monitoring the local DHT service.
 *
 * @param handle Handle to the DHT service.
 * @param type Type of blocks that are of interest.
 * @param key Key of data of interest, NULL for all.
 * @param get_cb Callback to process monitored get messages.
 * @param get_resp_cb Callback to process monitored get response messages.
 * @param put_cb Callback to process monitored put messages.
 * @param cb_cls Closure for cb.
 *
 * @return Handle to stop monitoring.
 */
struct GNUNET_DHT_MonitorHandle *
GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle,
                          enum GNUNET_BLOCK_Type type,
                          const GNUNET_HashCode *key,
                          GNUNET_DHT_MonitorGetCB get_cb,
                          GNUNET_DHT_MonitorGetRespCB get_resp_cb,
                          GNUNET_DHT_MonitorPutCB put_cb,
                          void *cb_cls)
{
  struct GNUNET_DHT_MonitorHandle *h;
  struct GNUNET_DHT_MonitorStartStopMessage *m;
  struct PendingMessage *pending;

  h = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorHandle));
  GNUNET_CONTAINER_DLL_insert(handle->monitor_head, handle->monitor_tail, h);

  h->get_cb = get_cb;
  h->get_resp_cb = get_resp_cb;
  h->put_cb = put_cb;
  h->cb_cls = cb_cls;
  h->type = type;
  h->dht_handle = handle;
  if (NULL != key)
  {
    h->key = GNUNET_malloc (sizeof(GNUNET_HashCode));
    memcpy (h->key, key, sizeof(GNUNET_HashCode));
  }

  pending = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorStartStopMessage) +
                           sizeof (struct PendingMessage));
  m = (struct GNUNET_DHT_MonitorStartStopMessage *) &pending[1];
  pending->msg = &m->header;
  pending->handle = handle;
  pending->free_on_send = GNUNET_YES;
  m->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_START);
  m->header.size = htons (sizeof (struct GNUNET_DHT_MonitorStartStopMessage));
  m->type = htonl(type);
  m->get = htons(NULL != get_cb);
  m->get_resp = htons(NULL != get_resp_cb);
  m->put = htons(NULL != put_cb);
  if (NULL != key) {
    m->filter_key = htons(1);
    memcpy (&m->key, key, sizeof(GNUNET_HashCode));
  }
  GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
                               pending);
  pending->in_pending_queue = GNUNET_YES;
  process_pending_messages (handle);

  return h;
}

/**
 * Try again to connect to the identity service.
 *
 * @param cls handle to the identity service.
 * @param tc scheduler context
 */
static void
reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct GNUNET_IDENTITY_Handle *h = cls;
  struct GNUNET_IDENTITY_Operation *op;
  struct GNUNET_MessageHeader msg;

  h->reconnect_task = NULL;
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Connecting to identity service.\n");
  GNUNET_assert (NULL == h->client);
  h->client = GNUNET_CLIENT_connect ("identity", h->cfg);
  GNUNET_assert (NULL != h->client);
  if ( (NULL == h->op_head) ||
       (GNUNET_MESSAGE_TYPE_IDENTITY_START != ntohs (h->op_head->msg->type)) )
  {
    op = GNUNET_malloc (sizeof (struct GNUNET_IDENTITY_Operation) +
			sizeof (struct GNUNET_MessageHeader));
    op->h = h;
    op->msg = (const struct GNUNET_MessageHeader *) &op[1];
    msg.size = htons (sizeof (msg));
    msg.type = htons (GNUNET_MESSAGE_TYPE_IDENTITY_START);
    memcpy (&op[1], &msg, sizeof (msg));
    GNUNET_CONTAINER_DLL_insert (h->op_head,
				 h->op_tail,
				 op);
  }
  transmit_next (h);
  GNUNET_assert (NULL != h->th);
}

static int
iface_proc (void *cls, const char *name, int isDefault,
            const struct sockaddr *addr, const struct sockaddr *broadcast_addr,
            const struct sockaddr *netmask, socklen_t addrlen)
{
  struct Plugin *plugin = cls;

  if (addr != NULL)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "address %s for interface %s %p\n ",
                GNUNET_a2s (addr, addrlen), name, addr);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "broadcast address %s for interface %s %p\n ",
                GNUNET_a2s (broadcast_addr, addrlen), name, broadcast_addr);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "netmask %s for interface %s %p\n ",
                GNUNET_a2s (netmask, addrlen), name, netmask);

    /* Collecting broadcast addresses */
    if (broadcast_addr != NULL)
    {
      struct BroadcastAddress *ba =
          GNUNET_malloc (sizeof (struct BroadcastAddress));
      ba->addr = GNUNET_malloc (addrlen);
      memcpy (ba->addr, broadcast_addr, addrlen);
      ba->addrlen = addrlen;
      GNUNET_CONTAINER_DLL_insert (plugin->ipv4_broadcast_head,
                                   plugin->ipv4_broadcast_tail, ba);
    }
  }
  return GNUNET_OK;
}

/**
 * Store an item in the datastore.
 *
 * @param cls closure
 * @param key key for the item
 * @param size number of bytes in data
 * @param data content stored
 * @param type type of the content
 * @param priority priority of the content
 * @param anonymity anonymity-level for the content
 * @param replication replication-level for the content
 * @param expiration expiration time for the content
 * @param msg set to error message
 * @return GNUNET_OK on success
 */
static int
heap_plugin_put (void *cls, 
		 const struct GNUNET_HashCode * key, 
		 uint32_t size,
		 const void *data, 
		 enum GNUNET_BLOCK_Type type,
		 uint32_t priority, uint32_t anonymity,
		 uint32_t replication,
		 struct GNUNET_TIME_Absolute expiration, char **msg)
{
  struct Plugin *plugin = cls;
  struct Value *value;

  value = GNUNET_malloc (sizeof (struct Value) + size);
  value->key = *key;
  value->data = &value[1];
  value->expire_heap = GNUNET_CONTAINER_heap_insert (plugin->by_expiration,
						     value,
						     expiration.abs_value);
  value->replication_heap = GNUNET_CONTAINER_heap_insert (plugin->by_replication,
							  value,
							  replication);
  value->expiration = expiration;
  if (0 == anonymity)
  {
    struct ZeroAnonByType *zabt;

    for (zabt = plugin->zero_head; NULL != zabt; zabt = zabt->next)
      if (zabt->type == type)
	break;
    if (NULL == zabt)
    {
      zabt = GNUNET_malloc (sizeof (struct ZeroAnonByType));
      zabt->type = type;
      GNUNET_CONTAINER_DLL_insert (plugin->zero_head,
				   plugin->zero_tail,
				   zabt);
    }
    if (zabt->array_size == zabt->array_pos)
    {
      GNUNET_array_grow (zabt->array,
			 zabt->array_size,
			 zabt->array_size * 2 + 4);
    }
    value->zero_anon_offset = zabt->array_pos;
    zabt->array[zabt->array_pos++] = value;
  }
  value->size = size;
  value->priority = priority;
  value->anonymity = anonymity;
  value->replication = replication;
  value->type = type;
  memcpy (&value[1], data, size);
  GNUNET_CONTAINER_multihashmap_put (plugin->keyvalue,
				     &value->key,
				     value,
				     GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
  plugin->size += size;
  return GNUNET_OK;
}

/**
 * Add or remove an address from this peer's HELLO message.
 *
 * @param addremove GNUNET_YES to add, GNUNET_NO to remove
 * @param address address to add or remove
 */
void
GST_hello_modify_addresses (int addremove,
                            const struct GNUNET_HELLO_Address *address)
{
  struct OwnAddressList *al;

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              (addremove ==
               GNUNET_YES) ? "Adding `%s' to the set of our addresses\n" :
              "Removing `%s' from the set of our addresses\n",
              GST_plugins_a2s (address));
  GNUNET_assert (address != NULL);
  if (GNUNET_NO == addremove)
  {
    for (al = oal_head; al != NULL; al = al->next)
      if (0 == GNUNET_HELLO_address_cmp (address, al->address))
      {
        GNUNET_CONTAINER_DLL_remove (oal_head, oal_tail, al);
        GNUNET_HELLO_address_free (al->address);
        GNUNET_free (al);
        refresh_hello ();
        return;
      }
    /* address to be removed not found!? */
    GNUNET_break (0);
    return;
  }
  al = GNUNET_malloc (sizeof (struct OwnAddressList));
  GNUNET_CONTAINER_DLL_insert (oal_head, oal_tail, al);
  al->address = GNUNET_HELLO_address_copy (address);
  refresh_hello ();
}

/**
 * Handler for monitor start messages
 *
 * @param cls closure for the service
 * @param client the client we received this message from
 * @param message the actual message received
 *
 */
static void
handle_dht_local_monitor (void *cls, struct GNUNET_SERVER_Client *client,
                          const struct GNUNET_MessageHeader *message)
{
  struct ClientMonitorRecord *r;
  const struct GNUNET_DHT_MonitorStartStopMessage *msg;

  msg = (struct GNUNET_DHT_MonitorStartStopMessage *) message;
  r = GNUNET_new (struct ClientMonitorRecord);

  r->client = find_active_client(client);
  r->type = ntohl(msg->type);
  r->get = ntohs(msg->get);
  r->get_resp = ntohs(msg->get_resp);
  r->put = ntohs(msg->put);
  if (0 == ntohs(msg->filter_key))
      r->key = NULL;
  else
  {
    r->key = GNUNET_new (struct GNUNET_HashCode);
    memcpy (r->key, &msg->key, sizeof (struct GNUNET_HashCode));
  }
  GNUNET_CONTAINER_DLL_insert (monitor_head, monitor_tail, r);
  GNUNET_SERVER_receive_done (client, GNUNET_OK);
}

/**
 * We've validated the hash of the file we're about to index.  Signal
 * success to the client and update our internal data structures.
 *
 * @param ii the index info entry for the request
 */
static void
signal_index_ok (struct IndexInfo *ii)
{
  struct IndexInfo *ir;
  if (GNUNET_SYSERR ==
      GNUNET_CONTAINER_multihashmap_put (ifm, &ii->file_id,
                                         ii,
                                         GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
  {
    ir = GNUNET_CONTAINER_multihashmap_get (ifm,
					    &ii->file_id);
    GNUNET_assert (NULL != ir);
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                _
                ("Index request received for file `%s' is already indexed as `%s'.  Permitting anyway.\n"),
                ii->filename,
		ir->filename);           
    GNUNET_SERVER_transmit_context_append_data (ii->tc, NULL, 0,
                                                GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
    GNUNET_SERVER_transmit_context_run (ii->tc, GNUNET_TIME_UNIT_MINUTES);
    GNUNET_free (ii);
    return;
  }
  GNUNET_CONTAINER_DLL_insert (indexed_files_head,
			       indexed_files_tail,
			       ii);
  write_index_list ();
  GNUNET_SERVER_transmit_context_append_data (ii->tc, NULL, 0,
                                              GNUNET_MESSAGE_TYPE_FS_INDEX_START_OK);
  GNUNET_SERVER_transmit_context_run (ii->tc, GNUNET_TIME_UNIT_MINUTES);
  ii->tc = NULL;
}

/**
 * Transmit the given message to the client.
 *
 * @param client target of the message
 * @param msg message to transmit, will be freed!
 */
static void
transmit (struct GNUNET_SERVER_Client *client, struct GNUNET_MessageHeader *msg)
{
  struct TransmitCallbackContext *tcc;

  if (GNUNET_YES == cleaning_done)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _("Shutdown in progress, aborting transmission.\n"));
    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
    GNUNET_free (msg);
    return;
  }
  tcc = GNUNET_new (struct TransmitCallbackContext);
  tcc->msg = msg;
  tcc->client = client;
  if (NULL ==
      (tcc->th =
       GNUNET_SERVER_notify_transmit_ready (client, ntohs (msg->size),
                                            GNUNET_TIME_UNIT_FOREVER_REL,
                                            &transmit_callback, tcc)))
  {
    GNUNET_break (0);
    GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
    GNUNET_free (msg);
    GNUNET_free (tcc);
    return;
  }
  GNUNET_SERVER_client_keep (client);
  GNUNET_CONTAINER_DLL_insert (tcc_head, tcc_tail, tcc);
}

/**
 * Process the given file from the "EGODIR".  Parses the file
 * and creates the respective 'struct Ego' in memory.
 *
 * @param cls NULL
 * @param filename name of the file to parse
 * @return #GNUNET_OK to continue to iterate,
 *  #GNUNET_NO to stop iteration with no error,
 *  #GNUNET_SYSERR to abort iteration with error!
 */
static int
process_ego_file (void *cls,
		  const char *filename)
{
  struct Ego *ego;
  const char *fn;

  fn = strrchr (filename, (int) DIR_SEPARATOR);
  if (NULL == fn)
  {
    GNUNET_break (0);
    return GNUNET_OK;
  }
  ego = GNUNET_new (struct Ego);
  ego->pk = GNUNET_CRYPTO_ecdsa_key_create_from_file (filename);
  if (NULL == ego->pk)
  {
    GNUNET_free (ego);
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _("Failed to parse ego information in `%s'\n"),
                filename);
    return GNUNET_OK;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Loaded ego `%s'\n",
	      fn + 1);
  ego->identifier = GNUNET_strdup (fn + 1);
  GNUNET_CONTAINER_DLL_insert (ego_head,
			       ego_tail,
			       ego);
  return GNUNET_OK;
}

/**
 * Initiate a connection from p1 to p2 by offering p1 p2's HELLO message
 *
 * Remarks: start_peer's notify_connect callback can be called before.
 *
 * @param tth transport testing handle
 * @param p1 peer 1
 * @param p2 peer 2
 * @param cb the callback to call when both peers notified that they are connected
 * @param cls callback cls
 * @return a connect request handle
 */
GNUNET_TRANSPORT_TESTING_ConnectRequest
GNUNET_TRANSPORT_TESTING_connect_peers (struct GNUNET_TRANSPORT_TESTING_handle *tth,
                                        struct PeerContext *p1,
                                        struct PeerContext *p2,
                                        GNUNET_TRANSPORT_TESTING_connect_cb cb,
                                        void *cls)
{
  GNUNET_assert (tth != NULL);

  struct ConnectingContext *cc =
      GNUNET_new (struct ConnectingContext);

  GNUNET_assert (p1 != NULL);
  GNUNET_assert (p2 != NULL);
  cc->p1 = p1;
  cc->p2 = p2;
  cc->cb = cb;
  if (cls != NULL)
    cc->cb_cls = cls;
  else
    cc->cb_cls = cc;
  cc->th_p1 = p1->th;
  cc->th_p2 = p2->th;
  GNUNET_assert (cc->th_p1 != NULL);
  GNUNET_assert (cc->th_p2 != NULL);
  GNUNET_CONTAINER_DLL_insert (tth->cc_head, tth->cc_tail, cc);
  cc->tct = GNUNET_SCHEDULER_add_now (&try_connect, cc);
  GNUNET_log_from (GNUNET_ERROR_TYPE_DEBUG, "transport-testing",
                   "New connect request %p\n", cc);

  return cc;
}

/**
 * Queue a request from a client for transmission to a particular peer.
 *
 * @param car request to queue; this handle is then shared between
 *         the caller (CLIENTS subsystem) and SESSIONS and must not
 *         be released by either until either #GSC_SESSIONS_dequeue(),
 *         #GSC_SESSIONS_transmit() or #GSC_CLIENTS_failed()
 *         have been invoked on it
 */
void
GSC_SESSIONS_queue_request (struct GSC_ClientActiveRequest *car)
{
  struct Session *session;

  session = find_session (&car->target);
  if (NULL == session)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Dropped client request for transmission (am disconnected)\n");
    GNUNET_break (0);           /* should have been rejected earlier */
    GSC_CLIENTS_reject_request (car,
                                GNUNET_NO);
    return;
  }
  if (car->msize > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    GSC_CLIENTS_reject_request (car,
                                GNUNET_YES);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Received client transmission request. queueing\n");
  GNUNET_CONTAINER_DLL_insert (session->active_client_request_head,
                               session->active_client_request_tail,
                               car);
  try_transmission (session);
}

/**
 * Add another child node to the tree.
 *
 * @param parent parent of the child, NULL for top level
 * @param filename name of the file or directory
 * @param is_directory GNUNET_YES for directories
 * @return new entry that was just created
 */
static struct GNUNET_FS_ShareTreeItem *
expand_tree (struct GNUNET_FS_ShareTreeItem *parent,
	     const char *filename,
	     int is_directory)
{
  struct GNUNET_FS_ShareTreeItem *chld;
  size_t slen;

  chld = GNUNET_malloc (sizeof (struct GNUNET_FS_ShareTreeItem));
  chld->parent = parent;
  chld->filename = GNUNET_strdup (filename);
  GNUNET_asprintf (&chld->short_filename,
		   "%s%s",
		   GNUNET_STRINGS_get_short_name (filename),
		   is_directory == GNUNET_YES ? "/" : "");
  /* make sure we do not end with '//' */
  slen = strlen (chld->short_filename);
  if ( (slen >= 2) &&
       (chld->short_filename[slen-1] == '/') &&
       (chld->short_filename[slen-2] == '/') )
    chld->short_filename[slen-1] = '\0';
  chld->is_directory = is_directory;
  if (NULL != parent)
      GNUNET_CONTAINER_DLL_insert (parent->children_head,
				   parent->children_tail,
				   chld);  
  return chld;
}

static void
resolve_validation_address (const struct GNUNET_PeerIdentity *id,
    const struct GNUNET_HELLO_Address *address, int numeric,
    struct GNUNET_TIME_Absolute last_validation,
    struct GNUNET_TIME_Absolute valid_until,
    struct GNUNET_TIME_Absolute next_validation,
    enum GNUNET_TRANSPORT_ValidationState state)
{
  struct ValidationResolutionContext *vc;

  vc = GNUNET_new (struct ValidationResolutionContext);
  GNUNET_assert(NULL != vc);
  GNUNET_CONTAINER_DLL_insert(vc_head, vc_tail, vc);
  address_resolutions++;

  vc->id = (*id);
  vc->transport = GNUNET_strdup(address->transport_name);
  vc->addrcp = GNUNET_HELLO_address_copy (address);
  vc->printed = GNUNET_NO;
  vc->state = state;
  vc->last_validation = last_validation;
  vc->valid_until = valid_until;
  vc->next_validation = next_validation;

  /* Resolve address to string */
  vc->asc = GNUNET_TRANSPORT_address_to_string (cfg, address, numeric,
      RESOLUTION_TIMEOUT, &process_validation_string, vc);
}

/**
 * Force plugin to terminate session due to communication
 * issue.
 *
 * @param plugin_name name of the plugin
 * @param session session to termiante
 */
static void
kill_session (const char *plugin_name,
              struct GNUNET_ATS_Session *session)
{
  struct GNUNET_TRANSPORT_PluginFunctions *plugin;
  struct GNUNET_ATS_SessionKiller *sk;

  for (sk = sk_head; NULL != sk; sk = sk->next)
    if (sk->session == session)
      return;
  plugin = GST_plugins_find (plugin_name);
  if (NULL == plugin)
  {
    GNUNET_break(0);
    return;
  }
  /* need to issue disconnect asynchronously */
  sk = GNUNET_new (struct GNUNET_ATS_SessionKiller);
  sk->session = session;
  sk->plugin = plugin;
  sk->task = GNUNET_SCHEDULER_add_now (&kill_session_task, sk);
  GNUNET_CONTAINER_DLL_insert (sk_head,
                               sk_tail,
                               sk);
}

/**
 * Stop monitoring.
 *
 * @param handle The handle to the monitor request returned by monitor_start.
 *
 * On return get_handle will no longer be valid, caller must not use again!!!
 */
void
GNUNET_DHT_monitor_stop (struct GNUNET_DHT_MonitorHandle *handle)
{
  struct GNUNET_DHT_MonitorStartStopMessage *m;
  struct PendingMessage *pending;

  GNUNET_CONTAINER_DLL_remove (handle->dht_handle->monitor_head,
                               handle->dht_handle->monitor_tail,
                               handle);

  pending = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorStartStopMessage) +
                           sizeof (struct PendingMessage));
  m = (struct GNUNET_DHT_MonitorStartStopMessage *) &pending[1];
  pending->msg = &m->header;
  pending->handle = handle->dht_handle;
  pending->free_on_send = GNUNET_YES;
  m->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_STOP);
  m->header.size = htons (sizeof (struct GNUNET_DHT_MonitorStartStopMessage));
  m->type = htonl(handle->type);
  m->get = htons(NULL != handle->get_cb);
  m->get_resp = htons(NULL != handle->get_resp_cb);
  m->put = htons(NULL != handle->put_cb);
  if (NULL != handle->key) {
    m->filter_key = htons(1);
    memcpy (&m->key, handle->key, sizeof(GNUNET_HashCode));
  }
  GNUNET_CONTAINER_DLL_insert (handle->dht_handle->pending_head,
                               handle->dht_handle->pending_tail,
                               pending);
  pending->in_pending_queue = GNUNET_YES;
  process_pending_messages (handle->dht_handle);
  
  GNUNET_free_non_null (handle->key);
  GNUNET_free (handle);
}

/**
 * Look for a block by directly contacting a particular peer.
 *
 * @param target peer that should have the block
 * @param query hash to query for the block
 * @param type desired type for the block
 * @param proc function to call with result
 * @param proc_cls closure for 'proc'
 * @return handle to cancel the operation
 */
struct GSF_StreamRequest *
GSF_stream_query (const struct GNUNET_PeerIdentity *target,
		  const struct GNUNET_HashCode *query,
		  enum GNUNET_BLOCK_Type type,
		  GSF_StreamReplyProcessor proc, void *proc_cls)
{
  struct StreamHandle *sh;
  struct GSF_StreamRequest *sr;

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Preparing to send query for %s via stream to %s\n",
	      GNUNET_h2s (query),
	      GNUNET_i2s (target));
  sh = get_stream (target);
  sr = GNUNET_malloc (sizeof (struct GSF_StreamRequest));
  sr->sh = sh;
  sr->proc = proc;
  sr->proc_cls = proc_cls;
  sr->type = type;
  sr->query = *query;
  GNUNET_CONTAINER_DLL_insert (sh->pending_head,
			       sh->pending_tail,
			       sr);
  if (GNUNET_YES == sh->is_ready)
    transmit_pending (sh);
  return sr;
}

/**
 * Start monitoring the local DHT service.
 *
 * @param handle Handle to the DHT service.
 * @param type Type of blocks that are of interest.
 * @param key Key of data of interest, NULL for all.
 * @param get_cb Callback to process monitored get messages.
 * @param get_resp_cb Callback to process monitored get response messages.
 * @param put_cb Callback to process monitored put messages.
 * @param cb_cls Closure for callbacks.
 * @return Handle to stop monitoring.
 */
struct GNUNET_DHT_MonitorHandle *
GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle,
                          enum GNUNET_BLOCK_Type type,
                          const struct GNUNET_HashCode *key,
                          GNUNET_DHT_MonitorGetCB get_cb,
                          GNUNET_DHT_MonitorGetRespCB get_resp_cb,
                          GNUNET_DHT_MonitorPutCB put_cb,
                          void *cb_cls)
{
  struct GNUNET_DHT_MonitorHandle *h;
  struct GNUNET_DHT_MonitorStartStopMessage *m;
  struct PendingMessage *pending;

  h = GNUNET_new (struct GNUNET_DHT_MonitorHandle);
  GNUNET_CONTAINER_DLL_insert(handle->monitor_head, handle->monitor_tail, h);

  h->get_cb = get_cb;
  h->get_resp_cb = get_resp_cb;
  h->put_cb = put_cb;
  h->cb_cls = cb_cls;
  h->type = type;
  h->dht_handle = handle;
  if (NULL != key)
  {
    h->key = GNUNET_new (struct GNUNET_HashCode);
    *h->key = *key;
  }