/**
 * Callback indicating that the hostkey was created for a peer.
 *
 * @param cls NULL
 * @param id the peer identity
 * @param d the daemon handle (pretty useless at this point, remove?)
 * @param emsg non-null on failure
 */
void
hostkey_callback (void *cls, const struct GNUNET_PeerIdentity *id,
                  struct GNUNET_TESTING_Daemon *d, const char *emsg)
{
  if (emsg != NULL)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
                "Hostkey callback received error: %s\n", emsg);
  }

#if VERBOSE > 1
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Hostkey (%d/%d) created for peer `%s'\n", num_peers - peers_left,
              num_peers, GNUNET_i2s (id));
#endif

#if VERBOSE
  if ((num_peers - peers_left) % modnum == 0)
  {
    if (num_peers - peers_left == 0)
      FPRINTF (stdout, "%s",  "0%%");
    else
      FPRINTF (stdout, "%d%%",
               (int) (((float) (num_peers - peers_left) / num_peers) * 100));

  }
  else if ((num_peers - peers_left) % dotnum == 0)
  {
    FPRINTF (stdout, "%s",  ".");
  }
  fflush (stdout);
#endif
  peers_left--;
  if (peers_left == 0)
  {
#if VERBOSE
    FPRINTF (stdout, "%s",  "100%%]\n");
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "All %d hostkeys created, now creating topology!\n", num_peers);
#endif
    GNUNET_SCHEDULER_cancel (die_task);
    /* Set up task in case topology creation doesn't finish
     * within a reasonable amount of time */
    die_task =
        GNUNET_SCHEDULER_add_delayed (test_timeout, &end_badly,
                                      "from create_topology");
    GNUNET_SCHEDULER_add_now (&create_topology, NULL);
    ok = 0;
  }
}

static void
testing_connect_cb (struct PeerContext *p1, struct PeerContext *p2, void *cls)
{
  cc = NULL;
  char *p1_c = GNUNET_strdup (GNUNET_i2s (&p1->id));

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peers connected: %u (%s) <-> %u (%s)\n",
              p1->no, p1_c, p2->no, GNUNET_i2s (&p2->id));
  GNUNET_free (p1_c);



  send_task = GNUNET_SCHEDULER_add_now (&sendtask, NULL);
}

static void
add_peer_done (void *cls, const char *emsg)
{
  if (NULL == emsg)
  {
    return;
  }
  else
  {
    GNUNET_break(0);
    GNUNET_SCHEDULER_cancel (timeout_task);
    timeout_task = GNUNET_SCHEDULER_add_now (&end_badly, NULL );
  }
}

/**
 * Function executed after stream has been established
 *
 * @param cls the closure from GNUNET_STREAM_open
 * @param socket socket to use to communicate with the other side (read/write)
 */
static void 
stream_open_cb (void *cls,
                struct GNUNET_STREAM_Socket *socket)
{
  struct PeerData *peer=cls;

  GNUNET_assert (&peer1 == peer);
  GNUNET_assert (socket == peer1.socket);
  GNUNET_assert (socket == peer->socket);

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream established from peer1\n");
  peer->bytes_wrote = 0;
  GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
}

/**
 * Stats callback. Finish the stats testbed operation and when all stats have
 * been iterated, shutdown the test.
 *
 * @param cls Closure (line number from which termination was requested).
 * @param op the operation that has been finished
 * @param emsg error message in case the operation has failed; will be NULL if
 *          operation has executed successfully.
 */
static void
stats_cont (void *cls, struct GNUNET_TESTBED_Operation *op, const char *emsg)
{
  GNUNET_log (GNUNET_ERROR_TYPE_INFO, " KA sent: %u, KA received: %u\n",
              ka_sent, ka_received);
  if (KEEPALIVE == test && (ka_sent < 2 || ka_sent > ka_received + 1))
    ok--;
  GNUNET_TESTBED_operation_done (stats_op);

  if (NULL != disconnect_task)
    GNUNET_SCHEDULER_cancel (disconnect_task);
  disconnect_task = GNUNET_SCHEDULER_add_now (&disconnect_cadet_peers,
					      cls);
}

static void
notify_receive (void *cls, const struct GNUNET_PeerIdentity *peer,
                const struct GNUNET_MessageHeader *message,
                const struct GNUNET_ATS_Information *ats, uint32_t ats_count)
{
  struct PeerContext *p = cls;
  struct PeerContext *t = NULL;

  if (0 == memcmp (peer, &p1->id, sizeof (struct GNUNET_PeerIdentity)))
    t = p1;
  if (0 == memcmp (peer, &p2->id, sizeof (struct GNUNET_PeerIdentity)))
    t = p2;
  GNUNET_assert (t != NULL);

  char *ps = GNUNET_strdup (GNUNET_i2s (&p->id));

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Peer %u (`%4s') received message of type %d and size %u size from peer %u (`%4s')!\n",
              p->no, ps, ntohs (message->type), ntohs (message->size), t->no,
              GNUNET_i2s (&t->id));
  GNUNET_free (ps);

  if ((MTYPE == ntohs (message->type)) &&
      (sizeof (struct GNUNET_MessageHeader) == ntohs (message->size)))
  {
    if (restarted == GNUNET_NO)
    {
      restarted = GNUNET_YES;
      restart (p1, cfg_file_p1);
      restart (p2, cfg_file_p2);
      return;
    }
    else
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Restarted peers connected, stopping test...\n");
      ok = 0;
      end ();
    }
  }
  else
  {
    GNUNET_break (0);
    ok = 1;
    if (die_task != GNUNET_SCHEDULER_NO_TASK)
      GNUNET_SCHEDULER_cancel (die_task);
    die_task = GNUNET_SCHEDULER_add_now (&end_badly, NULL);
  }
}

