/**
 * Function to process DHT string to regex matching.
 * Called on each result obtained for the DHT search.
 *
 * @param cls Closure (search context).
 * @param exp When will this value expire.
 * @param key Key of the result.
 * @param get_path Path of the get request.
 * @param get_path_length Lenght of get_path.
 * @param put_path Path of the put request.
 * @param put_path_length Length of the put_path.
 * @param type Type of the result.
 * @param size Number of bytes in data.
 * @param data Pointer to the result data.
 */
static void
dht_get_string_accept_handler (void *cls, struct GNUNET_TIME_Absolute exp,
                               const struct GNUNET_HashCode *key,
                               const struct GNUNET_PeerIdentity *get_path,
                               unsigned int get_path_length,
                               const struct GNUNET_PeerIdentity *put_path,
                               unsigned int put_path_length,
                               enum GNUNET_BLOCK_Type type,
                               size_t size, const void *data)
{
  const struct RegexAcceptBlock *block = data;
  struct RegexSearchContext *ctx = cls;
  struct REGEX_INTERNAL_Search *info = ctx->info;

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Regex result accept for %s (key %s)\n",
       info->description, GNUNET_h2s(key));

  GNUNET_STATISTICS_update (info->stats,
			    "# regex accepting blocks found",
                            1, GNUNET_NO);
  GNUNET_STATISTICS_update (info->stats,
			    "# regex accepting block bytes found",
                            size, GNUNET_NO);
  info->callback (info->callback_cls,
                  &block->peer,
                  get_path, get_path_length,
                  put_path, put_path_length);
}

/**
 * Task that triggers a NSE P2P transmission.
 *
 * @param cls the `struct NSEPeerEntry *`
 */
static void
transmit_task_cb (void *cls)
{
  struct NSEPeerEntry *peer_entry = cls;
  unsigned int idx;
  struct GNUNET_MQ_Envelope *env;

  peer_entry->transmit_task = NULL;
  idx = estimate_index;
  if (GNUNET_NO == peer_entry->previous_round)
  {
    idx = (idx + HISTORY_SIZE - 1) % HISTORY_SIZE;
    peer_entry->previous_round = GNUNET_YES;
    peer_entry->transmit_task
      = GNUNET_SCHEDULER_add_delayed (get_transmit_delay (0),
				      &transmit_task_cb,
				      peer_entry);
  }
  if ((0 == ntohl (size_estimate_messages[idx].hop_count)) &&
      (NULL != proof_task))
  {
    GNUNET_STATISTICS_update (stats,
                              "# flood messages not generated (no proof yet)",
                              1,
			      GNUNET_NO);
    return;
  }
  if (0 == ntohs (size_estimate_messages[idx].header.size))
  {
    GNUNET_STATISTICS_update (stats,
                              "# flood messages not generated (lack of history)",
                              1,
			      GNUNET_NO);
    return;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "In round %s, sending to `%s' estimate with %u bits\n",
              GNUNET_STRINGS_absolute_time_to_string (GNUNET_TIME_absolute_ntoh (size_estimate_messages[idx].timestamp)),
              GNUNET_i2s (peer_entry->id),
              (unsigned int) ntohl (size_estimate_messages[idx].matching_bits));
  if (0 == ntohl (size_estimate_messages[idx].hop_count))
    GNUNET_STATISTICS_update (stats,
			      "# flood messages started",
			      1,
			      GNUNET_NO);
  GNUNET_STATISTICS_update (stats,
			    "# flood messages transmitted",
			    1,
                            GNUNET_NO);
#if ENABLE_NSE_HISTOGRAM
  peer_entry->transmitted_messages++;
  peer_entry->last_transmitted_size
    = ntohl(size_estimate_messages[idx].matching_bits);
#endif
  env = GNUNET_MQ_msg_copy (&size_estimate_messages[idx].header);
  GNUNET_MQ_send (peer_entry->mq,
		  env);
}

/**
 * This function is called *before* the DNS request has been 
 * given to a "local" DNS resolver.  Tunneling for DNS requests
 * was enabled, so we now need to send the request via some MESH
 * tunnel to a DNS EXIT for resolution.
 *
 * @param cls closure
 * @param rh request handle to user for reply
 * @param request_length number of bytes in request
 * @param request udp payload of the DNS request
 */
