package treecms.proto.cassandra.v1;

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.ConsistencyLevel;
import org.apache.cassandra.thrift.Mutation;

import treecms.proto.api.Node;

public class CassTreeImporter
{
	public CassBrowser m_browser;
	
	public CassTreeImporter(CassBrowser _browser)
	{
		m_browser = _browser;
	}
	
	public void importTree(Node _root)
	{
		List<Node> list = listNode(_root);
		
		Cassandra.Client client = m_browser.getClient();
		
		Map<String,Map<String,List<Mutation>>> mutation = new HashMap<String,Map<String,List<Mutation>>>();
		Map<String,List<Mutation>> map = new HashMap<String,List<Mutation>>();
		for(Node node : list){
			LinkedList<Mutation> columns = new LinkedList<Mutation>();
			
			Column className = new Column();
			className.name = CassBrowser.CLASSNAME.getBytes();
			className.value = node.getClassName().getBytes();
			columns.add(packColumn(className));
			
			Column title = new Column();
			title.name = CassBrowser.TITLE.getBytes();
			title.value = node.getTitle().getBytes();
			columns.add(packColumn(title));
			
			for(String key : node.getAttributeKeys()){
				Column attr = new Column();
				attr.name = key.getBytes();
				attr.value = node.getAttribute(key).getBytes();
				columns.add(packColumn(attr));
			}
			
			map.put(node.getID().toString(),columns);
		}
		
		mutation.put(m_browser.getColumnFamily(),map);
		
		try{
			client.batch_mutate(m_browser.getKeySpace(),mutation,ConsistencyLevel.ONE);
		}catch(Exception _e){
			_e.printStackTrace();
		}
	}
	
	public Mutation packColumn(Column _col)
	{
		ColumnOrSuperColumn cors = new ColumnOrSuperColumn();
		
		cors.column = _col;

		Mutation mut = new Mutation();
		mut.column_or_supercolumn = cors;
		
		return mut;
	}
	
	private List<Node> listNode(Node _root)
	{
		LinkedList<Node> list = new LinkedList<Node>();
		list.addAll(_root.getChildren());
		for(Node child : list){
			list.addAll(listNode(child));
		}
		return list;
	}
}