/**
 * Get a result for a particular key from the namestore.  The processor
 * will only be called once.  
 *
 * @param h handle to the namestore
 * @param zone zone to look up a record from
 * @param name name to look up
 * @param record_type desired record type, 0 for all
 * @param proc function to call on the matching records, or with
 *        NULL (rd_count == 0) if there are no matching records
 * @param proc_cls closure for proc
 * @return a handle that can be used to
 *         cancel
 */
struct GNUNET_NAMESTORE_QueueEntry *
GNUNET_NAMESTORE_lookup_record (struct GNUNET_NAMESTORE_Handle *h, 
			      const struct GNUNET_CRYPTO_ShortHashCode *zone,
			      const char *name,
			      uint32_t record_type,
			      GNUNET_NAMESTORE_RecordProcessor proc, void *proc_cls)
{
  struct GNUNET_NAMESTORE_QueueEntry *qe;
  struct PendingMessage *pe;
  size_t msg_size = 0;
  size_t name_len = 0;
  uint32_t rid = 0;

  GNUNET_assert (NULL != h);
  GNUNET_assert (NULL != zone);
  GNUNET_assert (NULL != name);

  name_len = strlen (name) + 1;
  if ((name_len == 0) || (name_len > 256))
  {
    GNUNET_break (0);
    return NULL;
  }

  rid = get_op_id(h);
  qe = GNUNET_malloc(sizeof (struct GNUNET_NAMESTORE_QueueEntry));
  qe->nsh = h;
  qe->proc = proc;
  qe->proc_cls = proc_cls;
  qe->op_id = rid;
  GNUNET_CONTAINER_DLL_insert_tail(h->op_head, h->op_tail, qe);

  /* set msg_size*/
  msg_size = sizeof (struct LookupNameMessage) + name_len;
  pe = GNUNET_malloc(sizeof (struct PendingMessage) + msg_size);

  /* create msg here */
  struct LookupNameMessage * msg;
  pe->size = msg_size;
  pe->is_init = GNUNET_NO;
  msg = (struct LookupNameMessage *) &pe[1];
  msg->gns_header.header.type = htons (GNUNET_MESSAGE_TYPE_NAMESTORE_LOOKUP_NAME);
  msg->gns_header.header.size = htons (msg_size);
  msg->gns_header.r_id = htonl (rid);
  msg->record_type = htonl (record_type);
  msg->name_len = htonl (name_len);
  msg->zone = *zone;
  memcpy (&msg[1], name, name_len);

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message for name `%s'\n", "NAMESTORE_LOOKUP_NAME", name);

  /* transmit message */
  GNUNET_CONTAINER_DLL_insert_tail (h->pending_head, h->pending_tail, pe);
  do_transmit(h);

  return qe;
}

/**
 * Call a method for each known matching host and change its trust
 * value.  The callback method will be invoked once for each matching
 * host and then finally once with a NULL pointer.  After that final
 * invocation, the iterator context must no longer be used.
 *
 * Instead of calling this function with 'peer == NULL' it is often
 * better to use 'GNUNET_PEERINFO_notify'.
 *
 * @param h handle to the peerinfo service
 * @param peer restrict iteration to this peer only (can be NULL)
 * @param timeout how long to wait until timing out
 * @param callback the method to call for each peer
 * @param callback_cls closure for callback
 * @return iterator context
 */
struct GNUNET_PEERINFO_IteratorContext *
GNUNET_PEERINFO_iterate (struct GNUNET_PEERINFO_Handle *h,
                         const struct GNUNET_PeerIdentity *peer,
                         struct GNUNET_TIME_Relative timeout,
                         GNUNET_PEERINFO_Processor callback, void *callback_cls)
{
  struct GNUNET_MessageHeader *lapm;
  struct ListPeerMessage *lpm;
  struct GNUNET_PEERINFO_IteratorContext *ic;
  struct GNUNET_PEERINFO_AddContext *ac;

  ic = GNUNET_malloc (sizeof (struct GNUNET_PEERINFO_IteratorContext));
  if (NULL == peer)
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Requesting list of peers from PEERINFO service\n");
    ac =
        GNUNET_malloc (sizeof (struct GNUNET_PEERINFO_AddContext) +
                       sizeof (struct GNUNET_MessageHeader));
    ac->size = sizeof (struct GNUNET_MessageHeader);
    lapm = (struct GNUNET_MessageHeader *) &ac[1];
    lapm->size = htons (sizeof (struct GNUNET_MessageHeader));
    lapm->type = htons (GNUNET_MESSAGE_TYPE_PEERINFO_GET_ALL);
  }
  else
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Requesting information on peer `%4s' from PEERINFO service\n",
         GNUNET_i2s (peer));
    ac =
        GNUNET_malloc (sizeof (struct GNUNET_PEERINFO_AddContext) +
                       sizeof (struct ListPeerMessage));
    ac->size = sizeof (struct ListPeerMessage);
    lpm = (struct ListPeerMessage *) &ac[1];
    lpm->header.size = htons (sizeof (struct ListPeerMessage));
    lpm->header.type = htons (GNUNET_MESSAGE_TYPE_PEERINFO_GET);
    memcpy (&lpm->peer, peer, sizeof (struct GNUNET_PeerIdentity));
    ic->have_peer = GNUNET_YES;
    ic->peer = *peer;
  }
  ic->h = h;
  ic->ac = ac;
  ic->callback = callback;
  ic->callback_cls = callback_cls;
  ic->timeout = GNUNET_TIME_relative_to_absolute (timeout);
  ic->timeout_task =
      GNUNET_SCHEDULER_add_delayed (timeout, &signal_timeout, ic);
  ac->cont = &iterator_start_receive;
  ac->cont_cls = ic;
  GNUNET_CONTAINER_DLL_insert_tail (h->ac_head, h->ac_tail, ac);
  GNUNET_CONTAINER_DLL_insert_tail (h->ic_head,
				    h->ic_tail,
				    ic);
  trigger_transmit (h);
  return ic;
}

