import com.gemstone.gemfire.cache.CacheFactory;
import com.gemstone.gemfire.cache.Declarable;
import com.gemstone.gemfire.cache.Region;

import com.gemstone.gemfire.cache.execute.Function;
import com.gemstone.gemfire.cache.execute.FunctionContext;
import com.gemstone.gemfire.cache.execute.RegionFunctionContext;

import com.gemstone.gemfire.cache.partition.PartitionRegionHelper;

import com.gemstone.gemfire.internal.cache.CachedDeserializable;
import com.gemstone.gemfire.internal.cache.EntrySnapshot;
import com.gemstone.gemfire.internal.cache.LocalRegion;
import com.gemstone.gemfire.internal.cache.RegionEntry;

import java.util.Iterator;
import java.util.Properties;
import java.util.Set;

public class GetValueBatchesFunction implements Function, Declarable {
  
  public void execute(FunctionContext context) {
    RegionFunctionContext rfc = (RegionFunctionContext) context;
    Region primaryDataSet = PartitionRegionHelper.getLocalDataForContext(rfc);
    ValueBatch batch = new ValueBatch();
    Set entries = primaryDataSet.entrySet();
    System.out.println(Thread.currentThread().getName() + " executing " + getId() + " for " + entries.size() + " primary entries");
    for (Iterator i = entries.iterator(); i.hasNext();) {
      EntrySnapshot entry = (EntrySnapshot) i.next();
      RegionEntry re = entry.getRegionEntry();
      Object valueInVm = re.getValue(null);
      batch.addValue(valueInVm);
      if (batch.isFull()) {
        context.getResultSender().sendResult(batch);
        batch.clear();
      }
    }
    context.getResultSender().lastResult(batch);
  }
  
  public String getId() {
    return getClass().getSimpleName();
  }

  public boolean optimizeForWrite() {
    return true;
  }

  public boolean hasResult() {
    return true;
  }

  public boolean isHA() {
    return true;
  }

  public void init(Properties properties) {
  }
}