package com.hammurapi.store;

import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;

import com.hammurapi.common.concurrent.NonBlockingExecutorService;
import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.Predicate;
import com.hammurapi.store.Index.Type;

/**
 * This class operates on master's data. It uses master's locks and storage, and is aware if the master already holds a certain lock. 
 * In this case this implementation doesn't acquire its own lock but operates
 * within master's lock because it acts on behalf of the master.
 * Attempt to acquire write lock if the master holds read lock results in exception.
 * Deputy stores are used by the AbstractStore implementations to operate in asynchronous tasks.
 * @author Pavel Vlasov
 *
 * @param <T>
 * @param <PK>
 */
public abstract class DeputyStore<T,PK,S extends Store<T,PK,S>> extends AbstractStore<T,PK,S> {
        
        protected AbstractStore<T,PK,S> master;
        public int hashCode() {
                return master.hashCode();
        }

        public Handle<T, PK, S> put(T obj) {
                return master.put(obj);
        }

        public boolean equals(Object obj) {
                return master.equals(obj);
        }

        public Iterable<T> getAll() {
                return master.getAll();
        }

        public T getByPrimaryKey(PK primaryKey) {
                return master.getByPrimaryKey(primaryKey);
        }

        public Iterable<T> get(Predicate<T, S> selector) {
                return master.get(selector);
        }

        public <V> Iterable<V> get(
                        Predicate<T, S> selector,
                        Extractor<T, V, S> extractor, 
                        boolean ordered,
                        Comparator<V> comparator) {
                return master.get(selector, extractor, ordered, comparator);
        }

        public String toString() {
                return master.toString();
        }

        public void clear() {
                master.clear();
        }

        public boolean remove(T obj) {
                return master.remove(obj);
        }
        
        public boolean removeByPrimaryKey(PK primaryKey) {
                return master.removeByPrimaryKey(primaryKey); 
        }

        public int remove(Predicate<T, S> selector) {
                return master.remove(selector);
        }

        public int update(
                        Predicate<T, S> selector,
                        UpdateTask<T, PK, S> updater) {
                
                return master.update(selector, updater);
        }
        
        @Override
        public <V, ST extends T> Index<T, ST, PK, V, S> addIndex(
                        Predicate<T, S> predicate,
                        Extractor<ST, V, S> extractor, 
                        Type type,
                        boolean ordered, 
                        Comparator<V> comparator) {
                
                return master.addIndex(predicate, extractor, type, ordered, comparator);
        }

        public Extractor<T, PK, S> getPrimaryKeyExtractor() {
                return master.getPrimaryKeyExtractor();
        }

        public int queryAll(QueryTask<T, PK, S> query) {
                return master.queryAll(query);
        }

        public int query(
                        Predicate<T, S> selector,
                        QueryTask<T, PK, S> query) {
                
                return master.query(selector, query);
        }

        public Iterator<T> iterator() {
                return master.iterator();
        }

        public S createView(Predicate<T, S> selector, ViewType viewType) {
                return master.createView(selector, viewType);
        }

        public Lock readLock() {
                return masterLock.readLock();
        }

        public Lock writeLock() {
                return masterLock.writeLock();
        }

        private ReadWriteLock masterLock;
        
        private NonBlockingExecutorService nonBlockingExecutorService;
        
        /**
         * Creates deputy store.
         * @param master Master store.
         * @param masterLock Lock which delegates to master lock and can be used 
         * from multiple threads/processes, i.e. the lock owner is not the current
         * thread but the master store.
         */
        public DeputyStore(AbstractStore<T,PK,S> master, ReadWriteLock masterLock) {
                this.master = master;
                this.masterLock = masterLock;
                if (master.getExecutorService() instanceof NonBlockingExecutorService) {
                        this.nonBlockingExecutorService = (NonBlockingExecutorService) master.getExecutorService();
                } else {
                        this.nonBlockingExecutorService = new NonBlockingExecutorService(master.getExecutorService());
                }
        }

        @Override
        protected Map<PK, Handle<T, PK, S>> getPkStore() {
                return master.getPkStore();
        }

        @Override
        protected Collection<Handle<T, PK, S>> getNoPkStore() {             
                return master.getNoPkStore();
        }

        @Override
        protected ExecutorService getExecutorService() {
                return nonBlockingExecutorService;
        }

        @Override
        protected Handle<T, PK, S> createHandle(
                        T obj,
                        PK primaryKey,
                        Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache,
                        Predicate<T, S>[] validators) {
                return master.createHandle(obj, primaryKey, cache, validators);
        }

        @Override
        protected Collection<Handle<T, PK, S>> createIndexCollection(boolean unique, boolean ordered, Comparator<T> comparator) {
                return master.createIndexCollection(unique, ordered, comparator);
        }

        @Override
        protected Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> createCache() {
                return master.createCache();
        }
                
        @Override
        protected ReadWriteLock createMasterLock() {
                if (masterLock instanceof DeputyReadWriteLock) {
                        return ((DeputyReadWriteLock) masterLock).createDeputy();
                }
                
                throw new UnsupportedOperationException("Creating master locks from non-deputy locks is not implemented.");
        }

        @Override
        protected void onRemoved(int removed) {
                master.onRemoved(removed);
        }

        @Override
        protected Collection<AbstractIndex<T, ? extends T, PK, ?, S>> getIndices() {
                return master.getIndices();
        }

        @Override
        protected <V, ST extends T> AbstractIndex<T, ST, PK, V, S> createIndex(
                        Predicate<T, S> predicate,
                        Extractor<ST, V, S> extractor, Type type,
                        boolean ordered, Comparator<V> comparator) {
                return master.createIndex(predicate, extractor, type, ordered, comparator);
        }

        protected AbstractStore<T, PK, S> getMaster() {
                return master;
        }
}