package de.fernunihagen.dna.jkn.scalephant.storage;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import de.fernunihagen.dna.jkn.scalephant.ScalephantService;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.BoundingBox;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.DeletedTuple;
import de.fernunihagen.dna.jkn.scalephant.storage.entity.Tuple;

public class Memtable implements ScalephantService, Storage {
	
	/**
	 * The name of the corresponding table
	 */
	protected final String table;
	
	/**
	 * The memtable
	 */
	protected final Tuple[] data;
	
	/**
	 * The next free position in the data array
	 */
	protected int freePos;
	
	/**
	 * Maximal number of entries keep in memory
	 */
	protected final int entries;
	
	/**
	 * Maximal size of memtable in KB
	 */
	protected final int maxsize;
	
	/**
	 * Current size in KB
	 */
	protected int currentSize;
	
	/**
	 * The lock for synchronization
	 */
	protected final Lock lock = new ReentrantLock();

	/**
	 * The Logger
	 */
	private final static Logger logger = LoggerFactory.getLogger(Memtable.class);
	
	public Memtable(final String table, final int entries, final int maxsize) {
		this.table = table;
		this.entries = entries;
		this.maxsize = maxsize;
		
		this.data = new Tuple[entries];
		freePos = -1;
		currentSize = 0;
	}

	@Override
	public void init() {
		if(freePos != -1) {
			logger.error("init() called on an initalized memtable");
			return;
		}
		
		logger.info("Initializing a new memtable for table: " + table);
		freePos = 0;
	}

	@Override
	public void shutdown() {
		
	}

	@Override
	public void put(final Tuple value) throws StorageManagerException {
		
		if(freePos >= entries) {
			throw new StorageManagerException("Unable to store a new tuple, all memtable slots are full");
		}
		
		lock.lock();
		
		try {
			data[freePos] = value;
			freePos++;
			currentSize = currentSize + value.getSize();
		} finally {
			lock.unlock();
		}
	}

	/**
	 * Get the most recent version of the tuple for key
	 * 
	 */
	@Override
	public Tuple get(final String key) {
		
		Tuple tuple = null;
		
		for(int i = 0; i < freePos; i++) {
			final Tuple possibleTuple = data[i];
			
			if(possibleTuple != null && possibleTuple.getKey().equals(key)) {
				
				if(tuple == null) {
					tuple = possibleTuple;
				} else {
					if(tuple.getTimestamp() < possibleTuple.getTimestamp()) {
						tuple = possibleTuple;
					}
				}
			}
		}
		
		return tuple;
	}
	
	/**
	 * Get all tuples that are inside of the bounding box. The result list may 
	 * contain duplicates
	 */
	@Override
	public Collection<Tuple> getTuplesInside(final BoundingBox boundingBox)
			throws StorageManagerException {
		
		final List<Tuple> resultList = new ArrayList<Tuple>();
		
		for(int i = 0; i < freePos; i++) {
			final Tuple possibleTuple = data[i];
			if(possibleTuple.getBoundingBox().overlaps(boundingBox)) {
				resultList.add(possibleTuple);
			}
		}
		
		return resultList;
	}
	
	/**
	 * Delete a tuple, this is implemented by inserting a DeletedTuple object
	 *
	 */
	@Override
	public void delete(final String key) throws StorageManagerException {
		final Tuple deleteTuple = new DeletedTuple(key);
		put(deleteTuple);
	}
	

	/**
	 * Get a sorted list with all recent tuples
	 * @return 
	 * 
	 */
	public List<Tuple> getSortedTupleList() {
		final SortedMap<String, Tuple> allTuples = new TreeMap<String, Tuple>();
		
		for(int i = 0; i < freePos; i++) {
			final String key = data[i].getKey();
			
			if(! allTuples.containsKey(key)) {
				allTuples.put(key, data[i]);
			} else {
				if(allTuples.get(key).getTimestamp() < data[i].getTimestamp()) {
					
					if(data[i] instanceof DeletedTuple) {
						allTuples.remove(key);
					} else {
						allTuples.put(key, data[i]);
					}
				}
			}
		}
		
		final List<Tuple> resultList = new ArrayList<Tuple>(allTuples.size());
		resultList.addAll(allTuples.values());
		
		return resultList;
	}
	
	/**
	 * Clean the whole memtable, useful for testing
	 * 
	 */
	@Override
	public void clear() {
		logger.info("Clear on memtable " + table + " called");
		
		for(int i = 0; i < data.length; i++) {
			data[i] = null;
		}
		
		freePos = 0;
	}
	
	/**
	 * Is this memtable full and needs to be flushed to disk
	 * 
	 * @return
	 */
	public boolean isFull() {
		
		// Check size of the table
		if(currentSize >= maxsize) {
			return true;
		}
		
		// Check number of entries
		if(freePos + 1 >= entries) {
			return true;
		}
		
		return false;
	}

	@Override
	public String getServicename() {
		return "Memtable";
	}
}

@@ -1039,20 +1039,28 @@
  int max
-s
+S
 ize
+InMemory
 ;%0A%09%0A%09/**
@@ -1072,16 +1072,23 @@
 Current 
+memory 
 size in 
@@ -1110,27 +1110,28 @@
 ted int 
-currentSize
+sizeInMemory
 ;%0A%09%0A%09/**
@@ -1391,28 +1391,36 @@
 inal int max
-s
+S
 ize
+InMemory
 ) %7B%0A%09%09this.t
@@ -1473,22 +1473,38 @@
 .max
-s
+S
 ize
+InMemory
  = max
-s
+S
 ize
+InMemory
 ;%0A%09%09
@@ -1556,27 +1556,28 @@
 = -1;%0A%09%09
-currentSize
+sizeInMemory
  = 0;%0A%09%7D
@@ -2128,33 +2128,35 @@
 %0A%09%09%09
-currentSize = currentSize
+sizeInMemory = sizeInMemory
  + v
@@ -4587,30 +4587,39 @@
 %09if(
-currentSize
+sizeInMemory
  %3E= max
-s
+S
 ize
+InMemory
 ) %7B%0A
@@ -4688,17 +4688,16 @@
 os + 1 %3E
-=
  entries

Use the last slot of the memtable