/**
 * We've succeeded in establishing a connection.
 *
 * @param connection the connection we tried to establish
 */
static void
connect_success_continuation (struct GNUNET_CONNECTION_Handle *connection)
{
  LOG (GNUNET_ERROR_TYPE_DEBUG, 
       "Connection to `%s' succeeded! (%p)\n",
       GNUNET_a2s (connection->addr, connection->addrlen), connection);
  /* trigger jobs that waited for the connection */
  if (NULL != connection->receiver)
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Connection succeeded, starting with receiving data (%p)\n", 
	 connection);
    GNUNET_assert (GNUNET_SCHEDULER_NO_TASK == connection->read_task);
    connection->read_task =
      GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_absolute_get_remaining
                                     (connection->receive_timeout), connection->sock,
                                     &receive_ready, connection);
  }
  if (NULL != connection->nth.notify_ready)
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Connection succeeded, starting with sending data (%p)\n",
         connection);
    GNUNET_assert (connection->nth.timeout_task != GNUNET_SCHEDULER_NO_TASK);
    GNUNET_SCHEDULER_cancel (connection->nth.timeout_task);
    connection->nth.timeout_task = GNUNET_SCHEDULER_NO_TASK;
    GNUNET_assert (connection->write_task == GNUNET_SCHEDULER_NO_TASK);
    connection->write_task =
        GNUNET_SCHEDULER_add_write_net (GNUNET_TIME_absolute_get_remaining
                                        (connection->nth.transmit_timeout), connection->sock,
                                        &transmit_ready, connection);
  }
}

/**
 * PEERINFO calls this function to let us know about a possible peer
 * that we might want to connect to.
 *
 * @param cls closure (not used)
 * @param peer potential peer to connect to
 * @param hello HELLO for this peer (or NULL)
 * @param err_msg NULL if successful, otherwise contains error message
 */