static void
on_lookup_result(void *cls, uint32_t rd_count,
                 const struct GNUNET_NAMESTORE_RecordData *rd)
{
  struct in_addr a;
  int i;
  char* addr;
  
  if (GNUNET_SCHEDULER_NO_TASK != die_task)
  {
      GNUNET_SCHEDULER_cancel (die_task);
      die_task = GNUNET_SCHEDULER_NO_TASK;
  }

  GNUNET_NAMESTORE_disconnect (namestore_handle);
  namestore_handle = NULL;
  if (rd_count == 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Lookup failed, this is good!\n");
    ok = 0;
  }
  else
  {
    ok = 1;
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "name: %s\n", (char*)cls);
    for (i=0; i<rd_count; i++)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "type: %d\n", rd[i].record_type);
      if (rd[i].record_type == GNUNET_GNS_RECORD_A)
      {
        memcpy(&a, rd[i].data, sizeof(a));
        addr = inet_ntoa(a);
        GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "address: %s\n", addr);
        if (0 == strcmp(addr, TEST_IP))
        {
          GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                    "%s incorrectly resolved to %s!\n", TEST_DOMAIN, addr);
          ok = 2;
        }
      }
      else
      {
        GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "No resolution!\n");
      }
    }
  }
  GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);
}

void ats_perf_cb (void *cls,
                  const struct
                  GNUNET_HELLO_Address *
                  address,
                  struct
                  GNUNET_BANDWIDTH_Value32NBO
                  bandwidth_out,
                  struct
                  GNUNET_BANDWIDTH_Value32NBO
                  bandwidth_in,
                  const struct
                  GNUNET_ATS_Information *
                  ats, uint32_t ats_count)
{
  struct PendingResolutions * pr;


  if (NULL != address)
  {
    pr = GNUNET_malloc (sizeof (struct PendingResolutions) +
                        ats_count * sizeof (struct GNUNET_ATS_Information));

    pr->ats_count = ats_count;
    pr->ats = (struct GNUNET_ATS_Information *) &pr[1];
    if (ats_count > 0)
      memcpy (pr->ats, ats, ats_count * sizeof (struct GNUNET_ATS_Information));
    pr->address = GNUNET_HELLO_address_copy (address);
    pr->bandwidth_in = bandwidth_in;
    pr->bandwidth_out = bandwidth_out;
    pr->tats_ctx = GNUNET_TRANSPORT_address_to_string(cfg, address,
                      resolve_addresses_numeric, GNUNET_TIME_UNIT_FOREVER_REL, transport_addr_to_str_cb, pr);
    GNUNET_CONTAINER_DLL_insert (head, tail, pr);
    results++;
    pending++;
  }
  else
  {
    /* All messages received */
    receive_done = GNUNET_YES;
    alh = NULL;
    if (0 == pending)
    {
      /* All messages received and no resolutions pending*/
      if (end_task != GNUNET_SCHEDULER_NO_TASK)
        GNUNET_SCHEDULER_cancel (end_task);
      end_task = GNUNET_SCHEDULER_add_now (end, NULL);
    }
  }
}