/**
 * Look for an existing PKEY delegation record for a given public key.
 * Returns at most one result to the processor.
 *
 * @param h handle to the namestore
 * @param zone hash of public key of the zone to look up in, never NULL
 * @param value_zone hash of the public key of the target zone (value), never NULL
 * @param proc function to call on the matching records, or with
 *        NULL (rd_count == 0) if there are no matching records
 * @param proc_cls closure for proc
 * @return a handle that can be used to
 *         cancel
 */
struct GNUNET_NAMESTORE_QueueEntry *
GNUNET_NAMESTORE_zone_to_name (struct GNUNET_NAMESTORE_Handle *h,
                               const struct GNUNET_CRYPTO_ShortHashCode *zone,
                               const struct GNUNET_CRYPTO_ShortHashCode *value_zone,
                               GNUNET_NAMESTORE_RecordProcessor proc, void *proc_cls)
{
  struct GNUNET_NAMESTORE_QueueEntry *qe;
  struct PendingMessage *pe;
  size_t msg_size = 0;
  uint32_t rid = 0;

  GNUNET_assert (NULL != h);
  GNUNET_assert (NULL != zone);
  GNUNET_assert (NULL != value_zone);

  rid = get_op_id(h);
  qe = GNUNET_malloc(sizeof (struct GNUNET_NAMESTORE_QueueEntry));
  qe->nsh = h;
  qe->proc = proc;
  qe->proc_cls = proc_cls;
  qe->op_id = rid;
  GNUNET_CONTAINER_DLL_insert_tail(h->op_head, h->op_tail, qe);

  /* set msg_size*/
  msg_size = sizeof (struct ZoneToNameMessage);
  pe = GNUNET_malloc(sizeof (struct PendingMessage) + msg_size);

  /* create msg here */
  struct ZoneToNameMessage * msg;
  pe->size = msg_size;
  pe->is_init = GNUNET_NO;
  msg = (struct ZoneToNameMessage *) &pe[1];
  msg->gns_header.header.type = htons (GNUNET_MESSAGE_TYPE_NAMESTORE_ZONE_TO_NAME);
  msg->gns_header.header.size = htons (msg_size);
  msg->gns_header.r_id = htonl (rid);
  msg->zone = *zone;
  msg->value_zone = *value_zone;

  char * z_tmp = GNUNET_strdup (GNUNET_short_h2s (zone));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Sending `%s' message for zone `%s' in zone `%s'\n",
      "NAMESTORE_ZONE_TO_NAME",
      z_tmp,
      GNUNET_short_h2s (value_zone));
  GNUNET_free (z_tmp);

  /* transmit message */
  GNUNET_CONTAINER_DLL_insert_tail (h->pending_head, h->pending_tail, pe);
  do_transmit(h);

  return qe;
}

