package treecms.proto.cassandra.v1.test;

/*
 * Cassandra get_slice , get_range_slice test
 * 
 * 2011/01/19 Shoshi TAMAKI
 */

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

public class CassGetSliceTest
{
	public static void main(String _args[])throws Exception
	{
		Cassandra.Client client;
		
		String host = "localhost";
		int port = 9160;
		
		TTransport tr = new TSocket(host,port);
		TProtocol proto = new TBinaryProtocol(tr);
		client = new Cassandra.Client(proto);
		
		tr.open();
		
		System.out.println("ClusterName = "+client.describe_cluster_name());
		testGetRangeSlice(client);
		//testBatchInsert(client);
		
		tr.close();
	}
	
	public static void testBatchInsert(Cassandra.Client _client) throws InvalidRequestException, UnavailableException, TimedOutException, TException
	{
		List<ColumnOrSuperColumn> list = new LinkedList<ColumnOrSuperColumn>();
		
		ColumnOrSuperColumn c1 = new ColumnOrSuperColumn();
		c1.column = new Column();
		c1.column.name = "hogehoge".getBytes();
		c1.column.value = "fugafuga".getBytes();
		c1.column.timestamp = System.currentTimeMillis()/1000;
		
		ColumnOrSuperColumn c2 = new ColumnOrSuperColumn();
		c2.column = new Column();
		c2.column.name = "abc".getBytes();
		c2.column.value = "abcabc".getBytes();
		c2.column.timestamp = System.currentTimeMillis()/1000;
		
		ColumnOrSuperColumn c3 = new ColumnOrSuperColumn();
		c3.column = new Column();
		c3.column.name = "def".getBytes();
		c3.column.value = "defdef".getBytes();
		c3.column.timestamp = System.currentTimeMillis()/1000;
		
		list.add(c1);
		list.add(c2);
		list.add(c3);
		
		Map<String,List<ColumnOrSuperColumn>> map = new HashMap<String,List<ColumnOrSuperColumn>>();
		map.put("TreeCMSCF01",list);
		
		_client.batch_insert("TreeCMSKS","hoge",map,ConsistencyLevel.ONE);
		
		SliceRange sr = new SliceRange();
		sr.setStart(new byte[0]);
		sr.setFinish(new byte[0]);
		
		SlicePredicate sp = new SlicePredicate();
		sp.setSlice_range(sr);
		
		List<ColumnOrSuperColumn> result = _client.get_slice("TreeCMSKS","hoge",new ColumnParent("TreeCMSCF01"),sp,ConsistencyLevel.ONE);
		
		for(ColumnOrSuperColumn val : result){
			System.out.print("name:"+new String(val.column.name));
			System.out.println("\tvalue:"+new String(val.column.value));
		}
	}
	
	public static void testGet(Cassandra.Client _client) throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException
	{
		ColumnPath path = new ColumnPath();
		path.column_family = "TreeCMSCF01";
		path.column = "value".getBytes();
		
		ColumnOrSuperColumn value = _client.get("TreeCMSKS","uuid@01",path,ConsistencyLevel.ONE);
		
		System.out.println(new String(value.column.getValue()));
	}
	
	public static void testGetSlice(Cassandra.Client _client) throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException
	{
		SliceRange sliceRange = new SliceRange();
		sliceRange.setStart(new byte[0]);
		sliceRange.setFinish(new byte[0]);
		
		SlicePredicate slicePredicate = new SlicePredicate();
		slicePredicate.setSlice_range(sliceRange);
		
		List<ColumnOrSuperColumn> values = _client.get_slice("TreeCMSKS","",new ColumnParent("TreeCMSCF03"),slicePredicate,ConsistencyLevel.ONE);
		
		for(ColumnOrSuperColumn value : values){
			System.out.println(new String(value.column.getName())+":"+new String(value.column.getValue()));
		}
	}
	
	public static void testGetRangeSlice(Cassandra.Client _client) throws InvalidRequestException, NotFoundException, UnavailableException, TimedOutException, TException
	{
		SliceRange sliceRange = new SliceRange();
		sliceRange.setStart(new byte[0]);
		sliceRange.setFinish(new byte[0]);
		
		SlicePredicate slicePredicate = new SlicePredicate();
		slicePredicate.setSlice_range(sliceRange);
		
		List<KeySlice> values = _client.get_range_slice("TreeCMSKS",new ColumnParent("TreeCMSCF03"),slicePredicate,"","",100,ConsistencyLevel.ONE);
		
		for(KeySlice key : values){
			System.out.println(new String(key.getKey()));
			for(ColumnOrSuperColumn column : key.getColumns()){
				System.out.println("\t"+new String(column.column.getName())+":"+new String(column.column.getValue()));
			}
		}
	}
}