static void
process_peer (void *cls, const struct GNUNET_PeerIdentity *peer,
              const struct GNUNET_HELLO_Message *hello, const char *err_msg)
{
  struct Peer *pos;

  if (err_msg != NULL)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                _("Error in communication with PEERINFO service: %s\n"),
                err_msg);
    GNUNET_PEERINFO_notify_cancel (peerinfo_notify);
    peerinfo_notify = GNUNET_PEERINFO_notify (cfg, &process_peer, NULL);
    return;
  }
  GNUNET_assert (peer != NULL);
  if (0 == memcmp (&my_identity, peer, sizeof (struct GNUNET_PeerIdentity)))
    return;                     /* that's me! */
  if (hello == NULL)
  {
    /* free existing HELLO, if any */
    pos = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey);
    if (NULL != pos)
    {
      GNUNET_free_non_null (pos->hello);
      pos->hello = NULL;
      if (pos->filter != NULL)
      {
        GNUNET_CONTAINER_bloomfilter_free (pos->filter);
        pos->filter = NULL;
      }
      if ((GNUNET_NO == pos->is_connected) && (GNUNET_NO == pos->is_friend) &&
          (0 ==
           GNUNET_TIME_absolute_get_remaining (pos->
                                               greylisted_until).rel_value))
        free_peer (NULL, &pos->pid.hashPubKey, pos);
    }
    return;
  }
  consider_for_advertising (hello);
  pos = GNUNET_CONTAINER_multihashmap_get (peers, &peer->hashPubKey);
  if (pos == NULL)
    pos = make_peer (peer, hello, GNUNET_NO);
  GNUNET_assert (NULL != pos);
  if (GNUNET_YES == pos->is_connected)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Already connected to peer `%s'\n",
                GNUNET_i2s (peer));
    return;
  }
  if (GNUNET_TIME_absolute_get_remaining (pos->greylisted_until).rel_value > 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Already tried peer `%s' recently\n",
                GNUNET_i2s (peer));
    return;                     /* peer still greylisted */
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Considering connecting to peer `%s'\n",
              GNUNET_i2s (peer));
  schedule_attempt_connect (pos);
}

/**
 * Figure out when and how to transmit to the given peer.
 *
 * @param cls the 'struct PeerPlan'
 * @param tc scheduler context
 */
static void
schedule_peer_transmission (void *cls,
                            const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct PeerPlan *pp = cls;
  struct GSF_RequestPlan *rp;
  size_t msize;
  struct GNUNET_TIME_Relative delay;

  pp->task = GNUNET_SCHEDULER_NO_TASK;
  if (NULL != pp->pth)
  {
    GSF_peer_transmit_cancel_ (pp->pth);
    pp->pth = NULL;
  }
  /* move ready requests to priority queue */
  while ((NULL != (rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap))) &&
         (GNUNET_TIME_absolute_get_remaining
          (rp->earliest_transmission).rel_value == 0))
  {
    GNUNET_assert (rp == GNUNET_CONTAINER_heap_remove_root (pp->delay_heap));
    rp->hn = GNUNET_CONTAINER_heap_insert (pp->priority_heap, rp, rp->priority);
  }
  if (0 == GNUNET_CONTAINER_heap_get_size (pp->priority_heap))
  {
    /* priority heap (still) empty, check for delay... */
    rp = GNUNET_CONTAINER_heap_peek (pp->delay_heap);
    if (NULL == rp)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "No active requests for plan %p.\n",
                  pp);
      return;                   /* both queues empty */
    }
    delay = GNUNET_TIME_absolute_get_remaining (rp->earliest_transmission);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Sleeping for %llu ms before retrying requests on plan %p.\n",
                (unsigned long long) delay.rel_value, pp);
    GNUNET_STATISTICS_set (GSF_stats, gettext_noop ("# delay heap timeout"),
                           delay.rel_value, GNUNET_NO);

    pp->task =
        GNUNET_SCHEDULER_add_delayed (delay, &schedule_peer_transmission, pp);
    return;
  }
#if INSANE_STATISTICS
  GNUNET_STATISTICS_update (GSF_stats, gettext_noop ("# query plans executed"),
                            1, GNUNET_NO);
#endif
  /* process from priority heap */
  rp = GNUNET_CONTAINER_heap_peek (pp->priority_heap);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Executing query plan %p\n", rp);
  GNUNET_assert (NULL != rp);
  msize = GSF_pending_request_get_message_ (get_latest (rp), 0, NULL);
  pp->pth =
      GSF_peer_transmit_ (pp->cp, GNUNET_YES, rp->priority,
                          GNUNET_TIME_UNIT_FOREVER_REL, msize,
                          &transmit_message_callback, pp);
  GNUNET_assert (NULL != pp->pth);
}

/**
 * Do address validation again to keep address valid.
 *
 * @param cls the 'struct ValidationEntry'
 * @param tc scheduler context (unused)
 */
static void
revalidate_address (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct ValidationEntry *ve = cls;
  struct GNUNET_TIME_Relative canonical_delay;
  struct GNUNET_TIME_Relative delay;
  struct GST_BlacklistCheck *bc;
  uint32_t rdelay;

  ve->revalidation_task = GNUNET_SCHEDULER_NO_TASK;
  delay = GNUNET_TIME_absolute_get_remaining (ve->revalidation_block);
  /* How long until we can possibly permit the next PING? */
  canonical_delay =
      (ve->in_use ==
       GNUNET_YES) ? CONNECTED_PING_FREQUENCY
      : ((GNUNET_TIME_absolute_get_remaining (ve->valid_until).rel_value >
          0) ? VALIDATED_PING_FREQUENCY : UNVALIDATED_PING_KEEPALIVE);
  if (delay.rel_value > canonical_delay.rel_value * 2)
  {
    /* situation changed, recalculate delay */
    delay = canonical_delay;
    ve->revalidation_block = GNUNET_TIME_relative_to_absolute (delay);
  }
  if (delay.rel_value > 0)
  {
    /* should wait a bit longer */
    ve->revalidation_task =
        GNUNET_SCHEDULER_add_delayed (delay, &revalidate_address, ve);
    return;
  }
  ve->revalidation_block = GNUNET_TIME_relative_to_absolute (canonical_delay);

  /* schedule next PINGing with some extra random delay to avoid synchronous re-validations */
  rdelay =
      GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_WEAK,
                                canonical_delay.rel_value);
  delay =
      GNUNET_TIME_relative_add (canonical_delay,
                                GNUNET_TIME_relative_multiply
                                (GNUNET_TIME_UNIT_MILLISECONDS, rdelay));
  ve->revalidation_task =
      GNUNET_SCHEDULER_add_delayed (delay, &revalidate_address, ve);

  /* start PINGing by checking blacklist */
  GNUNET_STATISTICS_update (GST_stats,
                            gettext_noop ("# address revalidations started"), 1,
                            GNUNET_NO);
  bc = GST_blacklist_test_allowed (&ve->pid, ve->address->transport_name,
                                   &transmit_ping_if_allowed, ve);
  if (NULL != bc)
    ve->bc = bc;                /* only set 'bc' if 'transmit_ping_if_allowed' was not already
                                 * called... */
}

/**
 * This task is run if we should re-try connection to the
 * service after a while.
 *
 * @param cls our `struct GNUNET_CLIENT_TransmitHandle` of the request
 * @param tc unused
 */
static void
client_delayed_retry (void *cls,
		      const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct GNUNET_CLIENT_TransmitHandle *th = cls;
  struct GNUNET_TIME_Relative delay;

  th->reconnect_task = NULL;
  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
  {
    /* give up, was shutdown */
    th->client->th = NULL;
    th->notify (th->notify_cls, 0, NULL);
    GNUNET_free (th);
    return;
  }
  th->client->connection =
    do_connect (th->client->service_name,
		th->client->cfg,
		th->client->attempts++);
  th->client->first_message = GNUNET_YES;
  if (NULL == th->client->connection)
  {
    /* could happen if we're out of sockets */
    delay = GNUNET_TIME_relative_min (GNUNET_TIME_absolute_get_remaining (th->timeout),
                                      th->client->back_off);
    th->client->back_off = GNUNET_TIME_STD_BACKOFF (th->client->back_off);
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Transmission failed %u times, trying again in %s.\n",
         MAX_ATTEMPTS - th->attempts_left,
         GNUNET_STRINGS_relative_time_to_string (delay, GNUNET_YES));
    GNUNET_assert (NULL == th->th);
    GNUNET_assert (NULL == th->reconnect_task);
    th->reconnect_task =
        GNUNET_SCHEDULER_add_delayed (delay, &client_delayed_retry, th);
    return;
  }
  th->th =
      GNUNET_CONNECTION_notify_transmit_ready (th->client->connection, th->size,
                                               GNUNET_TIME_absolute_get_remaining
                                               (th->timeout), &client_notify,
                                               th);
  if (NULL == th->th)
  {
    GNUNET_break (0);
    th->client->th = NULL;
    th->notify (th->notify_cls, 0, NULL);
    GNUNET_free (th);
    return;
  }
}

static void
do_stop (void *cls,
         const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  char *fn = cls;

  if (0 ==
      GNUNET_TIME_absolute_get_remaining (GNUNET_TIME_absolute_add (start_time,
                                                                    TIMEOUT)).rel_value_us)
  {
    GNUNET_break (0);
    ret = 1;
  }
  else
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Finished download, shutting down\n",
                (unsigned long long) FILESIZE);
  }
  if (NULL != fn)
  {
    GNUNET_DISK_directory_remove (fn);
    GNUNET_free (fn);
  }
  GNUNET_SCHEDULER_shutdown ();
}

/**
 * Iterator which adds the given address to the set of validated
 * addresses.
 *
 * @param cls original HELLO message
 * @param address the address
 * @param expiration expiration time
 * @return GNUNET_OK (keep the address)
 */
static int
add_valid_address (void *cls, const struct GNUNET_HELLO_Address *address,
                   struct GNUNET_TIME_Absolute expiration)
{
  const struct GNUNET_HELLO_Message *hello = cls;
  struct ValidationEntry *ve;
  struct GNUNET_PeerIdentity pid;
  struct GNUNET_CRYPTO_RsaPublicKeyBinaryEncoded public_key;

  if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value == 0)
    return GNUNET_OK;           /* expired */
  if ((GNUNET_OK != GNUNET_HELLO_get_id (hello, &pid)) ||
      (GNUNET_OK != GNUNET_HELLO_get_key (hello, &public_key)))
  {
    GNUNET_break (0);
    return GNUNET_OK;           /* invalid HELLO !? */
  }
  if (0 == memcmp (&GST_my_identity, &pid, sizeof (struct GNUNET_PeerIdentity)))
  {
    /* Peerinfo returned own identity, skip validation */
    return GNUNET_OK;
  }

  ve = find_validation_entry (&public_key, address);
  ve->valid_until = GNUNET_TIME_absolute_max (ve->valid_until, expiration);

  if (GNUNET_SCHEDULER_NO_TASK == ve->revalidation_task)
    ve->revalidation_task = GNUNET_SCHEDULER_add_now (&revalidate_address, ve);
  GNUNET_ATS_address_update (GST_ats, address, NULL, NULL, 0);
  return GNUNET_OK;
}

static void
do_stop (void *cls)
{
  struct GNUNET_TIME_Relative del;
  char *fancy;

  GNUNET_SCHEDULER_shutdown ();
  if (0 ==
      GNUNET_TIME_absolute_get_remaining (GNUNET_TIME_absolute_add (start_time,
                                                                    TIMEOUT)).rel_value_us)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Timeout during download, shutting down with error\n");
    ok = 1;
  }
  else
  {
    del = GNUNET_TIME_absolute_get_duration (start_time);
    if (del.rel_value_us == 0)
      del.rel_value_us = 1;
    fancy =
        GNUNET_STRINGS_byte_size_fancy (((unsigned long long) FILESIZE) *
                                        1000000LL / del.rel_value_us);
    FPRINTF (stdout,
             "Download speed was %s/s\n",
             fancy);
    GNUNET_free (fancy);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Finished download, shutting down\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;
}

/**
 * We've transmitted the iteration request.  Now get ready to process
 * the results (or handle transmission error).
 *
 * @param cls the 'struct GNUNET_PEERINFO_IteratorContext'
 * @param emsg error message, NULL if transmission worked
 */
static void
iterator_start_receive (void *cls, const char *emsg)
{
  struct GNUNET_PEERINFO_IteratorContext *ic = cls;
  struct GNUNET_PEERINFO_Handle *h = ic->h;
  GNUNET_PEERINFO_Processor cb;
  void *cb_cls;

  ic->ac = NULL;
  if (NULL != emsg)
  {
    cb = ic->callback;
    cb_cls = ic->callback_cls;
    GNUNET_PEERINFO_iterate_cancel (ic);
    reconnect (h);
    if (NULL != cb)
      cb (cb_cls, NULL, NULL, emsg);
    return;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Waiting for response from `%s' service.\n",
       "PEERINFO");
  ic->in_receive = GNUNET_YES;
  if (GNUNET_NO == h->in_receive)
  {
    h->in_receive = GNUNET_YES;
    GNUNET_CLIENT_receive (h->client, &peerinfo_handler, h,
			   GNUNET_TIME_absolute_get_remaining (ic->timeout));
  }
}