static void
on_lookup_result(void *cls, uint32_t rd_count,
                 const struct GNUNET_NAMESTORE_RecordData *rd)
{
  int i;
  uint16_t *srv_data;
  char* srv;
  
  if (GNUNET_SCHEDULER_NO_TASK != die_task)
  {
      GNUNET_SCHEDULER_cancel (die_task);
      die_task = GNUNET_SCHEDULER_NO_TASK;
  }

  GNUNET_NAMESTORE_disconnect (namestore_handle);
  if (rd_count == 0)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
                "Lookup failed, rp_filtering?\n");
    ok = 2;
  }
  else
  {
    ok = 1;
    GNUNET_log (GNUNET_ERROR_TYPE_INFO, "name: %s\n", (char*)cls);
    for (i=0; i<rd_count; i++)
    {
      GNUNET_log (GNUNET_ERROR_TYPE_INFO, "type: %d\n", rd[i].record_type);
      if (rd[i].record_type == GNUNET_GNS_RECORD_SRV)
      {
        srv_data = (uint16_t*)rd[i].data;
        srv = (char*)&srv_data[3];
        GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                    "Got SRV %s with p=%d,w=%d,port=%d\n",
                    srv, srv_data, &srv_data[1], &srv_data[2]);
        if (0 == strcmp(srv, TEST_EXPECTED_SRV))
        {
          GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                      "%s correctly resolved to %s!\n", TEST_DOMAIN,
                      TEST_EXPECTED_SRV);
          ok = 0;
        }
      }
    }
  }

  GNUNET_SCHEDULER_add_now (&shutdown_task, NULL);

}

static void
send_test_messages (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct TestMessageContext *pos = cls;

  if ((pos == test_messages) && (settle_time.rel_value > 0))
  {
    topology_connections = 0;
    GNUNET_TESTING_get_topology (pg, &topology_cb, NULL);
  }
  if (((tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN) != 0) || (cls == NULL))
    return;

  if (die_task == GNUNET_SCHEDULER_NO_TASK)
  {
    die_task =
        GNUNET_SCHEDULER_add_delayed (TEST_TIMEOUT, &end_badly,
                                      "from send test messages (timeout)");
  }

  if (total_server_connections >= MAX_OUTSTANDING_CONNECTIONS)
  {
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                  (GNUNET_TIME_UNIT_SECONDS, 1),
                                  &send_test_messages, pos);
    return;                     /* Otherwise we'll double schedule messages here! */
  }

  /*
   * Connect to the sending peer
   */
  pos->peer1handle =
      GNUNET_CORE_connect (pos->peer1->cfg, pos, &init_notify_peer1,
                           &connect_notify_peers, NULL, NULL, GNUNET_NO, NULL,
                           GNUNET_NO, no_handlers);

  GNUNET_assert (pos->peer1handle != NULL);

  if (total_server_connections < MAX_OUTSTANDING_CONNECTIONS)
  {
    GNUNET_SCHEDULER_add_now (&send_test_messages, pos->next);
  }
  else
  {
    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
                                  (GNUNET_TIME_UNIT_SECONDS, 1),
                                  &send_test_messages, pos->next);
  }
}

/**
 * Main point of test execution
 */
