package com.couchbase.demo.callingstatskey;

import com.couchbase.client.CouchbaseClient;
import com.couchbase.client.CouchbaseConnectionFactoryBuilder;
import com.couchbase.client.vbucket.VBucketNodeLocator;
import java.io.IOException;
import java.net.SocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import net.spy.memcached.AddrUtil;
import net.spy.memcached.ConnectionFactoryBuilder;
import net.spy.memcached.MemcachedClient;
import net.spy.memcached.auth.AuthDescriptor;
import net.spy.memcached.auth.PlainCallbackHandler;
import net.spy.memcached.internal.OperationFuture;
import net.spy.memcached.transcoders.SerializingTranscoder;

/**
 * Get some stats from Couchbase Server for a key.
 *
 * @author ingenthr
 */
public class CallingStats {

  /**
   * A simple test file to demonstrate getting stats for a given key from
   * Couchbase Server.
   *
   * @param args the command line arguments
   * @throws URISyntaxException
   * @throws ExecutionException
   * @throws InterruptedException
   * @throws IOException
   */
  public static void main(String[] args) throws URISyntaxException, IOException, InterruptedException, ExecutionException {


    if (args.length <= 1) {
      System.err.println("usage: java -jar CallingStats.jar bucketname ipaddr [ipaddr ...]");
      System.exit(1);
    }

    String bucketname = args[0];
    String[] hostArgs = Arrays.copyOfRange(args, 1, args.length);
    System.err.println(Arrays.toString(hostArgs));

    List<URI> baselist = new ArrayList<URI>();

    for (String argument : hostArgs) {
      String oneHost = "http://" + argument + ":8091/pools";
      try {
        URI node = new URI(oneHost);
        baselist.add(node);

      } catch (URISyntaxException ex) {
        System.err.println("Could not create a URI from: " + argument);
        System.exit(1);
      }
    }
    
    SerializingTranscoder st = new SerializingTranscoder();
    st.setCompressionThreshold(Integer.MAX_VALUE);
    
    CouchbaseConnectionFactoryBuilder cfb = new CouchbaseConnectionFactoryBuilder();
    cfb.setOpTimeout(10000).setTranscoder(st);

    CouchbaseClient cbclient;
    cbclient = new CouchbaseClient(cfb.buildCouchbaseConnection(baselist, bucketname, ""));
    StringBuffer appendValue = new StringBuffer();
    for (int i = 0; i<1024; i++) {
      appendValue.append("a");
    }
    
    int sizeForArray = (1024*1024*20)-1;
    ByteBuffer bb = ByteBuffer.allocate(sizeForArray);
    for (int i=0; i<sizeForArray/4; i++) {
      bb.putInt(170);
    }
    
    AuthDescriptor myAuth = new AuthDescriptor(new String[] { "PLAIN" },
              new PlainCallbackHandler(bucketname, ""));
    MemcachedClient mbclient = new MemcachedClient(new ConnectionFactoryBuilder()
        .setProtocol(ConnectionFactoryBuilder.Protocol.BINARY)
        .setAuthDescriptor(myAuth)
        .build(), AddrUtil.getAddresses(hostArgs[0] + ":11211"));    
//    MemcachedClient mbclient = cbclient;
    
    System.out.println("Checking on append");
    
    String appendKey = "appendhere";
    mbclient.set(appendKey, 0, "0");
    for (int i=1; i<=40; i++) {
      System.out.println("Appending byte array with length: " + bb.array().length);
      OperationFuture<Boolean> appendResult = mbclient.append(0, appendKey, bb.array());
      if (!appendResult.get()) {
        System.err.println("Failed to append to value.  Status is: "
                + appendResult.getStatus().getMessage());
        System.exit(1);
      }
      
      if (i%100 == 0) {
        printKeyStats(cbclient, appendKey);
        System.out.println("Current length in kib is: " +
          ((String)mbclient.get(appendKey)).length()/1024);
        System.out.println("Appended length in kib is: " +
          appendValue.toString().length()/1024);
      }
      
    }
    
    System.out.println("appended value: " + mbclient.get(appendKey));
    System.exit(0);

    

//    // Now we'll delete it if it's in there, then set it.
//    cbclient.delete("foo").get();  // we don't care if the delete fails
//
//    System.out.println("**** Stats before set:");
//    printKeyStats(cbclient, "foo");
//
//    System.out.println("**** Stats after set:");
//    if(!cbclient.set("foo", 0, "bar").get()) {
//      System.err.println("Could not set foo.");
//      System.exit(1);
//    }
//    printKeyStats(cbclient, "foo");
//
//    // after a bit of time, we should see the key's not dirty, etc.
//    Thread.sleep(10000);
//    System.out.println("**** Stats after set and sleep:");
//    printKeyStats(cbclient, "foo");



    mbclient.shutdown(10, TimeUnit.SECONDS);

  }

  private static void printKeyStats(CouchbaseClient client, String key) {
    VBucketNodeLocator nodeLocator = (VBucketNodeLocator) client.getNodeLocator();
    int vBucketIndex = nodeLocator.getVBucketIndex(key);
    // the key stats require the vbucket on the end of the argument, so look up
    // the vbucket index    
    int vbucketIndex = vBucketIndex;

    Map<SocketAddress, Map<String, String>> stats = client.getStats("key " + key + " " + vbucketIndex);

    // since the stats is called for all nodes, just reach into the one we really care about
    SocketAddress primaryNode = nodeLocator.getPrimary(key).getSocketAddress();
    Map<String, String> statsForKey = stats.get(primaryNode);

    for (Map.Entry<String, String> thestats : statsForKey.entrySet()) {
      StringBuilder sb = new StringBuilder();
      sb.append("Stat: ").append(thestats.getKey())
              .append(" Value: ").append(thestats.getValue());
      System.out.println(sb.toString());
    }
  }
}