/**
 * If possible, write a shutdown message to the target
 * buffer and destroy the client connection.
 *
 * @param cls the "struct GNUNET_CLIENT_Connection" to destroy
 * @param size number of bytes available in buf
 * @param buf NULL on error, otherwise target buffer
 * @return number of bytes written to buf
 */
static size_t
write_shutdown (void *cls, size_t size, void *buf)
{
  struct GNUNET_MessageHeader *msg;
  struct ShutdownContext *shutdown_ctx = cls;

  if (size < sizeof (struct GNUNET_MessageHeader))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
		  _("Failed to transmit shutdown request to client.\n"));
      shutdown_ctx->cont (shutdown_ctx->cont_cls, GNUNET_SYSERR);
      GNUNET_CLIENT_disconnect (shutdown_ctx->sock);
      GNUNET_free (shutdown_ctx);
      return 0;			/* client disconnected */
    }

  GNUNET_CLIENT_receive (shutdown_ctx->sock, &service_shutdown_handler,
			 shutdown_ctx, GNUNET_TIME_UNIT_FOREVER_REL);
  shutdown_ctx->cancel_task =
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
				  (shutdown_ctx->timeout),
				  &service_shutdown_cancel, shutdown_ctx);
  msg = (struct GNUNET_MessageHeader *) buf;
  msg->type = htons (GNUNET_MESSAGE_TYPE_ARM_SHUTDOWN);
  msg->size = htons (sizeof (struct GNUNET_MessageHeader));
  return sizeof (struct GNUNET_MessageHeader);
}