/**
 * Send an message to the helper.
 *
 * @param h helper to send message to
 * @param msg message to send
 * @param can_drop can the message be dropped if there is already one in the queue?
 * @param cont continuation to run once the message is out (#GNUNET_OK on succees, #GNUNET_NO
 *             if the helper process died, #GNUNET_SYSERR during #GNUNET_HELPER_destroy).
 * @param cont_cls closure for @a cont
 * @return NULL if the message was dropped,
 *         otherwise handle to cancel *cont* (actual transmission may
 *         not be abortable)
 */
struct GNUNET_HELPER_SendHandle *
GNUNET_HELPER_send (struct GNUNET_HELPER_Handle *h,
		    const struct GNUNET_MessageHeader *msg,
		    int can_drop,
		    GNUNET_HELPER_Continuation cont,
		    void *cont_cls)
{
  struct GNUNET_HELPER_SendHandle *sh;
  uint16_t mlen;

  if (NULL == h->fh_to_helper)
    return NULL;
  if ( (GNUNET_YES == can_drop) &&
       (NULL != h->sh_head) )
    return NULL;
  mlen = ntohs (msg->size);
  sh = GNUNET_malloc (sizeof (struct GNUNET_HELPER_SendHandle) + mlen);
  sh->msg = (const struct GNUNET_MessageHeader*) &sh[1];
  GNUNET_memcpy (&sh[1], msg, mlen);
  sh->h = h;
  sh->cont = cont;
  sh->cont_cls = cont_cls;
  GNUNET_CONTAINER_DLL_insert_tail (h->sh_head,
				    h->sh_tail,
				    sh);
  if (NULL == h->write_task)
    h->write_task = GNUNET_SCHEDULER_add_write_file (GNUNET_TIME_UNIT_FOREVER_REL,
						     h->fh_to_helper,
						     &helper_write,
						     h);

  return sh;
}

/**
 * Iterate over records matching supplied key information
 *
 * @param h handle to the PEERSTORE service
 * @param sub_system name of sub system
 * @param peer Peer identity (can be NULL)
 * @param key entry key string (can be NULL)
 * @param timeout time after which the iterate request is canceled
 * @param callback function called with each matching record, all NULL's on end
 * @param callback_cls closure for @a callback
 * @return Handle to iteration request
 */
