package com.hammurapi.store.local;

import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.Predicate;
import com.hammurapi.store.AbstractIndex;
import com.hammurapi.store.Store;
import com.hammurapi.store.Store.Handle;
import com.hammurapi.store.StoreException;

class LocalIndex<ST, T extends ST, PK, V, S extends Store<ST,PK,S>> extends AbstractIndex<ST, T, PK, V, S> {

	private Map<V, Handle<ST, PK, S>> uniqueStore;
	private Map<V, Collection<Handle<ST, PK, S>>> multiStore;
	private ReadWriteLock lock = new ReentrantReadWriteLock();

	protected LocalIndex(
			Predicate<ST, S> predicate,
			Extractor<T, V, S> extractor,
			Type type, 
			boolean ordered,
			Comparator<V> comparator, 
			LocalStoreBase<ST, PK, S> store) {
		
		super(predicate, extractor, type, ordered, comparator, store);
		
		if (Type.UNIQUE.equals(type)) {
			if (ordered) {
				if (comparator==null) {
					uniqueStore = new TreeMap<V, Handle<ST, PK, S>>();
				} else {
					uniqueStore = new TreeMap<V, Handle<ST, PK, S>>(comparator);
				}
			} else {
				uniqueStore = new HashMap<V, Handle<ST, PK, S>>();
			}
		} else {
			if (ordered) {
				if (comparator==null) {
					multiStore = new TreeMap<V, Collection<Handle<ST, PK, S>>>();
				} else {
					multiStore = new TreeMap<V, Collection<Handle<ST, PK, S>>>(comparator);
				}
			} else {
				multiStore = new HashMap<V, Collection<Handle<ST, PK, S>>>();				
			}
		}
	}		

	@Override
	protected Map<V, Handle<ST, PK, S>> getIndexUniqueStore() {
		return uniqueStore;
	}

	@Override
	protected Map<V, Collection<Handle<ST, PK, S>>> getIndexMultiStore() {
		return multiStore;
	}

	@Override
	protected Lock readLock() {
		return lock.readLock();
	}

	@Override
	protected Lock writeLock() {
		return lock.writeLock();
	}

	@Override
	public void upgrade(Type type, boolean ordered, Comparator<V> comparator) {
		if (!Type.UNIQUE.equals(this.type) && Type.UNIQUE.equals(type)) {
			// Check uniqueness
			for (Collection<Handle<ST, PK, S>> h: multiStore.values()) {
				if (h.size()>1) {
					throw new StoreException("Cannot upgrade index to unique, there are duplicate values");
				}
			}
			
			if (ordered || this.ordered) {
				if (comparator==null) {
					uniqueStore = new TreeMap<V, Handle<ST, PK, S>>();
				} else {
					uniqueStore = new TreeMap<V, Handle<ST, PK, S>>(comparator);
				}
			} else {
				uniqueStore = new HashMap<V, Handle<ST, PK, S>>();
			}
			
			for (Map.Entry<V,Collection<Handle<ST, PK, S>>> entry: multiStore.entrySet()) {
				for (Handle<ST, PK, S> h: entry.getValue()) {
					if (h.isValid()) {
						uniqueStore.put(entry.getKey(), h);
					}
				}
			}
			
			multiStore.clear();
			multiStore = null;			
		} else if (!this.ordered && ordered) {
			if (Type.UNIQUE.equals(type)) {
				if (comparator==null) {
					uniqueStore = new TreeMap<V, Handle<ST, PK, S>>(uniqueStore);
				} else {
					TreeMap<V, Handle<ST, PK, S>> newStore = new TreeMap<V, Handle<ST, PK, S>>(comparator);
					newStore.putAll(uniqueStore);
					uniqueStore = newStore;
				}
			} else {
				if (comparator==null) {
					multiStore = new TreeMap<V, Collection<Handle<ST, PK, S>>>(multiStore);
				} else {
					TreeMap<V, Collection<Handle<ST, PK, S>>> newStore = new TreeMap<V, Collection<Handle<ST, PK, S>>>(comparator);
					newStore.putAll(multiStore);
					multiStore = newStore;
				}				
			}
		}

		this.type = type;
		this.ordered = ordered;
		this.comparator = comparator;
	}
	
	private Collection<UpdateEntry> pendingUpdates = new LinkedList<AbstractIndex<ST,T,PK,V,S>.UpdateEntry>();

	@Override
	protected Collection<UpdateEntry> getPendingUpdates() {
		return pendingUpdates;
	}

	@SuppressWarnings("rawtypes")
	@Override
	protected ExecutorService getExecutorService() {
		return ((LocalStoreBase) store).getExecutorService();
	}

	/**
	 * Override to customize handling of asynchronous index exceptions.
	 */
	@SuppressWarnings("rawtypes")
	@Override
	protected void handleAsynchException(Exception e) {
		((LocalStoreBase) store).handleIndexAsynchException(e);
	}

}