/**
 * Task that looks at the 'retry_heap' and transmits all of the requests
 * on the heap that are ready for transmission.  Then re-schedules
 * itself (unless the heap is empty).
 *
 * @param cls unused
 * @param tc scheduler context
 */
static void
transmit_next_request_task (void *cls,
                            const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct ClientQueryRecord *cqr;
  struct GNUNET_TIME_Relative delay;

  retry_task = GNUNET_SCHEDULER_NO_TASK;
  if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
    return;
  while (NULL != (cqr = GNUNET_CONTAINER_heap_remove_root (retry_heap)))
  {
    cqr->hnode = NULL;
    delay = GNUNET_TIME_absolute_get_remaining (cqr->retry_time);
    if (delay.rel_value_us > 0)
    {
      cqr->hnode =
          GNUNET_CONTAINER_heap_insert (retry_heap, cqr,
                                        cqr->retry_time.abs_value_us);
      retry_task =
          GNUNET_SCHEDULER_add_delayed (delay, &transmit_next_request_task,
                                        NULL);
      return;
    }
    transmit_request (cqr);
    cqr->hnode =
        GNUNET_CONTAINER_heap_insert (retry_heap, cqr,
                                      cqr->retry_time.abs_value_us);
  }
}

/**
 * Function called to notify a client about the socket
 * begin ready to queue the message.  @a buf will be
 * NULL and @a size zero if the socket was closed for
 * writing in the meantime.
 *
 * @param cls closure of type `struct TransmitGetResponseContext *`
 * @param size number of bytes available in @a buf
 * @param buf where the callee should write the message
 * @return number of bytes written to @a buf
 */