static void 
dns_pre_request_handler (void *cls,
			 struct GNUNET_DNS_RequestHandle *rh,
			 size_t request_length,
			 const char *request)
{
  struct RequestContext *rc;
  size_t mlen;
  struct GNUNET_MessageHeader hdr;
  struct GNUNET_TUN_DnsHeader dns;

  GNUNET_STATISTICS_update (stats,
			    gettext_noop ("# DNS requests intercepted"),
			    1, GNUNET_NO);
  if (0 == dns_exit_available)
  {
    GNUNET_STATISTICS_update (stats,
			      gettext_noop ("# DNS requests dropped (DNS mesh tunnel down)"),
			      1, GNUNET_NO);
    GNUNET_DNS_request_drop (rh);
    return;
  }
  if (request_length < sizeof (dns))
  {
    GNUNET_STATISTICS_update (stats,
			      gettext_noop ("# DNS requests dropped (malformed)"),
			      1, GNUNET_NO);
    GNUNET_DNS_request_drop (rh);
    return;
  }
  memcpy (&dns, request, sizeof (dns));
  GNUNET_assert (NULL != mesh_tunnel);
  mlen = sizeof (struct GNUNET_MessageHeader) + request_length;
  rc = GNUNET_malloc (sizeof (struct RequestContext) + mlen);
  rc->rh = rh;
  rc->mesh_message = (const struct GNUNET_MessageHeader*) &rc[1];
  rc->timeout_task = GNUNET_SCHEDULER_add_delayed (TIMEOUT,
						   &timeout_request,
						   rc);
  rc->dns_id = dns.id;
  hdr.type = htons (GNUNET_MESSAGE_TYPE_VPN_DNS_TO_INTERNET);
  hdr.size = htons (mlen);
  memcpy (&rc[1], &hdr, sizeof (struct GNUNET_MessageHeader));
  memcpy (&(((char*)&rc[1])[sizeof (struct GNUNET_MessageHeader)]),
	  request,
	  request_length);
  GNUNET_CONTAINER_DLL_insert_tail (transmit_queue_head,
				    transmit_queue_tail,
				    rc);
  if (NULL == mesh_th)
    mesh_th = GNUNET_MESH_notify_transmit_ready (mesh_tunnel,
						 GNUNET_NO, 0,
						 TIMEOUT,
						 NULL, mlen,
						 &transmit_dns_request_to_mesh,
						 NULL);
}

/**
 * Function called to get a message for transmission.
 *
 * @param cls closure
 * @param buf_size number of bytes available in buf
 * @param buf where to copy the message, NULL on error (peer disconnect)
 * @return number of bytes copied to 'buf', can be 0 (without indicating an error)
 */
static size_t
transmit_message_callback (void *cls, size_t buf_size, void *buf)
{
  struct PeerPlan *pp = cls;
  struct GSF_RequestPlan *rp;
  size_t msize;

  pp->pth = NULL;
  if (NULL == buf)
  {
    /* failed, try again... */
    if (GNUNET_SCHEDULER_NO_TASK != pp->task)
      GNUNET_SCHEDULER_cancel (pp->task);

    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
    GNUNET_STATISTICS_update (GSF_stats,
                              gettext_noop
                              ("# transmission failed (core has no bandwidth)"),
                              1, GNUNET_NO);
    return 0;
  }
  rp = GNUNET_CONTAINER_heap_peek (pp->priority_heap);
  if (NULL == rp)
  {
    if (GNUNET_SCHEDULER_NO_TASK != pp->task)
      GNUNET_SCHEDULER_cancel (pp->task);
    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
    return 0;
  }
  msize = GSF_pending_request_get_message_ (get_latest (rp), buf_size, buf);
  if (msize > buf_size)
  {
    if (GNUNET_SCHEDULER_NO_TASK != pp->task)
      GNUNET_SCHEDULER_cancel (pp->task);
    /* buffer to small (message changed), try again */
    pp->task = GNUNET_SCHEDULER_add_now (&schedule_peer_transmission, pp);
    return 0;
  }
  /* remove from root, add again elsewhere... */
  GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->priority_heap));
  rp->hn = NULL;
  rp->last_transmission = GNUNET_TIME_absolute_get ();
  rp->transmission_counter++;
  total_delay++;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Executing plan %p executed %u times, planning retransmission\n",
              rp, rp->transmission_counter);
  plan (pp, rp);
  GNUNET_STATISTICS_update (GSF_stats,
                            gettext_noop
                            ("# query messages sent to other peers"), 1,
                            GNUNET_NO);
  return msize;
}

