package me.prettyprint.cassandra.model;

import static me.prettyprint.cassandra.model.HFactory.createColumnPath;
import static me.prettyprint.cassandra.model.HFactory.createSuperColumnPath;
import static me.prettyprint.cassandra.utils.Assert.noneNull;

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

import me.prettyprint.cassandra.service.BatchMutation;
import me.prettyprint.cassandra.service.Keyspace;

import org.apache.cassandra.thrift.Deletion;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SuperColumn;


/**
 * A Mutator inserts or deltes values from the cluster.
 * There are two main ways to use a mutator:
 * 1. Use the insert/delete methods to immediately insert of delete values.
 * or 2. Use the addInsertion/addDeletion methods to schedule batch operations and then execute()
 * all of them in batch.
 *
 * The class is not thread-safe.
 *
 * @author Ran Tavory
 * @author zznate
 */
public class Mutator {

  private final KeyspaceOperator ko;

  private BatchMutation pendingMutations;

  /*package*/ Mutator(KeyspaceOperator ko) {
    this.ko = ko;
  }

  // Simple and immediate insertion of a column
  public <N,V> MutationResult insert(final String key, final String cf, final HColumn<N,V> c) {
    return new MutationResult(ko.doExecute(new KeyspaceOperationCallback<Void>() {
      @Override
      public Void doInKeyspace(Keyspace ks) throws HectorException {
        ks.insert(key, createColumnPath(cf, c.getName(), c.getNameExtractor()), c.getValueBytes());
        return null;
      }
    }));
  }

  // overloaded insert-super
  public <SN,N,V> MutationResult insert(final String key, final String cf,
      final HSuperColumn<SN,N,V> superColumn) {
    noneNull(key, cf, superColumn);
    return new MutationResult(ko.doExecute(new KeyspaceOperationCallback<Void>() {
      @Override
      public Void doInKeyspace(Keyspace ks) throws HectorException {
        Map<String, List<SuperColumn>> superColumnMap =
            new HashMap<String, List<SuperColumn>>(superColumn.getSize());
        superColumnMap.put(cf, Arrays.asList(superColumn.toThrift()));
        ks.batchInsert(key, null, superColumnMap);
        return null;
      }
    }));
  }

  public <N> MutationResult delete(final String key, final String cf, final N columnName,
      final Extractor<N> nameExtractor) {
    return new MutationResult(ko.doExecute(new KeyspaceOperationCallback<Void>() {
      @Override
      public Void doInKeyspace(Keyspace ks) throws HectorException {
        ks.remove(key, createColumnPath(cf, columnName, nameExtractor));
        return null;
      }
    }));
  }

  public <SN,N> MutationResult superDelete(final String key, final String cf, final SN supercolumnName,
      final N columnName, final Extractor<SN> sNameExtractor, final Extractor<N> nameExtractor) {
    return new MutationResult(ko.doExecute(new KeyspaceOperationCallback<Void>() {
      @Override
      public Void doInKeyspace(Keyspace ks) throws HectorException {
        ks.remove(key, createSuperColumnPath(cf, supercolumnName, columnName, sNameExtractor, nameExtractor));
        return null;
      }
    }));
  }


  // schedule an insertion to be executed in batch by the execute method
  // CAVEAT: a large number of calls with a typo in one of them will leave things in an
  // indeterminant state if we dont validate against LIVE (but cached of course)
  // keyspaces and CFs on each add/delete call
  // also, should throw a typed StatementValidationException or similar perhaps?
  public <N,V> Mutator addInsertion(String key, String cf, HColumn<N,V> c) {
    getPendingMutations().addInsertion(key, Arrays.asList(cf), c.toThrift());
    return this;
  }

  /**
   * Schedule an insertion of a supercolumn to be inserted in batch mode by {@link #execute()}
   */
  public <SN,N,V> Mutator addInsertion(String key, String cf, HSuperColumn<SN,N,V> sc) {
    getPendingMutations().addSuperInsertion(key, Arrays.asList(cf), sc.toThrift());
    return this;
  }

  public <N> Mutator addDeletion(String key, String cf, N columnName, Extractor<N> nameExtractor) {
    SlicePredicate sp = new SlicePredicate();
    sp.addToColumn_names(nameExtractor.toBytes(columnName));
    Deletion d = new Deletion(ko.createTimestamp()).setPredicate(sp);
    getPendingMutations().addDeletion(key, Arrays.asList(cf), d);
    return this;
  }

