package com.netflix.astyanax.entitystore;

import java.lang.reflect.Field;
import java.util.Map;

import com.google.common.base.Preconditions;
import com.netflix.astyanax.ColumnListMutation;
import com.netflix.astyanax.Keyspace;
import com.netflix.astyanax.MutationBatch;
import com.netflix.astyanax.connectionpool.exceptions.ConnectionException;
import com.netflix.astyanax.model.ColumnFamily;
import com.netflix.astyanax.model.ColumnList;
import com.netflix.astyanax.model.ConsistencyLevel;
import com.netflix.astyanax.query.ColumnFamilyQuery;
import com.netflix.astyanax.query.RowQuery;

/**
 * @param <T> entity type 
 * @param <K> rowKey type
 */
public class EntityPersister<T, K> {

	//////////////////////////////////////////////////////////////////
	// Builder pattern

	public static class Builder<T, K> {

		private Class<T> clazz = null;
		private EntityAnnotation entityAnnotation = null;
		private Keyspace keyspace = null;
		private ColumnFamily<K, String> columnFamily = null;
		private ConsistencyLevel readConsitency = null;
		private ConsistencyLevel writeConsistency = null;
		private Integer ttl = null;

		public Builder() {

		}

		/**
		 * mandatory
		 * @param clazz entity class type
		 */
		public Builder<T, K> withEntity(Class<T> clazz) {
			Preconditions.checkNotNull(clazz);
			this.clazz = clazz;
			// TODO: check @Id type compatibility
			// TODO: do we need to require @Entity annotation
			this.entityAnnotation = new EntityAnnotation(clazz);
			return this;
		}

		/**
		 * mandatory
		 * @param keyspace
		 */
		public Builder<T, K> withKeyspace(Keyspace keyspace) {
			Preconditions.checkNotNull(keyspace);
			this.keyspace = keyspace;
			return this;
		}

		/**
		 * mandatory
		 * @param columnFamily column name type is fixed to String/UTF8
		 */
		public Builder<T, K> withColumnFamily(ColumnFamily<K, String> columnFamily) {
			Preconditions.checkNotNull(columnFamily);
			this.columnFamily = columnFamily;
			return this;
		}

		/**
		 * optional
		 * @param level
		 */
		public Builder<T, K> withReadConsistency(ConsistencyLevel level) {
			Preconditions.checkNotNull(level);
			this.readConsitency = level;
			return this;
		}

		/**
		 * optional
		 * @param level
		 */
		public Builder<T, K> withWriteConsistency(ConsistencyLevel level) {
			Preconditions.checkNotNull(level);
			this.writeConsistency = level;
			return this;
		}

		/**
		 * default TTL for all columns written to cassandra
		 * optional
		 * @return
		 */
		public Builder<T, K> withTTL(Integer ttl) {
			Preconditions.checkNotNull(ttl);
			this.ttl = ttl;
			return this;
		}

		public EntityPersister<T, K> build() {
			if(clazz == null)
				throw new IllegalStateException("clazz is not set");
			if(keyspace == null)
				throw new IllegalStateException("keyspace is not set");
			if(columnFamily == null)
				throw new IllegalStateException("columnFamily is not set");
			return new EntityPersister<T, K>(this);
		}
	}

	//////////////////////////////////////////////////////////////////
	// private members

	private final Class<T> clazz;
	private final EntityAnnotation entityAnnotation;
	private final Keyspace keyspace;
	private final ColumnFamily<K, String> columnFamily;
	private final ConsistencyLevel readConsitency;
	private final ConsistencyLevel writeConsistency;
	private final Integer ttl;

	private EntityPersister(Builder<T, K> builder) {
		clazz = builder.clazz;
		entityAnnotation = builder.entityAnnotation;
		keyspace = builder.keyspace;
		columnFamily = builder.columnFamily;
		readConsitency = builder.readConsitency;
		writeConsistency = builder.writeConsistency;
		ttl = builder.ttl;
	}

	private Integer getTTL(Field field) {
		// if there is no field level @TTL annotation,
		// return the default global value (which could also be valid null).
		TTL ttlAnnotation = field.getAnnotation(TTL.class);
		if(ttlAnnotation != null) {
			return ttlAnnotation.value();
		} else {
			return this.ttl;
		}
	}

	//////////////////////////////////////////////////////////////////
	// public APIs

	/**
	 * write entity to cassandra with mapped rowId and columns
	 * @param entity entity object
	 * @throws ConnectionException 
	 */
	public void put(T entity) {
		try {
			MutationBatch mb = keyspace.prepareMutationBatch();
			if(writeConsistency != null)
				mb.withConsistencyLevel(writeConsistency);

			Field idField = entityAnnotation.getId();
			@SuppressWarnings("unchecked")
			K rowKey = (K) idField.get(entity);
			ColumnListMutation<String> clm = mb.withRow(columnFamily, rowKey);

			Map<String, Field> columns = entityAnnotation.getColumns();
			for (Map.Entry<String, Field> entry : columns.entrySet()) {
				Coercions.setColumnMutationFromField(entity, entry.getValue(), entry.getKey(), clm, getTTL(entry.getValue()));
			}

			mb.execute();
		} catch(Exception e) {
			throw new RuntimeException("failed to write entity", e);
		}
	}

	/**
	 * fetch whole row and construct entity object mapping from columns
	 * @param id row key
	 * @return entity object
	 */
	public T get(K id) {
		try {
			ColumnFamilyQuery<K, String> cfq = keyspace.prepareQuery(columnFamily);
			if(readConsitency != null)
				cfq.setConsistencyLevel(readConsitency);
			RowQuery<K, String> rq = cfq.getKey(id);
			ColumnList<String> cl = rq.execute().getResult();

			T entity = clazz.newInstance();
			Field idField = entityAnnotation.getId();
			idField.set(entity, id);
			
			Map<String, Field> columns = entityAnnotation.getColumns();
			for (com.netflix.astyanax.model.Column<String> c : cl) {
				Field field = columns.get(c.getName());
	            if (field != null) {
	                Coercions.setFieldFromColumn(entity, field, c);
	            }
			}
			return entity;
		} catch(Exception e) {
			throw new RuntimeException("failed to write entity", e);
		}
	}

	/**
	 * delete the whole row
	 * @param id row key
	 */
	public void delete(K id) {
		try {
			MutationBatch mb = keyspace.prepareMutationBatch();
			ColumnListMutation<String> clm = mb.withRow(columnFamily, id);
			clm.delete();
			if(writeConsistency != null)
				mb.withConsistencyLevel(writeConsistency);
			mb.execute();
		} catch(Exception e) {
			throw new RuntimeException("failed to write entity", e);
		}
	}
}

@@ -2639,231 +2639,200 @@
 %0A%09%09%09
-if(clazz == null)%0A%09%09%09%09throw new IllegalStateException(%22clazz is not set%22);%0A%09%09%09if(keyspace == null)%0A%09%09%09%09throw new IllegalStateException(%22keyspace is not set%22);%0A%09%09%09if(columnFamily == null)%0A%09%09%09%09throw new IllegalStateException(
+// check mandatory fields%0A%09%09%09Preconditions.checkNotNull(clazz, %22clazz is not set%22);%0A%09%09%09Preconditions.checkNotNull(keyspace, %22keyspace is not set%22);%0A%09%09%09Preconditions.checkNotNull(columnFamily, 
 %22col
@@ -2855,16 +2855,35 @@
  set%22);%0A
+%09%09%09// build object%0A
 %09%09%09retur

Use guava Preconditions