/**
 * Periodically announce self id in the DHT
 *
 * @param cls closure
 * @param tc task context
 */
static void
announce_id (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct GNUNET_HashCode phash;
  const struct GNUNET_HELLO_Message *hello;
  size_t size;
  struct GNUNET_TIME_Absolute expiration;
  struct GNUNET_TIME_Relative retry_time;

  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
  {
    announce_id_task = NULL;
    return;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Announce ID\n");
  /* TODO
   * - Set data expiration in function of X
   * - Adapt X to churn
   */
  hello = GCH_get_mine ();
  if (NULL == hello || (size = GNUNET_HELLO_size (hello)) == 0)
  {
    /* Peerinfo gave us no hello yet, try again in a second. */
    announce_id_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
                                                     &announce_id, cls);
    LOG (GNUNET_ERROR_TYPE_DEBUG, "  no hello, waiting!\n");
    GNUNET_STATISTICS_update (stats, "# DHT announce skipped (no hello)",
                              1, GNUNET_NO);

    return;
  }
  expiration = GNUNET_HELLO_get_last_expiration (hello);
  retry_time = GNUNET_TIME_absolute_get_remaining (expiration);

  LOG (GNUNET_ERROR_TYPE_DEBUG, "Hello %p size: %u\n", hello, size);
  GNUNET_STATISTICS_update (stats, "# DHT announce",
                            1, GNUNET_NO);
  memset (&phash, 0, sizeof (phash));
  memcpy (&phash, &my_full_id, sizeof (my_full_id));
  GNUNET_DHT_put (dht_handle,   /* DHT handle */
                  &phash,       /* Key to use */
                  dht_replication_level,     /* Replication level */
                  GNUNET_DHT_RO_RECORD_ROUTE
                  | GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,    /* DHT options */
                  GNUNET_BLOCK_TYPE_DHT_HELLO,       /* Block type */
                  size,  /* Size of the data */
                  (const char *) hello, /* Data itself */
                  expiration,  /* Data expiration */
                  retry_time, /* Retry time */
                  NULL,         /* Continuation */
                  NULL);        /* Continuation closure */
  announce_id_task =
      GNUNET_SCHEDULER_add_delayed (id_announce_time, &announce_id, cls);
}

/**
 * Function called by plugins to notify the datacache
 * about content deletions.
 *
 * @param cls closure
 * @param key key of the content that was deleted
 * @param size number of bytes that were made available
 */
static void
env_delete_notify (void *cls, const struct GNUNET_HashCode * key, size_t size)
{
  struct GNUNET_DATACACHE_Handle *h = cls;

  LOG (GNUNET_ERROR_TYPE_DEBUG, "Content under key `%s' discarded\n",
       GNUNET_h2s (key));
  GNUNET_assert (h->utilization >= size);
  h->utilization -= size;
  GNUNET_CONTAINER_bloomfilter_remove (h->filter, key);
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# bytes stored"), - (long long) size,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# items stored"), -1,
                            GNUNET_NO);
}

/**
 * Schedule transmission for the given peer for the current round based
 * on what we know about the desired delay.
 *
 * @param cls unused
 * @param key hash of peer identity
 * @param value the `struct NSEPeerEntry`
 * @return #GNUNET_OK (continue to iterate)
 */
static int
schedule_current_round (void *cls,
			const struct GNUNET_PeerIdentity * key,
			void *value)
{
  struct NSEPeerEntry *peer_entry = value;
  struct GNUNET_TIME_Relative delay;

  if (NULL != peer_entry->transmit_task)
  {
    GNUNET_SCHEDULER_cancel (peer_entry->transmit_task);
    peer_entry->previous_round = GNUNET_NO;
  }
#if ENABLE_NSE_HISTOGRAM
  if (peer_entry->received_messages > 1)
    GNUNET_STATISTICS_update(stats,
			     "# extra messages",
                             peer_entry->received_messages - 1,
			     GNUNET_NO);
  peer_entry->transmitted_messages = 0;
  peer_entry->last_transmitted_size = 0;
  peer_entry->received_messages = 0;
#endif
  delay =
      get_transmit_delay ((GNUNET_NO == peer_entry->previous_round) ? -1 : 0);
  peer_entry->transmit_task =
      GNUNET_SCHEDULER_add_delayed (delay,
				    &transmit_task_cb,
				    peer_entry);
  return GNUNET_OK;
}