struct GNUNET_PEERSTORE_IterateContext *
GNUNET_PEERSTORE_iterate (struct GNUNET_PEERSTORE_Handle *h,
                          const char *sub_system,
                          const struct GNUNET_PeerIdentity *peer,
                          const char *key, struct GNUNET_TIME_Relative timeout,
                          GNUNET_PEERSTORE_Processor callback,
                          void *callback_cls)
{
  struct GNUNET_MQ_Envelope *ev;
  struct GNUNET_PEERSTORE_IterateContext *ic;

  ev = PEERSTORE_create_record_mq_envelope (sub_system, peer, key, NULL, 0,
                                            NULL, 0,
                                            GNUNET_MESSAGE_TYPE_PEERSTORE_ITERATE);
  ic = GNUNET_new (struct GNUNET_PEERSTORE_IterateContext);

  ic->callback = callback;
  ic->callback_cls = callback_cls;
  ic->h = h;
  ic->sub_system = GNUNET_strdup (sub_system);
  if (NULL != peer)
    ic->peer = *peer;
  if (NULL != key)
    ic->key = GNUNET_strdup (key);
  ic->timeout = timeout;
  GNUNET_CONTAINER_DLL_insert_tail (h->iterate_head,
                                    h->iterate_tail,
                                    ic);
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Sending an iterate request for sub system `%s'\n", sub_system);
  GNUNET_MQ_send (h->mq, ev);
  ic->timeout_task =
      GNUNET_SCHEDULER_add_delayed (timeout, &iterate_timeout, ic);
  return ic;
}

/**
 * Function to call to start a peer_create type operation once all
 * queues the operation is part of declare that the
 * operation can be activated.
 *
 * @param cls the closure from GNUNET_TESTBED_operation_create_()
 */
static void
opstart_peer_create (void *cls)
{
  struct OperationContext *opc = cls;
  struct PeerCreateData *data;
  struct GNUNET_TESTBED_PeerCreateMessage *msg;
  char *config;
  char *xconfig;
  size_t c_size;
  size_t xc_size;
  uint16_t msize;

  GNUNET_assert (OP_PEER_CREATE == opc->type);
  data = opc->data;
  GNUNET_assert (NULL != data);
  GNUNET_assert (NULL != data->peer);
  opc->state = OPC_STATE_STARTED;
  config = GNUNET_CONFIGURATION_serialize (data->cfg, &c_size);
  xc_size = GNUNET_TESTBED_compress_config_ (config, c_size, &xconfig);
  GNUNET_free (config);
  msize = xc_size + sizeof (struct GNUNET_TESTBED_PeerCreateMessage);
  msg = GNUNET_realloc (xconfig, msize);
  memmove (&msg[1], msg, xc_size);
  msg->header.size = htons (msize);
  msg->header.type = htons (GNUNET_MESSAGE_TYPE_TESTBED_CREATEPEER);
  msg->operation_id = GNUNET_htonll (opc->id);
  msg->host_id = htonl (GNUNET_TESTBED_host_get_id_ (data->peer->host));
  msg->peer_id = htonl (data->peer->unique_id);
  msg->config_size = htonl (c_size);
  GNUNET_CONTAINER_DLL_insert_tail (opc->c->ocq_head, opc->c->ocq_tail, opc);
  GNUNET_TESTBED_queue_message_ (opc->c, &msg->header);
}

/**
 * Queues a message in send queue of the logger handle
 *
 * @param h the logger handle
 * @param msg the message to queue
 */
static void
queue_message (struct GNUNET_TESTBED_LOGGER_Handle *h,
               struct GNUNET_MessageHeader *msg)
{
  struct MessageQueue *mq;
  uint16_t type;
  uint16_t size;

  type = ntohs (msg->type);
  size = ntohs (msg->size);
  mq = GNUNET_new (struct MessageQueue);
  mq->msg = msg;
  LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Queueing message of type %u, size %u for sending\n", type,
       ntohs (msg->size));
  GNUNET_CONTAINER_DLL_insert_tail (h->mq_head, h->mq_tail, mq);
  if (NULL == h->th)
  {
    h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
    h->th =
        GNUNET_CLIENT_notify_transmit_ready (h->client, size,
                                             h->retry_backoff, GNUNET_YES,
                                             &transmit_ready_notify,
                                             h);
  }
}

/**
 * Add a PendingMessage to the subscriber's list of messages to be sent
 *
 * @param subscriber the subscriber to send the message to
 * @param pending_message the actual message to send
 */
static void
add_pending_subscriber_message (struct RemoteSubscriberInfo *subscriber,
                                struct PendingMessage *pending_message)
{
  GNUNET_CONTAINER_DLL_insert_tail (subscriber->pending_head,
                                    subscriber->pending_tail, pending_message);
}

/**
 * Send request to peer to start add him to to the set of experimentation nodes
 *
 * @param peer the peer to send to
 */
static void
send_experimentation_request (const struct GNUNET_PeerIdentity *peer)
{
  struct Node *n;
  struct NodeComCtx *e_ctx;
  size_t size;
  size_t c_issuers;

  c_issuers = GNUNET_CONTAINER_multihashmap_size (valid_issuers);
  size = sizeof (struct Experimentation_Request) +
    c_issuers * sizeof (struct GNUNET_CRYPTO_EddsaPublicKey);
  n = GNUNET_new (struct Node);
  n->id = *peer;
  n->timeout_task = GNUNET_SCHEDULER_add_delayed (EXP_RESPONSE_TIMEOUT, &remove_request, n);
  n->capabilities = NONE;

  e_ctx = GNUNET_new (struct NodeComCtx);
  e_ctx->n = n;
  e_ctx->e = NULL;
  e_ctx->size = size;
  e_ctx->notify = &send_experimentation_request_cb;
  e_ctx->notify_cls = n;
  GNUNET_CONTAINER_DLL_insert_tail(n->e_req_head, n->e_req_tail, e_ctx);
  schedule_transmisson (e_ctx);

  GNUNET_assert (GNUNET_OK ==
		 GNUNET_CONTAINER_multipeermap_put (nodes_requested,
						    peer, n,
						    GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
  update_stats (nodes_requested);
}

/**
 * Transmit the given message to the given target.
 *
 * @param target peer that should receive the message (must be connected)
 * @param msg message to transmit
 * @param timeout by when should the transmission be done?
 */
void
GSC_NEIGHBOURS_transmit (const struct GNUNET_PeerIdentity *target,
                         const struct GNUNET_MessageHeader *msg,
                         struct GNUNET_TIME_Relative timeout)
{
  struct NeighbourMessageEntry *me;
  struct Neighbour *n;
  size_t msize;

  n = find_neighbour (target);
  if (NULL == n)
  {
    GNUNET_break (0);
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Peer %s not found\n",
                GNUNET_i2s (target));
    return;
  }
  msize = ntohs (msg->size);
  me = GNUNET_malloc (sizeof (struct NeighbourMessageEntry) + msize);
  me->deadline = GNUNET_TIME_relative_to_absolute (timeout);
  me->size = msize;
  memcpy (&me[1],
          msg,
          msize);
  GNUNET_CONTAINER_DLL_insert_tail (n->message_head,
                                    n->message_tail,
                                    me);
  n->queue_size++;
  process_queue (n);
}

/**
 * Queues a message in send queue for sending to the service
 *
 * @param client the client to whom the queued message has to be sent
 * @param msg the message to queue
 */
void
GST_queue_message (struct GNUNET_SERVER_Client *client,
                   struct GNUNET_MessageHeader *msg)
{
  struct MessageQueue *mq_entry;
  uint16_t type;
  uint16_t size;

  type = ntohs (msg->type);
  size = ntohs (msg->size);
  GNUNET_assert ((GNUNET_MESSAGE_TYPE_TESTBED_INIT <= type) &&
                 (GNUNET_MESSAGE_TYPE_TESTBED_MAX > type));
  mq_entry = GNUNET_new (struct MessageQueue);
  mq_entry->msg = msg;
  mq_entry->client = client;
  GNUNET_SERVER_client_keep (client);
  LOG_DEBUG ("Queueing message of type %u, size %u for sending\n", type,
             ntohs (msg->size));
  GNUNET_CONTAINER_DLL_insert_tail (mq_head, mq_tail, mq_entry);
  if (NULL == transmit_handle)
    transmit_handle =
        GNUNET_SERVER_notify_transmit_ready (client, size,
                                             GNUNET_TIME_UNIT_FOREVER_REL,
                                             &transmit_ready_notify, NULL);
}

/**
 * Store a message fragment sent to a channel.
 *
 * @param h Handle for the PSYCstore.
 * @param channel_key The channel the message belongs to.
 * @param message Message to store.
 * @param psycstore_flags Flags indicating whether the PSYC message contains
 *        state modifiers.
 * @param rcb Callback to call with the result of the operation.
 * @param rcb_cls Closure for the callback.
 *
 * @return Handle that can be used to cancel the operation.
 */
struct GNUNET_PSYCSTORE_OperationHandle *
GNUNET_PSYCSTORE_fragment_store (struct GNUNET_PSYCSTORE_Handle *h,
                                 const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
                                 const struct GNUNET_MULTICAST_MessageHeader *msg,
                                 enum GNUNET_PSYCSTORE_MessageFlags psycstore_flags,
                                 GNUNET_PSYCSTORE_ResultCallback rcb,
                                 void *rcb_cls)
{
  uint16_t size = ntohs (msg->header.size);
  struct FragmentStoreRequest *req;
  struct GNUNET_PSYCSTORE_OperationHandle *
    op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + size);
  op->h = h;
  op->res_cb = rcb;
  op->cls = rcb_cls;

  req = (struct FragmentStoreRequest *) &op[1];
  op->msg = (struct GNUNET_MessageHeader *) req;
  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_FRAGMENT_STORE);
  req->header.size = htons (sizeof (*req) + size);
  req->channel_key = *channel_key;
  req->psycstore_flags = htonl (psycstore_flags);
  memcpy (&req[1], msg, size);

  op->op_id = get_next_op_id (h);
  req->op_id = GNUNET_htonll (op->op_id);

  GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
  transmit_next (h);

  return op;
}

/**
 * Retrieve all state variables for a channel with the given prefix.
 *
 * @param h
 *        Handle for the PSYCstore.
 * @param channel_key
 *        The channel we are interested in.
 * @param name_prefix
 *        Prefix of state variable names to match.
 * @param scb
 *        Callback to return matching state variables.
 * @param rcb
 *        Callback to call with the result of the operation.
 * @param cls
 *        Closure for the callbacks.
 *
 * @return Handle that can be used to cancel the operation.
 */
struct GNUNET_PSYCSTORE_OperationHandle *
GNUNET_PSYCSTORE_state_get_prefix (struct GNUNET_PSYCSTORE_Handle *h,
                                   const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
                                   const char *name_prefix,
                                   GNUNET_PSYCSTORE_StateCallback scb,
                                   GNUNET_PSYCSTORE_ResultCallback rcb,
                                   void *cls)
{
  size_t name_size = strlen (name_prefix) + 1;
  struct OperationRequest *req;
  struct GNUNET_PSYCSTORE_OperationHandle *
    op = GNUNET_malloc (sizeof (*op) + sizeof (*req) + name_size);
  op->h = h;
  op->data_cb = (DataCallback) scb;
  op->res_cb = rcb;
  op->cls = cls;

  req = (struct OperationRequest *) &op[1];
  op->msg = (struct GNUNET_MessageHeader *) req;
  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_GET_PREFIX);
  req->header.size = htons (sizeof (*req) + name_size);
  req->channel_key = *channel_key;
  memcpy (&req[1], name_prefix, name_size);

  op->op_id = get_next_op_id (h);
  req->op_id = GNUNET_htonll (op->op_id);

  GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
  transmit_next (h);

  return op;
}

/**
 * Task to register all hosts available in the global host list
 *
 * @param cls NULL
 * @param tc the scheduler task context
 */
static void
register_hosts (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct DLLOperation *dll_op;
  static unsigned int reg_host;
  unsigned int slave;

  register_hosts_task = GNUNET_SCHEDULER_NO_TASK;  
  if (reg_host == num_hosts - 1)
  {
    LOG (GNUNET_ERROR_TYPE_DEBUG,
         "All hosts successfully registered\n");
    /* Start slaves */
    state = STATE_SLAVES_STARTING;
    for (slave = 1; slave < num_hosts; slave++)
    {
      dll_op = GNUNET_malloc (sizeof (struct DLLOperation));
      dll_op->op = GNUNET_TESTBED_controller_link (dll_op,
                                                   mc,
                                                   hosts[slave],
                                                   hosts[0],
                                                   cfg,
                                                   GNUNET_YES);
      GNUNET_CONTAINER_DLL_insert_tail (dll_op_head, dll_op_tail, dll_op);
    }
    return;
  }
  reg_handle = GNUNET_TESTBED_register_host (mc, hosts[++reg_host],
                                             host_registration_completion,
                                             NULL);
}

/**
 * Apply modifiers of a message to the current channel state.
 *
 * An error is returned if there are missing messages containing state
 * operations before the current one.
 *
 * @param h
 *        Handle for the PSYCstore.
 * @param channel_key
 *        The channel we are interested in.
 * @param message_id
 *        ID of the message that contains the @a modifiers.
 * @param state_delta
 *        Value of the _state_delta PSYC header variable of the message.
 * @param rcb
 *        Callback to call with the result of the operation.
 * @param rcb_cls
 *        Closure for the @a rcb callback.
 *
 * @return Handle that can be used to cancel the operation.
 */
struct GNUNET_PSYCSTORE_OperationHandle *
GNUNET_PSYCSTORE_state_modify (struct GNUNET_PSYCSTORE_Handle *h,
                               const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
                               uint64_t message_id,
                               uint64_t state_delta,
                               GNUNET_PSYCSTORE_ResultCallback rcb,
                               void *rcb_cls)
{
  struct GNUNET_PSYCSTORE_OperationHandle *op = NULL;
  struct StateModifyRequest *req;

  op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
  op->h = h;
  op->res_cb = rcb;
  op->cls = rcb_cls;

  req = (struct StateModifyRequest *) &op[1];
  op->msg = (struct GNUNET_MessageHeader *) req;
  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_MODIFY);
  req->header.size = htons (sizeof (*req));
  req->channel_key = *channel_key;
  req->message_id = GNUNET_htonll (message_id);
  req->state_delta = GNUNET_htonll (state_delta);

  op->op_id = get_next_op_id (h);
  req->op_id = GNUNET_htonll (op->op_id);

  GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
  transmit_next (h);

  return op;
  /* FIXME: only the last operation is returned,
   *        operation_cancel() should be able to cancel all of them.
   */
}

/**
 * Update signed values of state variables in the state store.
 *
 * @param h
 *        Handle for the PSYCstore.
 * @param channel_key
 *        The channel we are interested in.
 * @param message_id
 *        Message ID that contained the state @a hash.
 * @param hash
 *        Hash of the serialized full state.
 * @param rcb
 *        Callback to call with the result of the operation.
 * @param rcb_cls
 *        Closure for the callback.
 */
struct GNUNET_PSYCSTORE_OperationHandle *
GNUNET_PSYCSTORE_state_hash_update (struct GNUNET_PSYCSTORE_Handle *h,
                                    const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
                                    uint64_t message_id,
                                    const struct GNUNET_HashCode *hash,
                                    GNUNET_PSYCSTORE_ResultCallback rcb,
                                    void *rcb_cls)
{
  struct StateHashUpdateRequest *req;
  struct GNUNET_PSYCSTORE_OperationHandle *
    op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
  op->h = h;
  op->res_cb = rcb;
  op->cls = rcb_cls;

  req = (struct StateHashUpdateRequest *) &op[1];
  op->msg = (struct GNUNET_MessageHeader *) req;
  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_STATE_RESET);
  req->header.size = htons (sizeof (*req));
  req->channel_key = *channel_key;
  req->hash = *hash;

  op->op_id = get_next_op_id (h);
  req->op_id = GNUNET_htonll (op->op_id);

  GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
  transmit_next (h);

  return op;
}

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

/**
 * Retrieve latest values of counters for a channel master.
 *
 * The current value of counters are needed when a channel master is restarted,
 * so that it can continue incrementing the counters from their last value.
 *
 * @param h
 *        Handle for the PSYCstore.
 * @param channel_key
 *        Public key that identifies the channel.
 * @param ccb
 *        Callback to call with the result.
 * @param ccb_cls
 *        Closure for the @a ccb callback.
 *
 * @return Handle that can be used to cancel the operation.
 */
struct GNUNET_PSYCSTORE_OperationHandle *
GNUNET_PSYCSTORE_counters_get (struct GNUNET_PSYCSTORE_Handle *h,
                               struct GNUNET_CRYPTO_EddsaPublicKey *channel_key,
                               GNUNET_PSYCSTORE_CountersCallback ccb,
                               void *ccb_cls)
{
  struct OperationRequest *req;
  struct GNUNET_PSYCSTORE_OperationHandle *
    op = GNUNET_malloc (sizeof (*op) + sizeof (*req));
  op->h = h;
  op->data_cb = ccb;
  op->cls = ccb_cls;

  req = (struct OperationRequest *) &op[1];
  op->msg = (struct GNUNET_MessageHeader *) req;
  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYCSTORE_COUNTERS_GET);
  req->header.size = htons (sizeof (*req));
  req->channel_key = *channel_key;

  op->op_id = get_next_op_id (h);
  req->op_id = GNUNET_htonll (op->op_id);

  GNUNET_CONTAINER_DLL_insert_tail (h->transmit_head, h->transmit_tail, op);
  transmit_next (h);

  return op;
}

/**
 * Creates a new cache entry and then puts it into the cache's hashtable.
 *
 * @param peer_id the index of the peer to tag the newly created entry
 * @return the newly created entry
 */
static struct CacheEntry *
add_entry (unsigned int peer_id)
{
  struct CacheEntry *entry;

  GNUNET_assert (NULL != cache);
  if (cache_size == GNUNET_CONTAINER_multihashmap32_size (cache))
  {
    /* remove the LRU head */
    entry = cache_head;
    GNUNET_assert (GNUNET_OK ==
                   GNUNET_CONTAINER_multihashmap32_remove (cache, (uint32_t)
                                                           entry->peer_id,
                                                           entry));
    free_entry (entry);
  }
  entry = GNUNET_new (struct CacheEntry);
  entry->peer_id = peer_id;
  GNUNET_assert (GNUNET_OK ==
                 GNUNET_CONTAINER_multihashmap32_put (cache,
                                                      (uint32_t) peer_id,
                                                      entry,
                                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST));
  GNUNET_CONTAINER_DLL_insert_tail (cache_head, cache_tail, entry);
  return entry;
}

/**
 * Try to open a connection to the given neigbour.  If the connection is open
 * already, then it is re-used.  If not, the request is queued in the operation
 * queues responsible for bounding the total number of file descriptors.  The
 * actual connection will happen when the operation queue marks the
 * corresponding operation as active.
 *
 * @param n the neighbour to open a connection to
 * @param cb the notification callback to call when the connection is opened
 * @param cb_cls the closure for the above callback
 */
struct NeighbourConnectNotification *
GST_neighbour_get_connection (struct Neighbour *n,
                              GST_NeigbourConnectNotifyCallback cb,
                              void *cb_cls)
{
  struct NeighbourConnectNotification *h;

  GNUNET_assert (NULL != cb);
  LOG_DEBUG ("Attempting to get connection to controller on host %u\n",
             n->host_id);
  h = GNUNET_new (struct NeighbourConnectNotification);
  h->n = n;
  h->cb  = cb;
  h->cb_cls = cb_cls;
  GNUNET_CONTAINER_DLL_insert_tail (n->nl_head, n->nl_tail, h);
  if (NULL == n->conn_op)
  {
    GNUNET_assert (NULL == n->controller);
    n->conn_op = GNUNET_TESTBED_operation_create_ (n, &opstart_neighbour_conn,
                                                   &oprelease_neighbour_conn);
    GNUNET_TESTBED_operation_queue_insert_ (GST_opq_openfds, n->conn_op);
    GNUNET_TESTBED_operation_begin_wait_ (n->conn_op);
    return h;
  }
  trigger_notifications (n);
  return h;
}