static void
run (void *cls, char *const *args, const char *cfgfile,
     const struct GNUNET_CONFIGURATION_Handle *cfg)
{
  struct TestingContext *test_ctx;
  char *emsg;
  struct GNUNET_PeerIdentity id;
  struct GNUNET_TESTING_SharedService ss[] = {
    {"peerinfo", cfg, 2},
    {NULL, NULL, 0}
  };
  struct GNUNET_TESTING_Peer *peer;
  unsigned int cnt;

  test_ctx = GNUNET_new (struct TestingContext);
  test_ctx->system =
      GNUNET_TESTING_system_create ("test-gnunet-testing",
                                    "127.0.0.1", NULL, ss);
  emsg = NULL;
  if (NULL == test_ctx->system)
    goto end;
  test_ctx->cfg = GNUNET_CONFIGURATION_dup (cfg);
  for (cnt = 0; cnt < NUM_PEERS; cnt++)
  {
    peer = GNUNET_TESTING_peer_configure (test_ctx->system,
                                          test_ctx->cfg,
                                          0, &id, &emsg);
    if (NULL == peer)
    {
      if (NULL != emsg)
        printf ("Test failed upon error: %s", emsg);
      goto end;
    }
    if (GNUNET_OK != GNUNET_TESTING_peer_start (peer))
    {
      GNUNET_TESTING_peer_destroy (peer);
      goto end;
    }
    test_ctx->peers[cnt] = peer;
  }
  status = GNUNET_OK;
  GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_SECONDS,
                                &do_shutdown, test_ctx);
  return;

 end:
  GNUNET_SCHEDULER_add_now (&do_shutdown, test_ctx);
  GNUNET_free_non_null (emsg);
}

static void
put_cont (void *cls, int32_t success, const char *emsg)
{
  const char *name = cls;

  nsqe = NULL;
  GNUNET_assert (NULL != cls);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
	      "Name store added record for `%s': %s\n",
	      name,
	      (success == GNUNET_OK) ? "SUCCESS" : "FAIL");
  GNUNET_SCHEDULER_cancel (endbadly_task);
  endbadly_task = NULL;
  GNUNET_SCHEDULER_add_now (&end, NULL);
}

static void
test_get_iterator (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)
{
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "test_get_iterator called (we got a result), stopping get request!\n");
  GNUNET_SCHEDULER_add_now (&test_get_stop,
                            NULL);
}