/**
 * Method called whenever a peer disconnects. Deletes the PeerEntry and cancels
 * any pending transmission requests to that peer.
 *
 * @param cls closure
 * @param peer peer identity this notification is about
 * @parma internal_cls the `struct NSEPeerEntry` for the @a peer
 */
static void
handle_core_disconnect (void *cls,
			const struct GNUNET_PeerIdentity *peer,
			void *internal_cls)
{
  struct NSEPeerEntry *pos = internal_cls;

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Peer `%s' disconnected from us\n",
              GNUNET_i2s (peer));
  GNUNET_assert (GNUNET_YES ==
                 GNUNET_CONTAINER_multipeermap_remove (peers,
						       peer,
                                                       pos));
  if (NULL != pos->transmit_task)
  {
    GNUNET_SCHEDULER_cancel (pos->transmit_task);
    pos->transmit_task = NULL;
  }
  GNUNET_free (pos);
  GNUNET_STATISTICS_update (stats,
			    "# peers connected",
			    -1,
			    GNUNET_NO);
}

/**
 * Search DHT for paths to @a peeR_id
 *
 * @param peer_id peer to search for
 * @return handle to abort search
 */
struct GCD_search_handle *
GCD_search (const struct GNUNET_PeerIdentity *peer_id)
{
  struct GNUNET_HashCode phash;
  struct GCD_search_handle *h;

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Starting DHT GET for peer %s\n",
       GNUNET_i2s (peer_id));
  GNUNET_STATISTICS_update (stats,
                            "# DHT search",
                            1,
                            GNUNET_NO);
  memset (&phash,
          0,
          sizeof (phash));
  GNUNET_memcpy (&phash,
                 peer_id,
                 sizeof (*peer_id));

  h = GNUNET_new (struct GCD_search_handle);
  h->dhtget = GNUNET_DHT_get_start (dht_handle,    /* handle */
                                    GNUNET_BLOCK_TYPE_DHT_HELLO, /* type */
                                    &phash,     /* key to search */
                                    dht_replication_level, /* replication level */
                                    GNUNET_DHT_RO_RECORD_ROUTE |
                                    GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                    NULL,       /* xquery */
                                    0,     /* xquery bits */
                                    &dht_get_id_handler,
				    h);
  return h;
}

/**
 * Handle request connect message
 *
 * @param cls closure (always NULL)
 * @param client identification of the client
 * @param message the actual message
 */
static void
clients_handle_request_connect (void *cls, struct GNUNET_SERVER_Client *client,
                                const struct GNUNET_MessageHeader *message)
{
  const struct TransportRequestConnectMessage *trcm =
      (const struct TransportRequestConnectMessage *) message;

  GNUNET_STATISTICS_update (GST_stats,
                            gettext_noop
                            ("# REQUEST CONNECT messages received"), 1,
                            GNUNET_NO);

  if (0 == memcmp (&trcm->peer, &GST_my_identity,
  		sizeof (struct GNUNET_PeerIdentity)))
  {
    GNUNET_break_op (0);
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Received a request connect message myself `%s'\n",
                GNUNET_i2s (&trcm->peer));
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Received a request connect message for peer `%s'\n",
                GNUNET_i2s (&trcm->peer));

    (void) GST_blacklist_test_allowed (&trcm->peer, NULL, &try_connect_if_allowed,
                                     NULL);
  }
  GNUNET_SERVER_receive_done (client, GNUNET_OK);
}

struct GCD_search_handle *
GCD_search (const struct GNUNET_PeerIdentity *peer_id,
            GCD_search_callback callback, void *cls)
{
  struct GNUNET_HashCode phash;
  struct GCD_search_handle *h;

  LOG (GNUNET_ERROR_TYPE_DEBUG, "Starting DHT GET for peer %s\n",
       GNUNET_i2s (peer_id));
  GNUNET_STATISTICS_update (stats, "# DHT search", 1, GNUNET_NO);
  memset (&phash, 0, sizeof (phash));
  GNUNET_memcpy (&phash, peer_id, sizeof (*peer_id));
  h = GNUNET_new (struct GCD_search_handle);
  h->peer_id = GNUNET_PEER_intern (peer_id);
  h->callback = callback;
  h->cls = cls;
  h->dhtget = GNUNET_DHT_get_start (dht_handle,    /* handle */
                                    GNUNET_BLOCK_TYPE_DHT_HELLO, /* type */
                                    &phash,     /* key to search */
                                    dht_replication_level, /* replication level */
                                    GNUNET_DHT_RO_RECORD_ROUTE |
                                    GNUNET_DHT_RO_DEMULTIPLEX_EVERYWHERE,
                                    NULL,       /* xquery */
                                    0,     /* xquery bits */
                                    &dht_get_id_handler,
				    h);
  GNUNET_CONTAINER_multihashmap32_put (get_requests,
				       h->peer_id,
				       h,
                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
  return h;
}

/**
 * Add a host to the list and notify clients about this event
 *
 * @param identity the identity of the host
 * @return the HostEntry
 */
static struct HostEntry *
add_host_to_known_hosts (const struct GNUNET_PeerIdentity *identity)
{
  struct HostEntry *entry;
  struct ReadHostFileContext r;
  char *fn;

  entry = GNUNET_CONTAINER_multipeermap_get (hostmap, identity);
  if (NULL == entry)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Adding new peer `%s'\n", GNUNET_i2s (identity));
    GNUNET_STATISTICS_update (stats, gettext_noop ("# peers known"), 1,
			      GNUNET_NO);
    entry = GNUNET_new (struct HostEntry);
    entry->identity = *identity;
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multipeermap_put (hostmap, &entry->identity, entry,
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
    notify_all (entry);
    fn = get_host_filename (identity);
    if (NULL != fn)
    {
      read_host_file (fn, GNUNET_YES, &r);
      if (NULL != r.hello)
      	update_hello (identity, r.hello);
      if (NULL != r.friend_only_hello)
      	update_hello (identity, r.friend_only_hello);
      GNUNET_free_non_null (r.hello);
      GNUNET_free_non_null (r.friend_only_hello);
      GNUNET_free (fn);
    }
  }

