package com.hammurapi.store;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReadWriteLock;

import com.hammurapi.common.Util;
import com.hammurapi.extract.CommutativeOr;
import com.hammurapi.extract.ComparisonResult;
import com.hammurapi.extract.ComparisonResult.Type;
import com.hammurapi.extract.CompositePredicate;
import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.IndexedExtractor;
import com.hammurapi.extract.Predicate;
import com.hammurapi.extract.True;

/**
 * Abstract base class for object stores.
 * @author Pavel Vlasov
 *
 * @param <T>
 * @param <S> Self-type - sub-type of store. Generic parameter for predicates, extractors, update tasks and views. 
 * S should be an interface for master store, deputy store and views to implement.
 */
public abstract class AbstractStore<T, PK, S extends Store<T,PK,S>> implements Store<T, PK, S> {
	
	/**
	 * @return Store for the mode with primary key.
	 */
	protected abstract Map<PK, Handle<T, PK, S>> getPkStore();
	
	/**
	 * Store for the mode without primary key or pkStore values collection.
	 */
	protected abstract Collection<Handle<T, PK, S>> getNoPkStore();
	
	/**
	 * @return Executor service to parallel processing.
	 */
	protected abstract ExecutorService getExecutorService();
	
	/**
	 * Creates handle.
	 */
	protected abstract Handle<T, PK, S> createHandle(T obj, PK primaryKey, Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache, Predicate<T, S>[] validators);
	
	/**
	 * Creates collection for index entries
	 * @param unique Collection should be unique (set)
	 * @param ordered Collection should be ordered by the comparator or by natural order.
	 * @param comparator
	 * @return
	 */
	protected abstract Collection<Handle<T, PK, S>> createIndexCollection(boolean unique, boolean ordered, Comparator<T> comparator);
	
	protected Handle<T, PK, S> createHandle(T obj, Predicate<T, S>[] validators) {
		return createHandle(obj, null, createCache(), validators);
	}
	
	/**
	 * @return Cache for extractors.
	 */
	protected abstract Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> createCache();
		
	/**
	 * Creates a store which encapsulates current call state (e.g. locks currently held) and can be
	 * passed to executor for asynchronous execution. Delegates operate on behalf of the master
	 * store. They don't acquire locks if the master store already holds them.
	 * If the master store holds read lock then attempt of a deputy to acquire write lock will result in
	 * StoreException. 
	 * @return
	 */
	protected abstract S createDeputy();
	protected abstract ReadWriteLock createMasterLock();
	
	/**
	 * Invoked when some handles have been removed.
	 * This operation shall remove invalidated handles from store and index collections.
	 * This method is invoked outside of the write lock.
	 * @param removed Number of removed handles.
	 */
	protected abstract void onRemoved(int removed);		
	
	/*
	 * Implementation note - removal of objects is done by marking Handle as removed. Handles may or may not be removed from store and index collections
	 * upon marking them as removed. Therefore there should be a "garbage collection" mechanism for removal of invalid handles. 
	 */
	
	@SuppressWarnings("unchecked")
	@Override
	public Handle<T, PK, S> put(T obj, Predicate<T, S>... validators) {
		// Evaluate validators.
		for (Predicate<T, S> v: validators) {
			if (!v.extract((S) this, null, Util.wrap(obj))) {
				return null;
			}
		}
		writeLock().lock();
		try {
			if (getPrimaryKeyExtractor()==null) {
				Handle<T, PK, S> ret = createHandle(obj, validators);
				if (getNoPkStore().add(ret)) {
					updateIndices(ret, true);
				} else {
					// Not unique
					for (Handle<T, PK, S> hi: getNoPkStore()) {
						if (ret.equals(hi)) {
							return hi;
						}
					}
				} 
				return ret;
			} else {
				Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache = createCache();			
				PK primaryKey = getPrimaryKeyExtractor().extract((S) this, cache, Util.wrap(obj));
				Handle<T, PK, S> hi = getPkStore().get(primaryKey);
				if (hi==null) {
					hi = createHandle(obj, primaryKey, cache, validators);
					updateIndices(hi, true);
					getPkStore().put(primaryKey, hi);
				} else {
					hi.update(obj);
				}
				return hi;
			}
		} finally {
			writeLock().unlock();
		}
	}

