/**
 * Compose #PeerIterateResponseMessage using the given peer and address.
 *
 * @param peer identity of the peer
 * @param address the address, NULL on disconnect
 * @return composed message
 */
static struct ValidationIterateResponseMessage *
compose_validation_iterate_response_message (const struct GNUNET_PeerIdentity *peer,
                                          const struct GNUNET_HELLO_Address *address)
{
  struct ValidationIterateResponseMessage *msg;
  size_t size;
  size_t tlen;
  size_t alen;
  char *addr;

  GNUNET_assert (NULL != peer);
  if (NULL != address)
  {
    tlen = strlen (address->transport_name) + 1;
    alen = address->address_length;
  }
  else
    tlen = alen = 0;
  size = (sizeof (struct ValidationIterateResponseMessage) + alen + tlen);
  msg = GNUNET_malloc (size);
  msg->header.size = htons (size);
  msg->header.type =
      htons (GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_VALIDATION_RESPONSE);
  msg->reserved = htonl (0);
  msg->peer = *peer;
  msg->addrlen = htonl (alen);
  msg->pluginlen = htonl (tlen);

  if (NULL != address)
  {
    msg->local_address_info = htonl((uint32_t) address->local_info);
    addr = (char *) &msg[1];
    memcpy (addr, address->address, alen);
    memcpy (&addr[alen], address->transport_name, tlen);
  }
  return msg;
}

static void
publish_cont (void *cls, const struct GNUNET_FS_Uri *ksk_uri, const char *emsg)
{
  char *msg;
  struct GNUNET_FS_Uri *sks_uri;
  char sbuf[1024];
  char buf[1024];
  char *ret;

  if (NULL != emsg)
  {
    FPRINTF (stderr, "Error publishing: %s\n", emsg);
    err = 1;
    GNUNET_FS_stop (fs);
    return;
  }
  ret = GNUNET_STRINGS_data_to_string (&nsid, sizeof (nsid), buf, sizeof (buf));
  GNUNET_assert (NULL != ret);
  ret[0] = '\0';
  GNUNET_snprintf (sbuf, sizeof (sbuf), "gnunet://fs/sks/%s/this", buf);
  sks_uri = GNUNET_FS_uri_parse (sbuf, &msg);
  if (NULL == sks_uri)
  {
    FPRINTF (stderr, "failed to parse URI `%s': %s\n", sbuf, msg);
    err = 1;
    GNUNET_FS_stop (fs);
    GNUNET_free_non_null (msg);
    return;
  }
  ksk_search =
      GNUNET_FS_search_start (fs, ksk_uri, 1, GNUNET_FS_SEARCH_OPTION_NONE,
                              "ksk_search");
  sks_search =
      GNUNET_FS_search_start (fs, sks_uri, 1, GNUNET_FS_SEARCH_OPTION_NONE,
                              "sks_search");
  GNUNET_FS_uri_destroy (sks_uri);
}

static size_t
notify_ready (void *cls, size_t size, void *buf)
{
  struct PeerContext *p = cls;
  struct GNUNET_MessageHeader *hdr;

  th = NULL;

  if (buf == NULL)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Timeout occurred while waiting for transmit_ready\n");
    if (GNUNET_SCHEDULER_NO_TASK != die_task)
      GNUNET_SCHEDULER_cancel (die_task);
    die_task = GNUNET_SCHEDULER_add_now (&end_badly, NULL);
    ok = 42;
    return 0;
  }

  GNUNET_assert (size >= TEST_MESSAGE_SIZE);
  if (buf != NULL)
  {
    memset (buf, '\0', TEST_MESSAGE_SIZE);
    hdr = buf;
    hdr->size = htons (TEST_MESSAGE_SIZE);
    hdr->type = htons (TEST_MESSAGE_TYPE);
  }

  char *ps = GNUNET_strdup (GNUNET_i2s (&p2->id));
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "Peer %u (`%4s') sending message with type %u and size %u bytes to peer %u (`%4s')\n",
              p2->no, ps, ntohs (hdr->type), ntohs (hdr->size), p->no,
              GNUNET_i2s (&p->id));
  GNUNET_free (ps);

  return TEST_MESSAGE_SIZE;
}