static void
handle_test (void *cls,
	     const struct GNUNET_MessageHeader *message)
{
  const struct GNUNET_PeerIdentity *peer = cls;

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Receiving message from `%s'.\n",
              GNUNET_i2s (peer));
  GNUNET_assert (ok == 5);
  OKPP;
  GNUNET_SCHEDULER_cancel (err_task);
  err_task = GNUNET_SCHEDULER_add_now (&terminate_task,
				       NULL);
}

/**
 * Callback which will be called to after a host registration succeeded or failed
 *
 * @param cls the closure
 * @param emsg the error message; NULL if host registration is successful
 */
static void
host_registration_completion (void *cls, const char *emsg)
{
  struct GNUNET_TESTBED_RunHandle *rc = cls;

  rc->reg_handle = NULL;
  if (NULL != emsg)
  {
    LOG (GNUNET_ERROR_TYPE_WARNING,
         _("Host registration failed for a host. Error: %s\n"), emsg);
    GNUNET_SCHEDULER_shutdown ();
    return;
  }
  rc->register_hosts_task = GNUNET_SCHEDULER_add_now (&register_hosts, rc);
}

/**
 * Iterator callback to go over all addresses and try to validate them
 * (unless blocked or already validated).
 *
 * @param cls pointer to a 'struct ValidateAddressContext'
 * @param address the address
 * @param expiration expiration time
 * @return GNUNET_OK (keep the address)
 */
static int
validate_address_iterator (void *cls,
                           const struct GNUNET_HELLO_Address *address,
                           struct GNUNET_TIME_Absolute expiration)
{
  const struct ValidateAddressContext *vac = cls;
  struct ValidationEntry *ve;

  if (GNUNET_TIME_absolute_get_remaining (expiration).rel_value == 0)
    return GNUNET_OK;           /* expired */
  ve = find_validation_entry (&vac->public_key, address);
  if (GNUNET_SCHEDULER_NO_TASK == ve->revalidation_task)
    ve->revalidation_task = GNUNET_SCHEDULER_add_now (&revalidate_address, ve);
  return GNUNET_OK;
}

static void
address_suggest_cb (void *cls,
                    const struct GNUNET_PeerIdentity *peer,
                    const struct GNUNET_HELLO_Address *address,
                    struct Session *session,
                    struct GNUNET_BANDWIDTH_Value32NBO bandwidth_out,
                    struct GNUNET_BANDWIDTH_Value32NBO bandwidth_in,
                    const struct GNUNET_ATS_Information *atsi,
                    uint32_t ats_count)

{
  GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Did not expect suggestion callback!\n");
  GNUNET_SCHEDULER_add_now (&end_badly, NULL);
  return;
}

static void
read_call (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
{
  struct GNUNET_DISK_FileHandle *stdout_read_handle = cls;
  char buf[16];

  memset (&buf, 0, sizeof (buf));
  int bytes;

  bytes = GNUNET_DISK_file_read (stdout_read_handle, &buf, sizeof (buf));

  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "bytes is %d\n", bytes);

  if (bytes < 1)
  {
    GNUNET_break (0);
    ok = 1;
    GNUNET_SCHEDULER_cancel (die_task);
    GNUNET_SCHEDULER_add_now (&end_task, NULL);
    return;
  }

  ok = strncmp (&buf[0], test_phrase, strlen (test_phrase));
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "read %s\n", &buf[0]);
  if (0 == ok)
  {
    GNUNET_SCHEDULER_cancel (die_task);
    GNUNET_SCHEDULER_add_now (&end_task, NULL);
    return;
  }

  GNUNET_SCHEDULER_add_read_file (GNUNET_TIME_UNIT_FOREVER_REL,
                                  stdout_read_handle, &read_call,
                                  stdout_read_handle);

}

/**
 * Functions of this type are called upon new stream connection from other peers
 *
 * @param cls the closure from GNUNET_STREAM_listen
 * @param socket the socket representing the stream
 * @param initiator the identity of the peer who wants to establish a stream
 *            with us
 * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 *             stream (the socket will be invalid after the call)
 */
static int
stream_listen_cb (void *cls, struct GNUNET_STREAM_Socket *socket,
		  const struct GNUNET_PeerIdentity *initiator)
{
  struct PeerData *pdata = cls;

  
  if ((NULL == socket) || (NULL == initiator))
  {
    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Binding error\n");
    if (GNUNET_SCHEDULER_NO_TASK != abort_task)
      GNUNET_SCHEDULER_cancel (abort_task);
    abort_task = GNUNET_SCHEDULER_add_now (&do_abort, NULL);
    return GNUNET_OK;
  }
  GNUNET_assert (NULL != socket);
  GNUNET_assert (pdata == &peer_data[1]);
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer connected: %s\n",
	      GNUNET_i2s(initiator));
  pdata->socket = socket;
  pdata->bytes_read = 0;
  read_task = GNUNET_SCHEDULER_add_now (&stream_read_task, pdata);
  return GNUNET_OK;
}

/**
 * Connect to the network size estimation service.
 *
 * @param cfg the configuration to use
 * @param func funtion to call with network size estimate
 * @param func_cls closure to pass for network size estimate callback
 *
 * @return handle to use
 */
struct GNUNET_NSE_Handle *
GNUNET_NSE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
                    GNUNET_NSE_Callback func, void *func_cls)
{
    struct GNUNET_NSE_Handle *ret;

    GNUNET_assert (func != NULL);
    ret = GNUNET_malloc (sizeof (struct GNUNET_NSE_Handle));
    ret->cfg = cfg;
    ret->recv_cb = func;
    ret->recv_cb_cls = func_cls;
    ret->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
    ret->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, ret);
    return ret;
}