#include <codesloop/common/test_timer.h>
#include <codesloop/common/logger.hh>
#include <codesloop/common/common.h>

#include <keyspace_client.h>

#include <assert.h>
#include <vector>


using namespace csl::common;

namespace test_keyspace
{
  keyspace_client_t client;

  char * value_8k;
  size_t value_8k_length= 8191;
  int sequence = 0;
  char key[256];
  pid_t pidprefix = getpid() % 10;

  void init()
  {
    unsigned int x;
    value_8k = new char[value_8k_length];

    for (x= 0; x < value_8k_length; x++)
      value_8k[x] = char (x % 127);

    memset(key, 0, sizeof(key));
  }

  inline void insert_key()
  {
    snprintf( key, 255, "%d%10d", pidprefix, sequence++ % 500000 );

    keyspace_client_set(client, key, strlen(key),
        value_8k, value_8k_length);
  }

  inline void read_key()
  {
    static int read_seq = 0;

    if ( read_seq == sequence )
      read_seq = 0;

    snprintf( key, 255, "%d%10d", pidprefix, read_seq++ % 500000 );

    // safe read
    keyspace_client_get_simple(client, key, strlen(key), value_8k, value_8k_length, 0);
  }


  inline void bulk_operations()
  {
    static const char * keys[] = {
      "key01",
      "key02",
      "key03",
      "key04",
      "key05",
      "key06",
      "key07",
      "key08",
      "key09",
      "key10",
    };
    static const size_t key_sizes[] = {
      5,5,5,5,5,5,5,5,5,5
    };


    // bulk set
    keyspace_client_begin(client);

    for ( sequence = 0; sequence < 10 ; sequence++ )
    {
      keyspace_client_set(client, keys[sequence], key_sizes[sequence],
          value_8k, value_8k_length);
    }
    keyspace_client_submit(client);

    // bulk gets 10 key/vals 10 times
    for ( int i = 0; i < 10 ; i++ )
    {
      // get from server
      keyspace_client_begin(client);
      for ( sequence = 0; sequence < 10; sequence++ )
      {
        keyspace_client_get_simple(client, keys[sequence], key_sizes[sequence],
            value_8k, value_8k_length, 0);
      }
      keyspace_client_submit(client);

      // fetch result
      keyspace_result_t result = keyspace_client_result(client);
      for (keyspace_result_begin(result);
          !keyspace_result_is_end(result);
          keyspace_result_next(result))
      {
        const void * rkey;
        const void * rval;
        unsigned keylen, vallen;
        if (keyspace_result_key(result, &rkey,
              &keylen) != KEYSPACE_SUCCESS ||
            keyspace_result_value(result, &rval,
              &vallen) != KEYSPACE_SUCCESS)
        {
          fprintf(stderr, "result failed");
        }
      } // iterate on result

      keyspace_result_close(result);
    } // for 10 multiget

  } // bulk_operations()

};


using namespace test_keyspace;


int main()
{
  client = keyspace_client_create();

  test_keyspace::init();

  const char * nodes[] = {
    "clust-01:7080",
    "clust-02:7080"
  };

  int status = keyspace_client_init(client, 2, nodes);
  if (status != KEYSPACE_SUCCESS)
  {
    fprintf(stderr, "Can not init client\n");
    exit(-1);
  }

  printf("Connected\n");
  keyspace_client_prune(client, "", 0 );
  printf("Prune done.\n");

  csl_common_print_results( "keyspace insert key        ", csl_common_test_timer_v0(insert_key),"" );
  csl_common_print_results( "keyspace read key          ", csl_common_test_timer_v0(read_key),"" );
  keyspace_client_prune(client, "", 0 );
  csl_common_print_results( "keyspace bulk insert/read  ", csl_common_test_timer_v0(bulk_operations),"" );

  keyspace_client_destroy(client);

  delete value_8k;
}