/**
 * Send a message to all of our current clients that have the right
 * options set.
 *
 * @param partner origin (or destination) of the message (used to check that this peer is
 *        known to be connected to the respective client)
 * @param msg message to multicast
 * @param can_drop can this message be discarded if the queue is too long
 * @param options mask to use
 * @param type type of the embedded message, 0 for none
 */
static void
send_to_all_clients (const struct GNUNET_PeerIdentity *partner,
                     const struct GNUNET_MessageHeader *msg,
                     int can_drop,
                     uint32_t options,
                     uint16_t type)
{
  struct GSC_Client *c;
  int tm;

  for (c = client_head; NULL != c; c = c->next)
  {
    tm = type_match (type, c);
    if (!  ( (0 != (c->options & options)) ||
	     ( (0 != (options & GNUNET_CORE_OPTION_SEND_FULL_INBOUND)) &&
	       (GNUNET_YES == tm) ) ) )
      continue;  /* neither options nor type match permit the message */
    if ( (0 != (options & GNUNET_CORE_OPTION_SEND_HDR_INBOUND)) &&
	 ( (0 != (c->options & GNUNET_CORE_OPTION_SEND_FULL_INBOUND)) ||
	   (GNUNET_YES == tm) ) )
      continue;
    if ( (0 != (options & GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND)) &&
	 (0 != (c->options & GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND)) )
      continue;
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sending %u message with %u bytes to client interested in messages of type %u.\n",
		options,
		ntohs (msg->size),
                (unsigned int) type);
    GNUNET_assert ( (0 == (c->options & GNUNET_CORE_OPTION_SEND_FULL_INBOUND)) ||
		    (GNUNET_YES != tm) ||
		    (GNUNET_YES ==
		     GNUNET_CONTAINER_multipeermap_contains (c->connectmap,
							     partner)) );
    send_to_client (c, msg, can_drop);
  }
}

static void
do_shutdown ()
{
  if (mhd_task_id != GNUNET_SCHEDULER_NO_TASK)
  {
    GNUNET_SCHEDULER_cancel (mhd_task_id);
    mhd_task_id = GNUNET_SCHEDULER_NO_TASK;
  }
  if (curl_task_id != GNUNET_SCHEDULER_NO_TASK)
  {
    GNUNET_SCHEDULER_cancel (curl_task_id);
    curl_task_id = GNUNET_SCHEDULER_NO_TASK;
  }
  if (NULL != mhd)
  {
    MHD_stop_daemon (mhd);
    mhd = NULL;
  }
  GNUNET_free_non_null (url);

  if (NULL != tmp_cfgfile)
    {
      if (0 != remove (tmp_cfgfile))
	GNUNET_log_strerror_file (GNUNET_ERROR_TYPE_WARNING, "remove", tmp_cfgfile);
      GNUNET_free (tmp_cfgfile);
      tmp_cfgfile = NULL;
    }
  if (NULL != proxy_proc)
    {
      (void) GNUNET_OS_process_kill (proxy_proc, SIGKILL);
      GNUNET_assert (GNUNET_OK == GNUNET_OS_process_wait (proxy_proc));
      GNUNET_OS_process_destroy (proxy_proc);
      proxy_proc = NULL;
    }
  url = NULL;
  GNUNET_SCHEDULER_shutdown ();
}

int
main (int argc, char *argv_ign[])
{
  int ok = 1;

  char *const argv[] = { "test-statistics-api",
    "-c",
    "test_statistics_api_data.conf",
    NULL
  };
  struct GNUNET_GETOPT_CommandLineOption options[] = {
    GNUNET_GETOPT_OPTION_END
  };
  struct GNUNET_OS_Process *proc;
  char *binary;

  binary = GNUNET_OS_get_libexec_binary_path ("gnunet-service-statistics");
  proc =
    GNUNET_OS_start_process (GNUNET_YES, GNUNET_OS_INHERIT_STD_OUT_AND_ERR,
			     NULL, NULL, NULL,
			     binary,
			     "gnunet-service-statistics",
			     "-c", "test_statistics_api_data.conf", NULL);
  GNUNET_assert (NULL != proc);
  GNUNET_PROGRAM_run (3, argv, "test-statistics-api", "nohelp", options, &run,
                      &ok);
  if (0 != GNUNET_OS_process_kill (proc, GNUNET_TERM_SIG))
  {
    GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "kill");
    ok = 1;
  }
  GNUNET_OS_process_wait (proc);
  GNUNET_OS_process_destroy (proc);
  proc = NULL;
  GNUNET_free (binary);
  return ok;
}