	@Override
	public Iterable<T> getAll() {
		readLock().lock();
		try {
			Collection<T> ret = new ArrayList<T>();
			for (Handle<T, PK, S> hi: getNoPkStore()) {
				if (hi.isValid()) {
					ret.add(hi.get());
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}

	@Override
	public T getByPrimaryKey(PK primaryKey) {
		readLock().lock();
		try {
			if (getPrimaryKeyExtractor()==null) {
				throw new IllegalStateException("Object store is configured without primary key");
			}
			Handle<T, PK, S> hi = getPkStore().get(primaryKey);
			if (hi==null || !hi.isValid()) {
				// TODO - get from the backing store, update indices
				return null;
			}
			return hi.get();
		} finally {
			readLock().unlock();
		}
	}

	@Override
	public Iterable<T> get(Predicate<T, S> selector) {
		Collection<T> ret = new ArrayList<T>();
		for (Handle<T, PK, S> h: getHandles(selector)) {
			ret.add(h.get());
		}
		return ret;
	}
	
	protected class SelectorTask implements Callable<Handle<T, PK, S>>, Serializable {
		
		private Predicate<T, S> selector;
		private Handle<T, PK, S> handle;

		public SelectorTask(Handle<T, PK, S> handle, Predicate<T, S> selector) {
			this.handle = handle;
			this.selector = selector;
		}

		/**
		 * Returns object if predicate matches.
		 */
		@Override
		public Handle<T, PK, S> call() throws Exception {
			if (handle.isValid() && (selector==null || handle.extract(selector))) {
				return handle;
			}
			return null;
		}
		
	}
	
	/**
	 * Result of applying index optimizations to data manipulation parameters.
	 * @author Pavel Vlasov
	 *
	 */
	private class IndexizationResult {
		private Iterable<Handle<T, PK, S>> source;
		private Predicate<T, S> selector;
		private boolean ordered;
		
		public IndexizationResult(
				Iterable<Handle<T, PK, S>> source,
				Predicate<T, S> selector, 
				boolean ordered) {
			super();
			this.source = source;
			this.selector = selector;
			this.ordered = ordered;
		}
		
		/**
		 * @return Data to iterate over for further refinement.
		 */
		public Iterable<Handle<T, PK, S>> getSource() {
			return source;
		}
		
		/**
		 * @return Refined selector with parts addressed by indices already removed.
		 */
		public Predicate<T, S> getSelector() {
			return selector;
		}
		
		/**
		 * @return true if source is already ordered as required.
		 */
		public boolean isOrdered() {
			return ordered;
		}
		
	}
	
	private <ST extends T> ComparisonResult.Type comparePredicates(
			Predicate<T, S> selector,
			AbstractIndex<T, ST,PK,?,S> index
			) {
		
		if (selector==null) {
			if (index.getPredicate()==null || True.getInstance().equals(index.getPredicate())) {
				return Type.LESS_RESTRICTIVE;
			}
			return Type.NOT_EQUAL;
		}
		
		if (index.getPredicate()==null || True.getInstance().equals(index.getPredicate())) {
			return Type.LESS_RESTRICTIVE;
		}
		
		ComparisonResult cr = index.getPredicate().compareTo(selector);
		if (!cr.isOneToOneMapping()) {
			return Type.NOT_EQUAL;
		}
		
		switch (cr.getType()) {
		case EQUAL:
		case LESS_RESTRICTIVE:
			return cr.getType();
		default:
			return Type.NOT_EQUAL;
		}
	}
	
	/**
	 * @param <V>
	 * @param extractor
	 * @param comparator
	 * @return
	 */
	private <V, ST extends T> boolean compatibleOrder(Extractor<T, V, S> extractor, Comparator<V> comparator, AbstractIndex<T, ST, PK,V,S> index) {
		if (index.isOrdered()) {
			if ((extractor==null && index.getExtractor()==null) || extractor.equals(index.getExtractor())) {
				if ((comparator==null && index.getComparator()==null) || comparator.equals(index.getComparator())) {
					return true;
				}
			}
		}
		return false;
	}
			
	/**
	 * Apply indexing optimizations.
	 * @return
	 */
	private <V> IndexizationResult indexize(			
			Predicate<T, S> selector,
			final Extractor<T, V, S> extractor, 
			final boolean ordered,
			Comparator<V> comparator) {
		
		class Candidate implements Comparable<Candidate> {
			AbstractIndex<T, ? extends T,PK,?,S> index;
			ComparisonResult.Type type;
			private boolean offersOrdering;
			
			public Candidate(AbstractIndex<T, ? extends T, PK, ?, S> index, Type type, boolean offersOrdering) {
				super();
				this.index = index;
				this.type = type;
				this.offersOrdering = offersOrdering;
			}

			@Override
			public int compareTo(Candidate o) {
				
				// If ordered, indices with compatible order come first, if not, unordered come first
				if (ordered) {
					if (offersOrdering) {
						if (!o.offersOrdering) {
							return -1;
						}
					} else {
						if (o.offersOrdering) {
							return 1;
						}
					}
				} else {
					if (index.isOrdered()) {
						if (!o.index.isOrdered()) {
							return 1;
						}
					} else {
						if (o.index.isOrdered()) {
							return -1;
						}
					}
				}
				
				// Predicates equal to selector come first.
				if (!o.type.equals(type)) {
					return Type.EQUAL.equals(type) ? -1 : 1;
				}
				
				// Less restrictive predicates come after more restrictive.
				if (index.getPredicate()==null) {
					if (o.index.getPredicate()!=null) {
						return 1;
					}
				} else {
					if (o.index.getPredicate()==null) {
						return -1;
					}
					
					ComparisonResult cr = index.getPredicate().compareTo(o.index.getPredicate());
					if (cr.isOneToOneMapping()) {
						switch (cr.getType()) {
						case LESS_RESTRICTIVE:
							return 1;
						case MORE_RESTRICTIVE:
							return -1;
						}
					}
				}
				
				if (index.isUnique()) {
					if (!o.index.isUnique()) {
						return -1;
					}
				} else {
					if (o.index.isUnique()) {
						return 1;
					}
				}
				
				// More costly index comes first
				Extractor<? extends T, ?, S> indexExtractor = index.getExtractor();
				Extractor<? extends T, ?, S> otherExtractor = o.index.getExtractor();
				double indexCost = indexExtractor==null ? 0 : indexExtractor.getCost();
				double otherCost = otherExtractor==null ? 0 : otherExtractor.getCost();
				if (indexCost>0 && otherCost>0) {
					if (indexCost*o.index.getGranularity()>otherCost*index.getGranularity()) {
						return -1;
					}
					if (indexCost*o.index.getGranularity()<otherCost*index.getGranularity()) {
						return -1;
					}
				}
				
				int gDelta = o.index.getGranularity() - index.getGranularity();
				if (gDelta!=0) {
					return gDelta;
				}
				
				// Apply synchronous first, lazy last
				int typeCmp = type.compareTo(o.type);
				if (typeCmp!=0) {
					return typeCmp;
				}
				
				return hashCode() - o.hashCode();
			}
						
		}
		
		List<Candidate> candidates = new ArrayList<Candidate>();
		for (AbstractIndex<T, ? extends T, PK,?,S> candidate: getIndices()) {
			ComparisonResult.Type type = comparePredicates(selector, candidate);
			if (!Type.NOT_EQUAL.equals(type)) {
				// Add no-predicate indexes only if they offer ordering.
				@SuppressWarnings("unchecked")
				boolean offersOrdering = ordered && compatibleOrder(extractor, comparator, (AbstractIndex<T,? extends T,PK,V,S>) candidate);
				if (candidate.getPredicate()!=null || offersOrdering) {				
					candidates.add(new Candidate(candidate, type, offersOrdering));
				}
			}
		}
		
		if (!candidates.isEmpty()) {
			Collections.sort(candidates);
			Candidate indexToUse = candidates.get(0);
			if (Type.EQUAL.equals(indexToUse.type)) {
				return new IndexizationResult(indexToUse.index.getHandles(), null, indexToUse.offersOrdering);				
			} else if (selector instanceof CompositePredicate) { 
				@SuppressWarnings("unchecked")
				Predicate<T, S> newSelector = ((CompositePredicate<T, ?, S, ?>) selector).remove(indexToUse.index.getPredicate());
				return new IndexizationResult(indexToUse.index.getHandles(), newSelector, indexToUse.offersOrdering);
			}
			// Less restrictive, not composite
			return new IndexizationResult(indexToUse.index.getHandles(), selector, indexToUse.offersOrdering);				
		}
				
		/**
		 * Or selector - we can use multiple indices to select, the we have to sort anyway.
		 */
		if (selector instanceof CommutativeOr && ((CommutativeOr) selector).getParts().size()>1) {
			Collection<Future<IndexizationResult>> indexizationResults = new LinkedList<Future<IndexizationResult>>();
			for (Predicate<T, S> part: ((CommutativeOr<T,S>) selector).getParts()) {
				indexizationResults.add(getExecutorService().submit(new IndexizationTask<V>(part, extractor)));
			}
			
			Collection<Handle<T, PK, S>> ret = new LinkedList<Handle<T, PK, S>>();
			Collection<Future<Handle<T, PK, S>>> fret = new LinkedList<Future<Handle<T, PK, S>>>();
			
			for (Future<IndexizationResult> fir: indexizationResults) {
				try {
					IndexizationResult ir = fir.get();
					if (ir.getSelector()==null || True.getInstance().equals(ir.getSelector())) {
						for (Handle<T, PK, S> h: ir.getSource()) {
							ret.add(h);
						}
					} else {
						for (Handle<T, PK, S> h: ir.getSource()) {
							fret.add(getExecutorService().submit(new SelectorTask(h, ir.getSelector())));
						}
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			
			for (Future<Handle<T, PK, S>> fh: fret) {
				try {
					Handle<T, PK, S> h = fh.get();
					if (h!=null) {
						ret.add(h);
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			
			return new IndexizationResult(ret, null, false);
		}		
		
		
		return new IndexizationResult(getNoPkStore(), selector, false);		
	}
	
	protected class IndexizationTask<V> implements Callable<IndexizationResult> {

		private Predicate<T, S> selector;
		private Extractor<T, V, S> extractor;

		public IndexizationTask(Predicate<T, S> selector, Extractor<T, V, S> extractor) {
			this.selector = selector;
			this.extractor = extractor;
		}
		
		@Override
		public IndexizationResult call() throws Exception {
			return indexize(selector, extractor, false, null);
		}
		
	}

	/**
	 * @param selector
	 * @return
	 */
	protected Iterable<Handle<T, PK, S>> getHandles(Predicate<T, S> selector) {
		readLock().lock();
		try {
			IndexizationResult ir = indexize(selector, null, false, null);
			Collection<Future<Handle<T, PK, S>>> selectResults = new ArrayList<Future<Handle<T, PK, S>>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				selectResults.add(getExecutorService().submit(new SelectorTask(h, ir.getSelector())));
			}
			Collection<Handle<T, PK, S>> ret = new ArrayList<Handle<T, PK, S>>();
			for (Future<Handle<T, PK, S>> f: selectResults) {
				try {
					Handle<T, PK, S> result = f.get();
					if (result!=null) {
						ret.add(result);
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}

	protected class ExtractorTask<V> implements Callable<V>, Serializable {
		
		private Predicate<T, S> selector;
		private Handle<T, PK, S> handle;
		private Extractor<T, V, S> extractor;

		public ExtractorTask(Handle<T, PK, S> handle, Predicate<T, S> selector, Extractor<T, V, S> extractor) {
			this.handle = handle;
			this.selector = selector;
			this.extractor = extractor;
		}

		/**
		 * Returns object if predicate matches.
		 */
		@Override
		public V call() throws Exception {
			if (handle.isValid() && (selector==null || handle.extract(selector))) {				
				if (extractor==null) {
					return ((V) handle.get());
				}
				return handle.extract(extractor);
			}
			return null;
		}
		
	}
	
	@Override
	public <V> Iterable<V> get(
			Predicate<T, S> selector,
			Extractor<T, V, S> extractor, 
			boolean ordered,
			Comparator<V> comparator) {
		
		readLock().lock();
		try {
			IndexizationResult ir = indexize(selector, extractor, ordered, comparator);
			Collection<Future<V>> extractResults = new ArrayList<Future<V>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				extractResults.add(getExecutorService().submit(new ExtractorTask<V>(h, ir.getSelector(), extractor)));
			}
			List<V> ret = new ArrayList<V>();
			for (Future<V> f: extractResults) {
				try {
					V result = f.get();
					if (result!=null) {
						ret.add(result);
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			if (ordered && !ir.isOrdered()) {
				if (comparator==null) {
					// Trick.
					Collections.sort(ret, new NaturalComparator<V>());
				} else {
					Collections.sort(ret, comparator);
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}
	
	protected class ValueSelectorTask<V> implements Callable<V>, Serializable {
		
		private Predicate<V, S> selector;
		private V value;

		public ValueSelectorTask(V value, Predicate<V, S> selector) {
			this.value = value;
			this.selector = selector;
		}

		/**
		 * Returns object if predicate matches.
		 */
		@Override
		public V call() throws Exception {
			if (selector.extract(createDeputy(), null, Util.wrap(value))) {
				return value;
			}
			return null;
		}
		
	}		

	@Override
	public <V> Iterable<V> getMultiple(
			Predicate<T, S> selector,
			Extractor<T, Iterable<V>, S> extractor, 
			Predicate<V, S> valueSelector, 					
			boolean ordered,
			Comparator<V> comparator) {
		readLock().lock();
		try {
//			IndexizationResult ir = indexize(selector, extractor, ordered, comparator);
			Collection<Future<Iterable<V>>> extractResults = new ArrayList<Future<Iterable<V>>>();
			for (Handle<T, PK, S> h: getNoPkStore()) {
				extractResults.add(getExecutorService().submit(new ExtractorTask<Iterable<V>>(h, selector, extractor)));
			}
			List<V> ret = new ArrayList<V>();
			if (valueSelector==null) {
				for (Future<Iterable<V>> f: extractResults) {
					try {
						Iterable<V> result = f.get();
						if (result!=null) {
							for (V v: result) {
								ret.add(v);
							}
						}
					} catch (Exception e) {
						throw new StoreException(e);
					}
				}
			} else {
				List<Future<V>> fRet = new LinkedList<Future<V>>();
				for (Future<Iterable<V>> f: extractResults) {
					try {
						Iterable<V> result = f.get();
						if (result!=null) {
							for (V v: result) {
								fRet.add(getExecutorService().submit(new ValueSelectorTask<V>(v, valueSelector)));
							}
						}
					} catch (Exception e) {
						throw new StoreException(e);
					}
				}
				for (Future<V> f: fRet) {
					try {
						V result = f.get();
						if (result!=null) {
							ret.add(result);
						}
					} catch (Exception e) {
						throw new StoreException(e);
					}
				}				
			}
			if (ordered) {
				if (comparator==null) {
					// Trick.
					Collections.sort(ret, new NaturalComparator<V>());
				} else {
					Collections.sort(ret, comparator);
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}

	@Override
	public void clear() {
		writeLock().lock();
		try {
			// Marking handles as removed.
			for (Handle<T, PK, S> h: getNoPkStore()) {
				h.remove();
			}
			// TODO - clear indices, remove from the backing store.
			if (getPrimaryKeyExtractor()==null) {
				getNoPkStore().clear();
			} else {
				getPkStore().clear();
			}
		} finally {
			writeLock().unlock();
		}
	}

	@Override
	public boolean remove(T obj) {
		int cleanedUp = 0;
		writeLock().lock();
		try {
			Iterator<Handle<T, PK, S>> it = getNoPkStore().iterator();
			while (it.hasNext()) {
				Handle<T, PK, S> next = it.next();
				if (obj.equals(next.get())) {
					next.remove(); // Handles shall just invalidate self without modifying the containing collection.
					it.remove();
					// TODO remove from indices, remove from the backing store.
					return true;
				}
				// Clean up also directly removed (through handle.remove()) entries.
				if (!next.isValid()) {
					it.remove();
					++cleanedUp;
				}
			}
			return false;
		} finally {
			writeLock().unlock();
			onRemoved(-cleanedUp); 
		}
	}
	
	public boolean removeByPrimaryKey(PK primaryKey) {
		writeLock().lock();
		try {
			if (getPrimaryKeyExtractor()==null) {
				throw new IllegalStateException("Object store is configured without primary key");
			}
			Handle<T, PK, S> hi = getPkStore().remove(primaryKey);
			if (hi==null || !hi.isValid()) {
				// TODO - get from the backing store, update indices
				return false;
			}
			hi.remove();
			return true;
		} finally {
			writeLock().unlock();
		}
		
	};

	protected class RemoveTask implements Callable<Boolean>, Serializable {
		
		private Predicate<T, S> selector;
		private Handle<T, PK, S> handle;

		public RemoveTask(Handle<T, PK, S> handle, Predicate<T, S> selector) {
			this.handle = handle;
			this.selector = selector;
		}

		/**
		 * Removes object and returns true if predicate matches.
		 */
		@Override
		public Boolean call() throws Exception {
			if (handle.isValid() && handle.extract(selector)) {
				handle.remove(); // TODO - remove from the backing store, shall be handled by handle.remove().
				return true;
			}
			return false;
		}
		
	}
	
	@Override
	public int remove(Predicate<T, S> selector) {
		int ret = 0;
		writeLock().lock();
		try {
			IndexizationResult ir = indexize(selector, null, false, null);
			Collection<Future<Boolean>> removeResults = new ArrayList<Future<Boolean>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				removeResults.add(getExecutorService().submit(new RemoveTask(h, ir.getSelector())));
			}
			for (Future<Boolean> f: removeResults) {
				try {
					if (f.get()) {
						++ret;
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			return ret;
		} finally {
			writeLock().unlock();
			onRemoved(ret);
		}
	}

	protected class UpdateCallable implements Callable<Boolean>, Serializable {
		
		private Predicate<T, S> selector;
		private Handle<T, PK, S> handle;
		private UpdateTask<T, PK, S> updater;
		private S storeDeputy;

		public UpdateCallable(Handle<T, PK, S> handle, Predicate<T, S> selector, UpdateTask<T,PK,S> updater, S storeDeputy) {
			this.handle = handle;
			this.selector = selector;
			this.updater = updater;
			this.storeDeputy = storeDeputy;
		}

		/**
		 * Updates object and returns true if predicate matches.
		 */
		@Override
		public Boolean call() throws Exception {
			if (handle.isValid() && handle.extract(selector)) {
				updater.execute(storeDeputy, handle); // TODO - update backing store.
				return true;
			}
			return false;
		}
		
	}
	
	@Override
	public int update(Predicate<T, S> selector, UpdateTask<T, PK, S> updater) {
		writeLock().lock();
		try {
			IndexizationResult ir = indexize(selector, null, false, null);
			Collection<Future<Boolean>> updateResults = new ArrayList<Future<Boolean>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				updateResults.add(getExecutorService().submit(new UpdateCallable(h, ir.getSelector(), updater, createDeputy())));
			}
			int ret = 0;
			for (Future<Boolean> f: updateResults) {
				try {
					if (f.get()) {
						++ret;
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			return ret;
		} finally {
			writeLock().unlock();
		}
	}
	
	protected abstract Collection<AbstractIndex<T,? extends T,PK, ?, S>> getIndices();
	
	protected abstract <V, ST extends T> AbstractIndex<T,ST,PK,V,S> createIndex(
			Predicate<T, S> predicate,
			Extractor<ST, V, S> extractor,
			Index.Type type,
			boolean ordered, 
			Comparator<V> comparator);
		
	@SuppressWarnings("unchecked")
	@Override
	public <V, ST extends T> Index<T,ST,PK,V,S> addIndex(
			Predicate<T, S> predicate,
			Extractor<ST, V, S> extractor,
			Index.Type type,
			boolean ordered, 
			Comparator<V> comparator) {
		
		writeLock().lock();
		try {
			for (AbstractIndex<T, ? extends T, PK, ?, S> index: getIndices()) {
				if ((predicate==null && index.getPredicate()==null) || (predicate!=null && predicate.equals(index.getPredicate()))) {
					if ((extractor == null && index.getExtractor()==null) || (extractor!=null && extractor.equals(index))) {
						if (ordered) {
							if (!index.isOrdered()) {
								// Upgrade index to ordered
								((AbstractIndex<T, ST, PK,V,S>) index).upgrade(type, ordered, comparator);
								return (Index<T, ST, PK, V, S>) index;
							} else {
								if ((index.getComparator()==null && comparator==null) || (comparator!=null && comparator.equals(index.getComparator()))) {
									if (index.type.compareTo(type)>0) {
										((AbstractIndex<T,ST,PK,V,S>) index).upgrade(type, ordered, comparator);
									}
									return (Index<T, ST, PK, V, S>) index;									
								}
							}
						} else {
							// Requested index not ordered, doesn't matter if existing index is ordered.
							// Upgrade type if required.
							if (index.type.compareTo(type)>0) {
								((AbstractIndex<T,ST,PK,V,S>) index).upgrade(type, ordered, comparator);
							}
							return (Index<T, ST, PK, V, S>) index;
						}
					}
				}
			}
			AbstractIndex<T, ST, PK, V, S> ret = createIndex(predicate, extractor, type, ordered, comparator);
			for (Handle<T, PK, S> hi: getNoPkStore()) {
				if (hi.isValid()) {
					ret.update(hi, true);
				}
			}
			getIndices().add(ret);
			return ret.createProxy();
		} finally {
			writeLock().unlock();
		}
		
	}
	
	private final Extractor<T,T, S> SELF_EXTRACTOR = new IndexedExtractor<T, S>(0);

	protected class QueryCallable<V> implements Callable<Boolean>, Serializable {
		
		private Predicate<T, S> selector;
		private Handle<T, PK, S> handle;
		private QueryTask<V, PK, S> query;
		private S storeDeputy;
		private Extractor<T, V, S> extractor;

		@SuppressWarnings("unchecked")
		public QueryCallable(
				Handle<T, PK, S> handle, 
				Predicate<T, S> selector,
				Extractor<T, V, S> extractor,
				QueryTask<V,PK, S> query, 
				S storeDeputy) {
			this.handle = handle;
			this.selector = selector;
			this.query = query;
			this.storeDeputy = storeDeputy.createUnmodifiableFacade();
			this.extractor = extractor;
			if (this.extractor==null) {
				this.extractor = (Extractor<T, V, S>) SELF_EXTRACTOR;
			}
		}

		/**
		 * Updates object and returns true if predicate matches.
		 */
		@Override
		public Boolean call() throws Exception {
			if (handle.isValid() && (selector==null || handle.extract(selector))) {
				query.execute(storeDeputy, handle.extract(extractor), handle.getPrimaryKey());
				return true;
			}
			return false;
		}
		
	}
		
	@Override
	public int queryAll(final QueryTask<T, PK, S> query) {
		readLock().lock();
		try {
			Collection<Future<Boolean>> queryResults = new ArrayList<Future<Boolean>>();
			for (Handle<T, PK, S> h: getNoPkStore()) {
				queryResults.add(getExecutorService().submit(new QueryCallable<T>(h, null, null, query, createDeputy())));
			}
			int ret = 0;
			for (Future<Boolean> f: queryResults) {
				try {
					if (f.get()) {
						++ret;
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}

	@Override
	public int query(Predicate<T, S> selector, QueryTask<T, PK, S> query) {
		readLock().lock();
		try {
			IndexizationResult ir = indexize(selector, null, false, null);
			Collection<Future<Boolean>> queryResults = new ArrayList<Future<Boolean>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				queryResults.add(getExecutorService().submit(new QueryCallable<T>(h, ir.getSelector(), null, query, createDeputy())));
			}
			int ret = 0;
			for (Future<Boolean> f: queryResults) {
				try {
					if (f.get()) {
						++ret;
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			
			return ret;
		} finally {
			readLock().unlock();
		}
	}
	
	@Override
	public <V> int query(
			Predicate<T, S> selector,
			Extractor<T, V, S> extractor,
			QueryTask<V, PK, S> query) {
		
		readLock().lock();
		try {
			IndexizationResult ir = indexize(selector, null, false, null);
			Collection<Future<Boolean>> queryResults = new ArrayList<Future<Boolean>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				queryResults.add(getExecutorService().submit(new QueryCallable<V>(h, ir.getSelector(), extractor, query, createDeputy())));
			}
			int ret = 0;
			for (Future<Boolean> f: queryResults) {
				try {
					if (f.get()) {
						++ret;
					}
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			
			return ret;
		} finally {
			readLock().unlock();
		}
	}
	
	protected class QueryMultipleCallable<V> implements Callable<Collection<Future<Integer>>>, Serializable {
		
		private Predicate<T, S> selector;
		private Handle<T, PK, S> handle;
		private QueryTask<V, PK, S> query;
		private S storeDeputy;
		private Extractor<T, Iterable<V>, S> extractor;
		private Predicate<V, S> valueSelector;

		public QueryMultipleCallable(
				Handle<T, PK, S> handle, 
				Predicate<T, S> selector,
				Extractor<T, Iterable<V>, S> extractor,
				Predicate<V, S> valueSelector,						
				QueryTask<V,PK,S> query, 
				S storeDeputy) {
			this.handle = handle;
			this.selector = selector;
			this.query = query;
			this.storeDeputy = storeDeputy;
			this.extractor = extractor;
			this.valueSelector = valueSelector;
		}

		/**
		 * Updates object and returns true if predicate matches.
		 */
		@Override
		public Collection<Future<Integer>> call() throws Exception {
			if (handle.isValid() && (selector==null || handle.extract(selector))) {
				Collection<Future<Integer>> queryItemTasks = new LinkedList<Future<Integer>>();
				for (V v: handle.extract(extractor)) {
					queryItemTasks.add(getExecutorService().submit(new QueryItemCallable<V>(v, valueSelector, handle.getPrimaryKey(), query, storeDeputy)));
				}
				return queryItemTasks;
			}
			return Collections.emptyList();
		}
		
	}
	
	protected class QueryItemCallable<V> implements Callable<Integer> {

		private V v;
		private PK primaryKey;
		private QueryTask<V, PK, S> query;
		private S storeDeputy;
		private Predicate<V, S> valueSelector;

		public QueryItemCallable(
				V v, 
				Predicate<V, S> valueSelector,
				PK primaryKey, QueryTask<V, PK, S> query, 
				S storeDeputy) {
			this.v = v;
			this.valueSelector = valueSelector;
			this.primaryKey = primaryKey;
			this.query = query;
			this.storeDeputy = storeDeputy.createUnmodifiableFacade();
		}

		@Override
		public Integer call() throws Exception {
			if (valueSelector==null || valueSelector.extract(storeDeputy, null, Util.wrap(v))) {
				query.execute(storeDeputy, v, primaryKey);
			}
			return 1;
		}
		
	}
	
	@Override
	public <V> int queryMultiple(
			Predicate<T, S> selector,
			Extractor<T, Iterable<V>, S> extractor,
			Predicate<V, S> valueSelector,					
			QueryTask<V, PK, S> query) {
		readLock().lock();
		try {
			IndexizationResult ir = indexize(selector, null, false, null);
			Collection<Future<Collection<Future<Integer>>>> queryResults = new ArrayList<Future<Collection<Future<Integer>>>>();
			for (Handle<T, PK, S> h: ir.getSource()) {
				queryResults.add(getExecutorService().submit(new QueryMultipleCallable<V>(h, ir.getSelector(), extractor, valueSelector, query, createDeputy())));
			}
			Collection<Future<Integer>> itemResults = new LinkedList<Future<Integer>>();
			for (Future<Collection<Future<Integer>>> f: queryResults) {
				try {
					itemResults.addAll(f.get());
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			
			int ret=0;
			for (Future<Integer> f: itemResults) {
				try {
					ret+=f.get();
				} catch (Exception e) {
					throw new StoreException(e);
				}
			}
			return ret;
		} finally {
			readLock().unlock();
		}
	}
	
	/**
	 * This method is invoked by handle implementations. 
	 * It shall not be invoked by client code.
	 * @param handle
	 * @param isNew
	 */
	protected void updateIndices(Handle<T, PK, S> handle, boolean isNew) {
		Iterator<AbstractIndex<T, ? extends T, PK, ?, S>> it = getIndices().iterator();
		while (it.hasNext()) {
			((AbstractIndex<T, ? extends T, PK, ?, S>) it.next()).update(handle, isNew);
		}
	}
	
	@Override
	public Iterator<T> iterator() {
		return getAll().iterator();
	}

	@Override
	public S createView(
			Predicate<T, S> selector, 
			ViewType viewType) {
		
		switch (viewType) {
		case LIVE:
			return createLiveView(selector);
		case SYNCHRONOUS:
		case ASYNCHRONOUS: 
		case LAZY:
		default:
			throw new UnsupportedOperationException("View type is not supported: "+viewType);			
		}
	}

	protected abstract S createLiveView(Predicate<T, S> selector);
	
//	protected 

}
