package jp.ac.u_ryukyu.ie.cr.bbs.local.bbs;

import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.serializers.UUIDSerializer;
import me.prettyprint.cassandra.service.template.SuperCfUpdater;
import me.prettyprint.cassandra.service.template.ThriftSuperCfTemplate;
import me.prettyprint.cassandra.utils.TimeUUIDUtils;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.*;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.RangeSuperSlicesQuery;
import me.prettyprint.hector.api.query.SuperSliceQuery;

import java.util.List;
import java.util.UUID;

public class HectorSample
{
	public static void main(String _args[])
	{
		Cluster myCluster = HFactory.getOrCreateCluster("test-cluster","localhost:9160");

		Keyspace ksp = HFactory.createKeyspace("cassaBBS",myCluster);
		/*
		ColumnFamilyDefinition newCF = HFactory.createColumnFamilyDefinition("DEMO","ccc",ComparatorType.UUIDTYPE);
		newCF.setColumnType(ColumnType.SUPER);
		myCluster.addColumnFamily(newCF);
		*/

		ThriftSuperCfTemplate<String,UUID,String> template =
				new ThriftSuperCfTemplate<String,UUID,String>(ksp,"boards",StringSerializer.get(),
						UUIDSerializer.get(),StringSerializer.get());
		UUID time = TimeUUIDUtils.getUniqueTimeUUIDinMillis();
		SuperCfUpdater<String,UUID,String> updater = template.createUpdater("board1",TimeUUIDUtils.getTimeUUID(0));
		updater.setString("name","peter");
		updater.setString("message",time.toString());

		template.update(updater);

		UUID start = TimeUUIDUtils.getTimeUUID(0);

		SuperSliceQuery<String, UUID, String, String> sq = HFactory.createSuperSliceQuery(ksp, StringSerializer.get(), UUIDSerializer.get(), StringSerializer.get(), StringSerializer.get());
		sq.setKey("board1").setColumnFamily("ccc").setRange(start,null,false,100);

		QueryResult<SuperSlice<UUID,String,String>> results = sq.execute();

		SuperSlice<UUID,String,String> ss = results.get();
		List<HSuperColumn<UUID,String,String>> list = ss.getSuperColumns();
		for(HSuperColumn<UUID,String,String> sc : list){
			HColumn<String,String> sub = sc.getSubColumnByName("name");
			System.out.println(sub.getValue());
			sub = sc.getSubColumnByName("message");
			System.out.println(sub.getValue());
		}

		RangeSuperSlicesQuery<String,UUID,String,String> rsq = HFactory.createRangeSuperSlicesQuery(ksp,StringSerializer.get(),
				UUIDSerializer.get(),StringSerializer.get(),StringSerializer.get());
		rsq.setKeys(null,null).setRange(null,null,false,0).setColumnFamily("ccc");

		QueryResult<OrderedSuperRows<String,UUID,String,String>> rsqResult = rsq.execute();
		OrderedSuperRows<String, UUID, String, String> rows = rsqResult.get();
		for(SuperRow<String, UUID, String, String>  row : rows.getList()){
			System.out.println(row.getKey());
		}
	}
}