/**
 * Store an item in the datastore.
 *
 * @param h handle to the datacache
 * @param key key to store data under
 * @param size number of bytes in data
 * @param data data to store
 * @param type type of the value
 * @param discard_time when to discard the value in any case
 * @param path_info_len number of entries in 'path_info'
 * @param path_info a path through the network
 * @return GNUNET_OK on success, GNUNET_SYSERR on error, GNUNET_NO if duplicate
 */
int
GNUNET_DATACACHE_put (struct GNUNET_DATACACHE_Handle *h,
                      const struct GNUNET_HashCode * key, size_t size,
                      const char *data, enum GNUNET_BLOCK_Type type,
                      struct GNUNET_TIME_Absolute discard_time,
		      unsigned int path_info_len,
		      const struct GNUNET_PeerIdentity *path_info)
{
  ssize_t used;

  used = h->api->put (h->api->cls, key, 
		      size, data, 
		      type, discard_time,
		      path_info_len, path_info);
  if (-1 == used)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  if (0 == used)
  {
    /* duplicate */
    return GNUNET_NO;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Stored data under key `%s' in cache\n",
       GNUNET_h2s (key));
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# bytes stored"), size,
                            GNUNET_NO);
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# items stored"), 1,
                            GNUNET_NO);
  if (NULL != h->filter)
    GNUNET_CONTAINER_bloomfilter_add (h->filter, key);
  while (h->utilization + used > h->env.quota)
    GNUNET_assert (GNUNET_OK == h->api->del (h->api->cls));
  h->utilization += used;
  return GNUNET_OK;
}

/**
 * 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);
    }
  }

/**
 * Handler for messages received from the GNS service
 *
 * @param cls the `struct GNUNET_GNS_Handle *`
 * @param loookup_msg the incoming message
 */
static void
handle_result (void *cls,
               const struct LookupResultMessage *lookup_msg)
{
    struct GNUNET_GNS_Handle *handle = cls;
    size_t mlen = ntohs (lookup_msg->header.size) - sizeof (*lookup_msg);
    uint32_t rd_count = ntohl (lookup_msg->rd_count);
    struct GNUNET_GNSRECORD_Data rd[rd_count];
    uint32_t r_id = ntohl (lookup_msg->id);
    struct GNUNET_GNS_LookupRequest *lr;
    GNUNET_GNS_LookupResultProcessor proc;
    void *proc_cls;

    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Received lookup reply from GNS service (%u records)\n",
         (unsigned int) rd_count);
    for (lr = handle->lookup_head; NULL != lr; lr = lr->next)
        if (lr->r_id == r_id)
            break;
    if (NULL == lr)
        return;
    proc = lr->lookup_proc;
    proc_cls = lr->proc_cls;
    GNUNET_CONTAINER_DLL_remove (handle->lookup_head,
                                 handle->lookup_tail,
                                 lr);
    GNUNET_free (lr);
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_GNSRECORD_records_deserialize (mlen,
                           (const char*) &lookup_msg[1],
                           rd_count,
                           rd));
    proc (proc_cls,
          rd_count,
          rd);
}

/**
 * Initialize framework and start test
 *
 * @param cls closure
 * @param cfg configuration of the peer that was started
 * @param peer identity of the peer that was created
 */
static void
run (void *cls, 
     const struct GNUNET_CONFIGURATION_Handle *cfg,
     struct GNUNET_TESTING_Peer *peer)
{
  struct GNUNET_PeerIdentity id;

  GNUNET_TESTING_peer_get_identity (peer, &id);
  config = cfg;
  peer2_listen_socket = 
      GNUNET_STREAM_listen (config, 10, &stream_listen_cb, &peer_data[1],
			    GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
                            &stream_connect,
                            GNUNET_STREAM_OPTION_MAX_PAYLOAD_SIZE,
                            payload_size[payload_size_index],
                            GNUNET_STREAM_OPTION_END);
  GNUNET_assert (NULL != peer2_listen_socket);
  peer_data[0].id = id;
  peer_data[1].id = id;
  abort_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                    (GNUNET_TIME_UNIT_SECONDS, 300), &do_abort,
                                    NULL);
}

/**
 * Get a random value from the datastore for content replication.
 * Returns a single, random value among those with the highest
 * replication score, lowering positive replication scores by one for
 * the chosen value (if only content with a replication score exists,
 * a random value is returned and replication scores are not changed).
 *
 * @param h handle to the datastore
 * @param queue_priority ranking of this request in the priority queue
 * @param max_queue_size at what queue size should this request be dropped
 *        (if other requests of higher priority are in the queue)
 * @param timeout how long to wait at most for a response
 * @param proc function to call on a random value; it
 *        will be called once with a value (if available)
 *        and always once with a value of NULL.
 * @param proc_cls closure for proc
 * @return NULL if the entry was not queued, otherwise a handle that can be used to
 *         cancel
 */
struct GNUNET_DATASTORE_QueueEntry *
GNUNET_DATASTORE_get_for_replication (struct GNUNET_DATASTORE_Handle *h,
                                      unsigned int queue_priority,
                                      unsigned int max_queue_size,
                                      struct GNUNET_TIME_Relative timeout,
                                      GNUNET_DATASTORE_DatumProcessor proc,
                                      void *proc_cls)
{
  struct GNUNET_DATASTORE_QueueEntry *qe;
  struct GNUNET_MessageHeader *m;
  union QueueContext qc;

  GNUNET_assert (NULL != proc);
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Asked to get replication entry in %llu ms\n",
       (unsigned long long) timeout.rel_value);
  qc.rc.proc = proc;
  qc.rc.proc_cls = proc_cls;
  qe = make_queue_entry (h, sizeof (struct GNUNET_MessageHeader),
                         queue_priority, max_queue_size, timeout,
                         &process_result_message, &qc);
  if (qe == NULL)
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Could not create queue entry for GET REPLICATION\n");
    return NULL;
  }
  GNUNET_STATISTICS_update (h->stats,
                            gettext_noop
                            ("# GET REPLICATION requests executed"), 1,
                            GNUNET_NO);
  m = (struct GNUNET_MessageHeader *) &qe[1];
  m->type = htons (GNUNET_MESSAGE_TYPE_DATASTORE_GET_REPLICATION);
  m->size = htons (sizeof (struct GNUNET_MessageHeader));
  process_queue (h);
  return qe;
}

/**
 * Write the pseudonym infomation into a file
 * @param cfg configuration to use
 * @param nsid hash code of a pseudonym
 * @param meta meta data to be written into a file
 * @param ranking ranking of a pseudonym
 * @param ns_name non-unique name of a pseudonym
 */
static void
write_pseudonym_info (const struct GNUNET_CONFIGURATION_Handle *cfg,
                      const struct GNUNET_HashCode * nsid,
                      const struct GNUNET_CONTAINER_MetaData *meta,
                      int32_t ranking, const char *ns_name)
{
  char *fn;
  struct GNUNET_BIO_WriteHandle *fileW;

  fn = get_data_filename (cfg, PS_METADATA_DIR, nsid);
  GNUNET_assert (fn != NULL);
  fileW = GNUNET_BIO_write_open (fn);
  if (NULL != fileW)
  {
    if ((GNUNET_OK != GNUNET_BIO_write_int32 (fileW, ranking)) ||
        (GNUNET_OK != GNUNET_BIO_write_string (fileW, ns_name)) ||
        (GNUNET_OK != GNUNET_BIO_write_meta_data (fileW, meta)))
    {
      (void) GNUNET_BIO_write_close (fileW);
      GNUNET_break (GNUNET_OK == GNUNET_DISK_directory_remove (fn));
      GNUNET_free (fn);
      return;
    }
    if (GNUNET_OK != GNUNET_BIO_write_close (fileW))
    {
      GNUNET_break (GNUNET_OK == GNUNET_DISK_directory_remove (fn));
      GNUNET_free (fn);
      return;
    }
  }
  GNUNET_free (fn);
  /* create entry for pseudonym name in names */
  if (ns_name != NULL)
    GNUNET_free_non_null (GNUNET_PSEUDONYM_name_uniquify (cfg, nsid, ns_name,
        NULL));
}

/**
 * Add a host to the persistent list.  This method operates in
 * semi-reliable mode: if the transmission is not completed by
 * the time 'GNUNET_PEERINFO_disconnect' is called, it will be
 * aborted.  Furthermore, if a second HELLO is added for the
 * same peer before the first one was transmitted, PEERINFO may
 * merge the two HELLOs prior to transmission to the service.
 *
 * @param h handle to the peerinfo service
 * @param hello the verified (!) HELLO message
 * @param cont continuation to call when done, NULL is allowed
 * @param cont_cls closure for 'cont'
 * @return handle to cancel add operation; all pending
 *         'add' operations will be cancelled automatically
 *        on disconnect, so it is not necessary to keep this
 *        handle (unless 'cont' is NULL and at some point
 *        calling 'cont' must be prevented)
 */
struct GNUNET_PEERINFO_AddContext *
GNUNET_PEERINFO_add_peer (struct GNUNET_PEERINFO_Handle *h,
                          const struct GNUNET_HELLO_Message *hello,
			  GNUNET_PEERINFO_Continuation cont,
			  void *cont_cls)
{
  uint16_t hs = GNUNET_HELLO_size (hello);
  struct GNUNET_PEERINFO_AddContext *ac;
  struct GNUNET_PeerIdentity peer;

  GNUNET_assert (GNUNET_OK == GNUNET_HELLO_get_id (hello, &peer));
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Adding peer `%s' to PEERINFO database (%u bytes of `%s')\n",
       GNUNET_i2s (&peer), hs, "HELLO");
  ac = GNUNET_malloc (sizeof (struct GNUNET_PEERINFO_AddContext) + hs);
  ac->h = h;
  ac->size = hs;
  ac->cont = cont;
  ac->cont_cls = cont_cls;
  memcpy (&ac[1], hello, hs);
  GNUNET_CONTAINER_DLL_insert_tail (h->ac_head, h->ac_tail, ac);
  trigger_transmit (h);
  return ac;
}