/**
 * Callback invoked from the VPN service once a redirection is
 * available.  Provides the IP address that can now be used to
 * reach the requested destination.  We substitute the active
 * record and then continue with 'submit_request' to look at
 * the other records.
 *
 * @param cls our 'struct ReplyContext'
 * @param af address family, AF_INET or AF_INET6; AF_UNSPEC on error;
 *                will match 'result_af' from the request
 * @param address IP address (struct in_addr or struct in_addr6, depending on 'af')
 *                that the VPN allocated for the redirection;
 *                traffic to this IP will now be redirected to the 
 *                specified target peer; NULL on error
 */
static void
vpn_allocation_callback (void *cls,
			 int af,
			 const void *address)
{
  struct ReplyContext *rc = cls;

  rc->rr = NULL;
  if (af == AF_UNSPEC)
  {
    GNUNET_DNS_request_drop (rc->rh);
    GNUNET_DNSPARSER_free_packet (rc->dns);
    GNUNET_free (rc);
    return;
  }
  GNUNET_STATISTICS_update (stats,
			    gettext_noop ("# DNS records modified"),
			    1, GNUNET_NO);
  switch (rc->rec->type)
  {
  case GNUNET_DNSPARSER_TYPE_A:
    GNUNET_assert (AF_INET == af);
    memcpy (rc->rec->data.raw.data, address, sizeof (struct in_addr));
    break;
  case GNUNET_DNSPARSER_TYPE_AAAA:
    GNUNET_assert (AF_INET6 == af);
    memcpy (rc->rec->data.raw.data, address, sizeof (struct in6_addr));
    break;
  default:
    GNUNET_assert (0);
    return;
  }
  rc->rec = NULL;
  submit_request (rc);
}

/**
 * We're done modifying all records in the response.  Submit the reply
 * and free the resources of the rc.
 *
 * @param rc context to process
 */