  /**
   * Batch executes all mutations scheduled to this Mutator instance by addInsertion, addDeletion etc.
   * May throw a HectorException which is a RuntimeException.
   * @return A MutationResult holds the status.
   */
  public MutationResult execute() {
    if (pendingMutations == null || pendingMutations.isEmpty()) {
      return new MutationResult(true, 0);
    }
    final BatchMutation mutations = pendingMutations.makeCopy();
    pendingMutations = null;
    return new MutationResult(ko.doExecute(new KeyspaceOperationCallback<Void>() {
      @Override
      public Void doInKeyspace(Keyspace ks) throws HectorException {
        ks.batchMutate(mutations);
        return null;
      }
    }));
  }

  /**
   * Discards all pending mutations.
   */
  public Mutator discardPendingMutations() {
    pendingMutations = null;
    return this;
  }

  @Override
  public String toString() {
    return "Mutator(" + ko.toString() + ")";
  }

  private BatchMutation getPendingMutations() {
    if (pendingMutations == null) {
      pendingMutations = new BatchMutation();
    }
    return pendingMutations;
  }

}

@@ -98,254 +98,49 @@
 eate
-ColumnPath;%0Aimport static me.prettyprint.cassandra.model.HFactory.createSuperColumnPath;%0Aimport static me.prettyprint.cassandra.utils.Assert.noneNull;%0A%0Aimport java.util.Arrays;%0Aimport java.util.HashMap;%0Aimport java.util.List;%0Aimport java.util.Map
+SuperColumnPath;%0A%0Aimport java.util.Arrays
 ;%0A%0Ai
@@ -344,56 +344,8 @@
 te;%0A
-import org.apache.cassandra.thrift.SuperColumn;%0A
 %0A%0A/*
@@ -1054,307 +1054,53 @@
     
-return new MutationResult(ko.doExecute(new KeyspaceOperationCallback%3CVoid%3E() %7B%0A      @Override%0A      public Void doInKeyspace(Keyspace ks) throws HectorException %7B%0A        ks.insert(key, createColumnPath(cf, c.getName(), c.getNameExtractor()), c.getValueBytes());%0A        return null;%0A      %7D%0A    %7D)
+addInsertion(key, cf, c);%0A    return execute(
 );%0A 
@@ -1263,16 +1263,20 @@
     
-noneNull
+addInsertion
 (key
@@ -1310,801 +1310,177 @@
 urn 
-new MutationResult(ko.doExecute(new KeyspaceOperationCallback%3CVoid%3E() %7B%0A      @Override%0A      public Void doInKeyspace(Keyspace ks) throws HectorException %7B%0A        Map%3CString, List%3CSuperColumn%3E%3E superColumnMap =%0A            new HashMap%3CString, List%3CSuperColumn%3E%3E(superColumn.getSize());%0A        superColumnMap.put(cf, Arrays.asList(superColumn.toThrift()));%0A        ks.batchInsert(key, null, superColumnMap);%0A        return null;%0A      %7D%0A    %7D));%0A  %7D%0A%0A  public %3CN%3E MutationResult delete(final String key, final String cf, final N columnName,%0A      final Extractor%3CN%3E nameExtractor) %7B%0A    return new MutationResult(ko.doExecute(new KeyspaceOperationCallback%3CVoid%3E() %7B%0A      @Override%0A      public Void doInKeyspace(Keyspace ks) throws HectorException %7B%0A        ks.remove(key, createColumnPath(
+execute();%0A  %7D%0A%0A  public %3CN%3E MutationResult delete(final String key, final String cf, final N columnName,%0A      final Extractor%3CN%3E nameExtractor) %7B%0A    addDeletion(key, 
 cf, 
@@ -1509,46 +1509,29 @@
 tor)
-)
 ;%0A    
- 
-   
 return 
-null;%0A      %7D%0A    %7D)
+execute(
 );%0A 
@@ -2058,17 +2058,16 @@
 );%0A  %7D%0A%0A
-%0A
   // sch

Use addInsertion and execute to implement insert and likewise for addDeletion