/**
 * Get (or create) a stream to talk to the given peer.
 *
 * @param target peer we want to communicate with
 */
static struct StreamHandle *
get_stream (const struct GNUNET_PeerIdentity *target)
{
  struct StreamHandle *sh;

  sh = GNUNET_CONTAINER_multihashmap_get (stream_map,
					  &target->hashPubKey);
  if (NULL != sh)
  {
    if (GNUNET_SCHEDULER_NO_TASK != sh->timeout_task)
    {
      GNUNET_SCHEDULER_cancel (sh->timeout_task);
      sh->timeout_task = GNUNET_SCHEDULER_NO_TASK;
    }
    return sh;
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Creating stream to %s\n",
	      GNUNET_i2s (target));
  sh = GNUNET_malloc (sizeof (struct StreamHandle));
  sh->mst = GNUNET_SERVER_mst_create (&reply_cb,
				      sh);
  sh->waiting_map = GNUNET_CONTAINER_multihashmap_create (512, GNUNET_YES);
  sh->target = *target;
  sh->stream = GNUNET_STREAM_open (GSF_cfg,
				   &sh->target,
				   GNUNET_APPLICATION_TYPE_FS_BLOCK_TRANSFER,
				   &stream_ready_cb, sh,
				   GNUNET_STREAM_OPTION_END);
  GNUNET_assert (GNUNET_OK ==
		 GNUNET_CONTAINER_multihashmap_put (stream_map,
						    &sh->target.hashPubKey,
						    sh,
						    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  return sh;
}

/**
 * Update the priority for a particular key in the datastore.  If
 * the expiration time in value is different than the time found in
 * the datastore, the higher value should be kept.  For the
 * anonymity level, the lower value is to be used.  The specified
 * priority should be added to the existing priority, ignoring the
 * priority in value.
 *
 * @param cls our "struct Plugin*"
 * @param uid unique identifier of the datum
 * @param delta by how much should the priority
 *     change?  If priority + delta < 0 the
 *     priority should be set to 0 (never go
 *     negative).
 * @param expire new expiration time should be the
 *     MAX of any existing expiration time and
 *     this value
 * @param msg set to error message
 * @return GNUNET_OK on success
 */
static int
heap_plugin_update (void *cls, 
		    uint64_t uid, 
		    int delta,
		    struct GNUNET_TIME_Absolute expire, char **msg)
{
  struct Plugin *plugin = cls;
  struct Value *value;

  value = (struct Value*) (long) uid;
  GNUNET_assert (NULL != value);
  if (value->expiration.abs_value != expire.abs_value)
  {
    value->expiration = expire;
    GNUNET_CONTAINER_heap_update_cost (plugin->by_expiration,
				       value->expire_heap,
				       expire.abs_value);
  }
  if ( (delta < 0) && (value->priority < - delta) )
    value->priority = 0;
  else
    value->priority += delta;
  return GNUNET_OK;
}

/**
 * Copy "frequent" metadata items over to the
 * target metadata container, free the counters.
 *
 * @param cls the 'struct TrimContext'
 * @param key key of the entry
 * @param value the 'struct KeywordCounter'
 * @return GNUNET_YES (always)
 */
static int
migrate_and_drop_metadata (void *cls, const GNUNET_HashCode * key, void *value)
{
  struct TrimContext *tc = cls;
  struct MetaCounter *counter = value;

  if (counter->count >= tc->move_threshold)
  {
    if (NULL == tc->pos->meta)
      tc->pos->meta = GNUNET_CONTAINER_meta_data_create ();
    GNUNET_CONTAINER_meta_data_insert (tc->pos->meta,
				       counter->plugin_name,
				       counter->type,
				       counter->format,
				       counter->data_mime_type, counter->data,
				       counter->data_size); 
  }
  GNUNET_assert (GNUNET_YES ==
		 GNUNET_CONTAINER_multihashmap_remove (tc->metacounter,
						       key,
						       counter));
  GNUNET_free (counter);
  return GNUNET_YES;
}

/**
 * Add the given keyword to the keyword statistics tracker.
 *
 * @param cls the multihashmap we store the keyword counters in
 * @param keyword the keyword to count
 * @param is_mandatory ignored
 * @return always GNUNET_OK
 */
static int
add_to_keyword_counter (void *cls, const char *keyword, int is_mandatory)
{
  struct GNUNET_CONTAINER_MultiHashMap *mcm = cls;
  struct KeywordCounter *cnt;
  GNUNET_HashCode hc;
  size_t klen;

  klen = strlen (keyword) + 1;
  GNUNET_CRYPTO_hash (keyword, klen - 1, &hc);
  cnt = GNUNET_CONTAINER_multihashmap_get (mcm, &hc);
  if (cnt == NULL)
  {
    cnt = GNUNET_malloc (sizeof (struct KeywordCounter) + klen);
    cnt->value = (const char *) &cnt[1];
    memcpy (&cnt[1], keyword, klen);
    GNUNET_assert (GNUNET_OK ==
		   GNUNET_CONTAINER_multihashmap_put (mcm, 
						      &hc, cnt,
						      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
  }
  cnt->count++;
  return GNUNET_OK;
}

/**
 * Store an item in the datastore.
 *
 * @param h handle to the datacache
 * @param key key to store data under
 * @param size number of bytes in data
 * @param data data to store
 * @param type type of the value
 * @param discard_time when to discard the value in any case
 * @return GNUNET_OK on success, GNUNET_SYSERR on error (full, etc.)
 */
int
GNUNET_DATACACHE_put (struct GNUNET_DATACACHE_Handle *h,
                      const GNUNET_HashCode * key, size_t size,
                      const char *data, enum GNUNET_BLOCK_Type type,
                      struct GNUNET_TIME_Absolute discard_time)
{
  uint32_t used;

  used = h->api->put (h->api->cls, key, size, data, type, discard_time);
  if (used == 0)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Stored data under key `%s' in cache\n",
       GNUNET_h2s (key));
  GNUNET_STATISTICS_update (h->stats, gettext_noop ("# bytes stored"), size,
                            GNUNET_NO);
  GNUNET_CONTAINER_bloomfilter_add (h->filter, key);
  while (h->utilization + used > h->env.quota)
    GNUNET_assert (GNUNET_OK == h->api->del (h->api->cls));
  h->utilization += used;
  return GNUNET_OK;
}

/**
 * The given request hit its timeout.  Remove from the
 * doubly-linked list and call the respective continuation.
 *
 * @param cls the transmit handle of the request that timed out
 * @param tc context, can be NULL (!)
 */
static void
transmission_timeout (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct PeerRecord *pr = cls;
  struct GNUNET_CORE_Handle *h = pr->ch;
  struct GNUNET_CORE_TransmitHandle *th;

  pr->timeout_task = GNUNET_SCHEDULER_NO_TASK;
  if (GNUNET_SCHEDULER_NO_TASK != pr->ntr_task)
  {
    GNUNET_SCHEDULER_cancel (pr->ntr_task);
    pr->ntr_task = GNUNET_SCHEDULER_NO_TASK;
  }
  th = &pr->th;
  th->peer = NULL;
  if ((NULL != pr->prev) || (NULL != pr->next) || (pr == h->ready_peer_head))
  {
    /* the request that was 'approved' by core was
     * canceled before it could be transmitted; remove
     * us from the 'ready' list */
    GNUNET_CONTAINER_DLL_remove (h->ready_peer_head, h->ready_peer_tail, pr);
  }
  if (NULL != th->cm)
  {
     /* we're currently in the control queue, remove */
    GNUNET_CONTAINER_DLL_remove (h->control_pending_head,
                                 h->control_pending_tail, th->cm);
    GNUNET_free (th->cm);
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Signalling timeout of request for transmission to peer `%s' via CORE\n",
       GNUNET_i2s (&pr->peer));
  trigger_next_request (h, GNUNET_NO);

  GNUNET_assert (0 == th->get_message (th->get_message_cls, 0, NULL));
}

/**
 * Calculate when we would like to send the next HELLO to this
 * peer and ask for it.
 *
 * @param cls for which peer to schedule the HELLO
 * @param tc task context
 */
static void
schedule_next_hello (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct Peer *pl = cls;
  struct FindAdvHelloContext fah;
  size_t next_want;
  struct GNUNET_TIME_Relative delay;

  pl->hello_delay_task = GNUNET_SCHEDULER_NO_TASK;
  GNUNET_assert (GNUNET_YES == pl->is_connected);
  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
    return;                     /* we're out of here */
  if (pl->hello_req != NULL)
    return;                     /* did not finish sending the previous one */
  /* find applicable HELLOs */
  fah.peer = pl;
  fah.result = NULL;
  fah.max_size = GNUNET_SERVER_MAX_MESSAGE_SIZE - 1;
  fah.next_adv = GNUNET_TIME_UNIT_FOREVER_REL;
  GNUNET_CONTAINER_multihashmap_iterate (peers, &find_advertisable_hello, &fah);
  pl->hello_delay_task =
      GNUNET_SCHEDULER_add_delayed (fah.next_adv, &schedule_next_hello, pl);
  if (fah.result == NULL)
    return;
  next_want = GNUNET_HELLO_size (fah.result->hello);
  delay = GNUNET_TIME_absolute_get_remaining (pl->next_hello_allowed);
  if (delay.rel_value == 0)
  {
    /* now! */
    pl->hello_req =
        GNUNET_CORE_notify_transmit_ready (handle, GNUNET_YES, 0,
                                           GNUNET_CONSTANTS_SERVICE_TIMEOUT,
                                           &pl->pid, next_want,
                                           &hello_advertising_ready, pl);
  }
}