static void
finish_request (struct ReplyContext *rc)
{
  char *buf;
  size_t buf_len;

  if (GNUNET_SYSERR ==
      GNUNET_DNSPARSER_pack (rc->dns,
			     MAX_DNS_SIZE,
			     &buf,
			     &buf_len))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
		_("Failed to pack DNS request.  Dropping.\n"));
    GNUNET_DNS_request_drop (rc->rh);
  }
  else
  {
    GNUNET_STATISTICS_update (stats,
			      gettext_noop ("# DNS requests mapped to VPN"),
			      1, GNUNET_NO);
    GNUNET_DNS_request_answer (rc->rh,
			       buf_len, buf);
    GNUNET_free (buf);
  }
  GNUNET_DNSPARSER_free_packet (rc->dns);
  GNUNET_free (rc);
}

/**
 * Read the friends file.
 */
static void
read_friends_file (const struct GNUNET_CONFIGURATION_Handle *cfg)
{
  unsigned int entries_found;

  entries_found = 0;
  if (GNUNET_OK !=
      GNUNET_FRIENDS_parse (cfg,
                            &handle_friend,
                            &entries_found))
  {
    if ( (GNUNET_YES == friends_only) ||
         (minimum_friend_count > 0))
      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                  _("Encountered errors parsing friends list!\n"));
  }
  GNUNET_STATISTICS_update (stats,
                            gettext_noop ("# friends in configuration"),
                            entries_found,
                            GNUNET_NO);
  if ( (minimum_friend_count > entries_found) &&
       (GNUNET_NO == friends_only) )
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _("Fewer friends specified than required by minimum friend count. Will only connect to friends.\n"));
  }
  if ( (minimum_friend_count > target_connection_count) &&
       (GNUNET_NO == friends_only))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _("More friendly connections required than target total number of connections.\n"));
  }
}

/**
 * An iterator over a set of items stored in the datastore
 * that deletes until we're happy with respect to our quota.
 *
 * @param cls closure
 * @param key key for the content
 * @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 expiration expiration time for the content
 * @param uid unique identifier for the datum;
 *        maybe 0 if no unique identifier is available
 *
 * @return GNUNET_SYSERR to abort the iteration, GNUNET_OK to continue
 *         (continue on call to "next", of course),
 *         GNUNET_NO to delete the item and continue (if supported)
 */
static int
quota_processor (void *cls, const struct GNUNET_HashCode * key, uint32_t size,
                 const void *data, enum GNUNET_BLOCK_Type type,
                 uint32_t priority, uint32_t anonymity,
                 struct GNUNET_TIME_Absolute expiration, uint64_t uid)
{
  unsigned long long *need = cls;

  if (NULL == key)
    return GNUNET_SYSERR;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Deleting %llu bytes of low-priority (%u) content `%s' of type %u at %s prior to expiration (still trying to free another %llu bytes)\n",
              (unsigned long long) (size + GNUNET_DATASTORE_ENTRY_OVERHEAD),
	      (unsigned int) priority,
              GNUNET_h2s (key), type,
	      GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (expiration),
						      GNUNET_YES),
	      *need);
  if (size + GNUNET_DATASTORE_ENTRY_OVERHEAD > *need)
    *need = 0;
  else
    *need -= size + GNUNET_DATASTORE_ENTRY_OVERHEAD;
  if (priority > 0)
    min_expiration = GNUNET_TIME_UNIT_FOREVER_ABS;
  else
    min_expiration = expiration;
  GNUNET_STATISTICS_update (stats,
                            gettext_noop ("# bytes purged (low-priority)"),
                            size, GNUNET_YES);
  GNUNET_CONTAINER_bloomfilter_remove (filter, key);
  return GNUNET_NO;
}

/**
 * Construct our HELLO message from all of the addresses of
 * all of the transports.
 *
 * @param cls unused
 * @param tc scheduler context
 */
static void
refresh_hello_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct GeneratorContext gc;
  int friend_only;

  hello_task = GNUNET_SCHEDULER_NO_TASK;
  gc.addr_pos = oal_head;
  gc.expiration = GNUNET_TIME_relative_to_absolute (hello_expiration);


  friend_only = GNUNET_HELLO_is_friend_only (our_hello);
  GNUNET_free (our_hello);
  our_hello = GNUNET_HELLO_create (&GST_my_public_key, &address_generator, &gc, friend_only);
  GNUNET_assert (NULL != our_hello);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Refreshed my %s `%s', new size is %d\n",
              (GNUNET_YES == GNUNET_HELLO_is_friend_only (our_hello)) ? "friend-only" : "public",
              "HELLO", GNUNET_HELLO_size (our_hello));
  GNUNET_STATISTICS_update (GST_stats, gettext_noop ("# refreshed my HELLO"), 1,
                            GNUNET_NO);
  if (NULL != hello_cb)
    hello_cb (hello_cb_cls, GST_hello_get ());
  GNUNET_PEERINFO_add_peer (GST_peerinfo, our_hello, NULL, NULL);
  hello_task =
      GNUNET_SCHEDULER_add_delayed (HELLO_REFRESH_PERIOD, &refresh_hello_task,
                                    NULL);

}