static size_t
transmit_for_response (void *cls,
		       size_t size,
		       void *buf)
{
  struct TransmitGetResponseContext *tc = cls;
  uint16_t msize;

  tc->client->tag = NULL;
  msize = ntohs (tc->hdr->size);
  if (NULL == buf)
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Could not submit request, not expecting to receive a response.\n");
    if (NULL != tc->rn)
      tc->rn (tc->rn_cls, NULL);
    GNUNET_free (tc);
    return 0;
  }
  GNUNET_assert (size >= msize);
  memcpy (buf, tc->hdr, msize);
  GNUNET_CLIENT_receive (tc->client,
                         tc->rn,
                         tc->rn_cls,
                         GNUNET_TIME_absolute_get_remaining (tc->timeout));
  GNUNET_free (tc);
  return msize;
}

/**
 * Send the 'TEST' message to the service.  If successful, prepare to
 * receive the reply.
 *
 * @param cls the `struct GNUNET_CLIENT_TestHandle` of the test
 * @param size number of bytes available in @a buf
 * @param buf where to write the message
 * @return number of bytes written to @a buf
 */
static size_t
write_test (void *cls, size_t size, void *buf)
{
  struct GNUNET_CLIENT_TestHandle *th = cls;
  struct GNUNET_MessageHeader *msg;

  th->th = NULL;
  if (size < sizeof (struct GNUNET_MessageHeader))
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
	 "Failed to transmit TEST request.\n");
    service_test_report (th, GNUNET_NO);
    return 0;                   /* client disconnected */
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Transmitting `%s' request.\n",
       "TEST");
  msg = (struct GNUNET_MessageHeader *) buf;
  msg->type = htons (GNUNET_MESSAGE_TYPE_TEST);
  msg->size = htons (sizeof (struct GNUNET_MessageHeader));
  GNUNET_CLIENT_receive (th->client,
			 &confirm_handler, th,
                         GNUNET_TIME_absolute_get_remaining
                         (th->test_deadline));
  return sizeof (struct GNUNET_MessageHeader);
}

/**
 * Receive data from the given connection.  Note that this function will
 * call "receiver" asynchronously using the scheduler.  It will
 * "immediately" return.  Note that there MUST only be one active
 * receive call per connection at any given point in time (so do not
 * call receive again until the receiver callback has been invoked).
 *
 * @param connection connection handle
 * @param max maximum number of bytes to read
 * @param timeout maximum amount of time to wait
 * @param receiver function to call with received data
 * @param receiver_cls closure for receiver
 */
void
GNUNET_CONNECTION_receive (struct GNUNET_CONNECTION_Handle *connection, size_t max,
                           struct GNUNET_TIME_Relative timeout,
                           GNUNET_CONNECTION_Receiver receiver,
                           void *receiver_cls)
{
  GNUNET_assert ((GNUNET_SCHEDULER_NO_TASK == connection->read_task) &&
                 (NULL == connection->receiver));
  GNUNET_assert (NULL != receiver);
  connection->receiver = receiver;
  connection->receiver_cls = receiver_cls;
  connection->receive_timeout = GNUNET_TIME_relative_to_absolute (timeout);
  connection->max = max;
  if (NULL != connection->sock)
  {
    connection->read_task =
      GNUNET_SCHEDULER_add_read_net (GNUNET_TIME_absolute_get_remaining
                                     (connection->receive_timeout), connection->sock,
                                     &receive_ready, connection);
    return;
  }
  if ((NULL == connection->dns_active) && (NULL == connection->ap_head))
  {
    connection->receiver = NULL;
    receiver (receiver_cls, NULL, 0, NULL, 0, ETIMEDOUT);
    return;
  }
}

/**
 * Request a list of running services.
 *
 * @param h handle to ARM
 * @param timeout how long to wait before failing for good
 * @param cont callback to invoke after request is sent or is not sent
 * @param cont_cls closure for callback
 */
void
GNUNET_ARM_request_service_list (struct GNUNET_ARM_Handle *h,
                                 struct GNUNET_TIME_Relative timeout,
                                 GNUNET_ARM_ServiceListCallback cont,
                                 void *cont_cls)
{
  struct ARMControlMessage *cm;
  struct GNUNET_ARM_Message *msg;

  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Requesting LIST from ARM service with timeout: %s\n",
       GNUNET_STRINGS_relative_time_to_string (timeout, GNUNET_YES));
  cm = GNUNET_new (struct ARMControlMessage);
  cm->h = h;
  cm->list_cont = cont;
  cm->cont_cls = cont_cls;
  cm->timeout = GNUNET_TIME_relative_to_absolute (timeout);
  msg = GNUNET_malloc (sizeof (struct GNUNET_ARM_Message));
  msg->header.size = htons (sizeof (struct GNUNET_ARM_Message));
  msg->header.type = htons (GNUNET_MESSAGE_TYPE_ARM_LIST);
  msg->reserved = htonl (0);
  cm->msg = msg;
  GNUNET_CONTAINER_DLL_insert_tail (h->control_pending_head,
                                    h->control_pending_tail, cm);
  cm->timeout_task_id =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
                                    (cm->timeout), &control_message_timeout, cm);
  trigger_next_request (h, GNUNET_NO);
}

/**
 * Find a peer that would be reasonable for advertising.
 *
 * @param cls closure
 * @param pid identity of a peer
 * @param value 'struct Peer*' for the peer we are considering
 * @return GNUNET_YES (continue iteration)
 */
static int
find_advertisable_hello (void *cls, const struct GNUNET_HashCode * pid, void *value)
{
  struct FindAdvHelloContext *fah = cls;
  struct Peer *pos = value;
  struct GNUNET_TIME_Relative rst_time;
  size_t hs;

  if (pos == fah->peer)
    return GNUNET_YES;
  if (pos->hello == NULL)
    return GNUNET_YES;
  rst_time = GNUNET_TIME_absolute_get_remaining (pos->filter_expiration);
  if (0 == rst_time.rel_value)
  {
    /* time to discard... */
    GNUNET_CONTAINER_bloomfilter_free (pos->filter);
    setup_filter (pos);
  }
  fah->next_adv = GNUNET_TIME_relative_min (rst_time, fah->next_adv);
  hs = GNUNET_HELLO_size (pos->hello);
  if (hs > fah->max_size)
    return GNUNET_YES;
  if (GNUNET_NO ==
      GNUNET_CONTAINER_bloomfilter_test (pos->filter,
                                         &fah->peer->pid.hashPubKey))
    fah->result = pos;
  return GNUNET_YES;
}

/**
 * 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
 */
static void
schedule_next_hello (void *cls)
{
  struct Peer *pl = cls;
  struct FindAdvHelloContext fah;
  struct GNUNET_MQ_Envelope *env;
  size_t want;
  struct GNUNET_TIME_Relative delay;
  struct GNUNET_HashCode hc;

  pl->hello_delay_task = NULL;
  GNUNET_assert (NULL != pl->mq);
  /* 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_multipeermap_iterate (peers,
                                         &find_advertisable_hello,
                                         &fah);
  pl->hello_delay_task =
      GNUNET_SCHEDULER_add_delayed (fah.next_adv,
                                    &schedule_next_hello,
                                    pl);
  if (NULL == fah.result)
    return;
  delay = GNUNET_TIME_absolute_get_remaining (pl->next_hello_allowed);
  if (0 != delay.rel_value_us)
    return;

  want = GNUNET_HELLO_size (fah.result->hello);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Sending HELLO with %u bytes",
	      (unsigned int) want);
  env = GNUNET_MQ_msg_copy (&fah.result->hello->header);
  GNUNET_MQ_send (pl->mq,
		  env);

  /* avoid sending this one again soon */
  GNUNET_CRYPTO_hash (&pl->pid,
		      sizeof (struct GNUNET_PeerIdentity),
		      &hc);
  GNUNET_CONTAINER_bloomfilter_add (fah.result->filter,
				    &hc);

  GNUNET_STATISTICS_update (stats,
			    gettext_noop ("# HELLO messages gossipped"),
			    1,
			    GNUNET_NO);
  /* prepare to send the next one */
  if (NULL != pl->hello_delay_task)
    GNUNET_SCHEDULER_cancel (pl->hello_delay_task);
  pl->next_hello_allowed
    = GNUNET_TIME_relative_to_absolute (HELLO_ADVERTISEMENT_MIN_FREQUENCY);
  pl->hello_delay_task
    = GNUNET_SCHEDULER_add_now (&schedule_next_hello,
				pl);
}

/**
 * Schedule a task to try to connect to the specified peer.
 *
 * @param pos peer to connect to
 */
static void
schedule_attempt_connect (struct Peer *pos)
{
  if (GNUNET_SCHEDULER_NO_TASK != pos->attempt_connect_task)
    return;
  pos->attempt_connect_task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining (next_connect_attempt),
							    &do_attempt_connect,
							    pos);
}

/**
 * Start or stop a service.
 *
 * @param h handle to ARM
 * @param service_name name of the service
 * @param timeout how long to wait before failing for good
 * @param cb callback to invoke when service is ready
 * @param cb_cls closure for callback
 * @param type type of the request
 */
static void
change_service (struct GNUNET_ARM_Handle *h, const char *service_name,
		struct GNUNET_TIME_Relative timeout, GNUNET_ARM_ResultCallback cb,
		void *cb_cls, uint16_t type)
{
  struct ARMControlMessage *cm;
  size_t slen;
  struct GNUNET_ARM_Message *msg;

  slen = strlen (service_name) + 1;
  if (slen + sizeof (struct GNUNET_ARM_Message) >=
      GNUNET_SERVER_MAX_MESSAGE_SIZE)
  {
    GNUNET_break (0);
    if (cb != NULL)
      cb (cb_cls, GNUNET_ARM_REQUEST_TOO_LONG, NULL, 0);
    return;
  }
  LOG (GNUNET_ERROR_TYPE_DEBUG, "Requesting %s of service `%s'.\n",
       (GNUNET_MESSAGE_TYPE_ARM_START == type) ? "start" : "termination",
       service_name);
  cm = GNUNET_malloc (sizeof (struct ARMControlMessage) + slen);
  cm->h = h;
  cm->result_cont = cb;
  cm->cont_cls = cb_cls;
  cm->timeout = GNUNET_TIME_relative_to_absolute (timeout);
  memcpy (&cm[1], service_name, slen);
  msg = GNUNET_malloc (sizeof (struct GNUNET_ARM_Message) + slen);
  msg->header.size = htons (sizeof (struct GNUNET_ARM_Message) + slen);
  msg->header.type = htons (type);
  msg->reserved = htonl (0);
  memcpy (&msg[1], service_name, slen);
  cm->msg = msg;
  LOG (GNUNET_ERROR_TYPE_DEBUG,
      "Inserting a control message into the queue. Timeout is %s\n",
       GNUNET_STRINGS_relative_time_to_string (GNUNET_TIME_absolute_get_remaining (cm->timeout),
					       GNUNET_NO));
  GNUNET_CONTAINER_DLL_insert_tail (h->control_pending_head,
                                    h->control_pending_tail, cm);
  cm->timeout_task_id =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
                                    (cm->timeout), &control_message_timeout, cm);
  trigger_next_request (h, GNUNET_NO);
}