/**
 * Handle a datum we've received from another peer.  Cache if
 * possible.
 *
 * @param expiration when will the reply expire
 * @param key the query this reply is for
 * @param put_path_length number of peers in 'put_path'
 * @param put_path path the reply took on put
 * @param type type of the reply
 * @param data_size number of bytes in 'data'
 * @param data application payload data
 */
void
GDS_DATACACHE_handle_put (struct GNUNET_TIME_Absolute expiration,
                          const struct GNUNET_HashCode * key,
                          unsigned int put_path_length,
                          const struct GNUNET_PeerIdentity *put_path,
                          enum GNUNET_BLOCK_Type type, size_t data_size,
                          const void *data)
{
  if (NULL == datacache)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                _("%s request received, but have no datacache!\n"), "PUT");
    return;
  }
  if (data_size >= GNUNET_SERVER_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    return;
  }
  /* Put size is actual data size plus struct overhead plus path length (if any) */
  GNUNET_STATISTICS_update (GDS_stats,
                            gettext_noop ("# ITEMS stored in datacache"), 1,
                            GNUNET_NO);
  (void) GNUNET_DATACACHE_put (datacache, key, 
			       data_size, data, type,
                               expiration, put_path_length, put_path);
}

/**
 * Transmit our current typemap message to the other peer.
 * (Done periodically until the typemap is confirmed).
 *
 * @param cls the `struct Session *`
 * @param tc unused
 */
static void
transmit_typemap_task (void *cls,
                       const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct Session *session = cls;
  struct GNUNET_MessageHeader *hdr;
  struct GNUNET_TIME_Relative delay;

  session->typemap_delay = GNUNET_TIME_STD_BACKOFF (session->typemap_delay);
  delay = session->typemap_delay;
  /* randomize a bit to avoid spont. sync */
  delay.rel_value_us +=
      GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK, 1000 * 1000);
  session->typemap_task =
      GNUNET_SCHEDULER_add_delayed (delay,
                                    &transmit_typemap_task, session);
  GNUNET_STATISTICS_update (GSC_stats,
                            gettext_noop ("# type map refreshes sent"),
                            1,
                            GNUNET_NO);
  hdr = GSC_TYPEMAP_compute_type_map_message ();
  GSC_KX_encrypt_and_transmit (session->kxinfo,
                               hdr,
                               ntohs (hdr->size));
  GNUNET_free (hdr);
}

/**
 * Sending a reply was completed, continue processing.
 *
 * @param cls closure with the struct StreamClient which sent the query
 * @param status result code for the operation
 * @param size number of bytes that were transmitted
 */
static void
write_continuation (void *cls,
		    enum GNUNET_STREAM_Status status,
		    size_t size)
{
  struct StreamClient *sc = cls;
  
  sc->wh = NULL;
  if ( (GNUNET_STREAM_OK == status) &&
       (size == sc->reply_size) )
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
		"Transmitted %u byte reply via stream\n",
		(unsigned int) size);
    GNUNET_STATISTICS_update (GSF_stats,
			      gettext_noop ("# Blocks transferred via stream"), 1,
			      GNUNET_NO);
    continue_reading (sc);
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
		"Transmission of reply failed, terminating stream\n");
    terminate_stream (sc);    
  }
}