package com.hammurapi.common;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.hammurapi.extract.CommutativeAnd;
import com.hammurapi.extract.Extractor;
import com.hammurapi.extract.Predicate;

/**
 * Helper class to join data from several inputs.
 * @author Pavel Vlasov
 * @param <T> Join element type.
 * @param <C> Context type passed from addInput to join.
 * @param <R> Return type passed from join to to addInput.
 */
public abstract class Joiner<T, C, R> {	
	
	/**
	 * Collector of objects to join. Implements a subset of collection 
	 * methods. 
	 * @author Pavel Vlasov
	 *
	 * @param <T>
	 */
	public interface Collector<T> extends Iterable<T> {
				
		/**
		 * Adds new object to the collector. This method shall not break currently open iterators.
		 * @param obj
		 * @return true if object was added to the collector.
		 */
		boolean add(T obj);
		
		/**
		 * Removes object (last entry) from the collector.
		 * @param obj
		 * @return true if success
		 */
		boolean remove(T obj);
		
		/**
		 * Removes all object from the collector.
		 */
		void clear();
		
	}
	
	/**
	 * Adapter for collections.
	 * @author Pavel Vlasov
	 *
	 * @param <T>
	 */
	public static class CollectionAdapter<T> implements Collector<T> {
		
		private Collection<T> master;
		
		public CollectionAdapter(Collection<T> master) {
			this.master = master;
		}

		public boolean add(T obj) {
			return master.add(obj);
		}

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

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

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

		@Override
		public String toString() {
			return "CollectionAdapter [master=" + master + "]";
		}
		
	}
	
	private static final Logger logger = Logger.getLogger(Joiner.class.getName());
	
	/**
	 * Interface to consume inputs in join() method.
	 * @author Pavel Vlasov
	 */
	public static class InputConsumer {
		
		boolean[] consumeFlags;

		InputConsumer(int size) {
			consumeFlags = new boolean[size];
		}
		
		/**
		 * Consumes input at given index.
		 * @param index
		 */
		public void consume(int index) {
			consumeFlags[index] = true;
		}
	}
	
	private Collector<T>[] inputCollectors;
	private boolean outerJoin;
	private Class<T> inputType;
	private boolean isFine;
	
	/**
	 * Predicates to be fired at a particular index.
	 */
	private Map<Integer, List<Predicate<T, C>>> predicates = new HashMap<Integer, List<Predicate<T, C>>>();
	
	/**
	 * Adds join predicate.
	 * @param predicate
	 */
	public void addPredicate(Predicate<T, C> predicate) {
		if (predicate instanceof CommutativeAnd) {
			// Break down the predicate.
			for (Predicate<T,C> part: ((CommutativeAnd<T,C>) predicate).getParts()) {
				addPredicate(part);
			}		
		} else {		
			int maxIndex = 0;
			for (Integer idx: predicate.parameterIndices()) {
				if (idx>maxIndex) {
					maxIndex = idx;
				}
			}
			List<Predicate<T, C>> pl = predicates.get(maxIndex);
			if (pl==null) {
				pl = new ArrayList<Predicate<T, C>>();
				predicates.put(maxIndex, pl);
			}
			pl.add(predicate);
		}
	}

	/**
	 * Creates joiner.
	 * @param inputCollectors Input collectors.
	 * @param inputType Input type is required because of erasure of generics and T is needed at runtime.
	 * @param outerJoin If true, each input addition triggers invocation of join, even if other input collectors
	 * don't have any data.
	 */
	protected Joiner(Collector<T>[] inputCollectors, Class<T> inputType, boolean outerJoin) {
		this.inputCollectors = inputCollectors;
		this.outerJoin = outerJoin;
		this.inputType = inputType;
		this.isFine = logger.isLoggable(Level.FINE);
	}
	
	/**
	 * Start join is invoked before join. Typically this method shall
	 * obtain a lock on input collectors.
	 */
	protected abstract void startJoin();
	
	/**
	 * This method is invoked after join is finished. Typically this method
	 * shall release the lock on input collectors.
	 */
	protected abstract void endJoin();
	
	/**
	 * Adds input. Input is joined with other accumulated inputs.
	 *  
	 * @param index Input index. 
	 * @param input Input.
	 */
	@SuppressWarnings("unchecked")
	public List<R> addInput(int index, T input, C context) throws Exception {
		if (isFine) {
			logger.fine("New join input: "+input+" at index "+index);
			StringBuilder sb = new StringBuilder();
			sb.append("Input collectors: ");
			for (Collector<T> c:inputCollectors) {
				sb.append(c+" ");
			}
			logger.fine(sb.toString());
		}
		if (isValidInput(index, input)) {
			startJoin();
			try {
				if (inputCollectors[index].add(input)) { // Join only if successfully added - for sets.
					T[] inputs = (T[]) Array.newInstance(inputType, inputCollectors.length);
					inputs[index] = input;
					InputConsumer consumer = new InputConsumer(inputCollectors.length);			
					List<R> ret = new ArrayList<R>();
					Map<C, Map<Extractor<T, ? super Boolean, C>, ? super Boolean>> cache = new ConcurrentHashMap<C, Map<Extractor<T,? super Boolean,C>,? super Boolean>>(); 
					join(inputs, 0, index, context, consumer, cache , ret);
					if (consumer.consumeFlags[index]) { // Current input was consumed.
						if (!inputCollectors[index].remove(input)) {
							throw new IllegalStateException("Cannot remove just added input");
						}
					}
					return ret;
				}
			} finally {
				endJoin();
			}
		}
		return null;
	}
	
	/**
	 * This method can be overridden to implement concurrent joining, i.e. at particular indices the sub-method wraps call to the super-method
	 * into a task executed in a separate thread.
	 * @param inputs
	 * @param currentIndex
	 * @param inputIndex
	 * @param context
	 * @param consumer
	 * @param cache
	 * @param resultCollector
	 * @throws Exception
	 */
	protected void join(T[] inputs, int currentIndex, int inputIndex, C context, InputConsumer consumer, Map<C, Map<Extractor<T, ? super Boolean, C>, ? super Boolean>> cache, List<R> resultCollector) throws Exception {
//		System.out.println(this+" Join "+currentIndex+" "+Arrays.toString(inputs));
		if (currentIndex<inputs.length) {
			
			List<Predicate<T, C>> pl = predicates.get(currentIndex);
			if (pl!=null) {
				for (Predicate<T, C> predicate: pl) {
					if (!predicate.extract(context, cache, inputs)) {
						return;
					}
				}
			}
			
			if (!partialJoin(inputs, context, consumer, currentIndex, inputIndex)) {
				return;
			}
		}
		
		if (currentIndex==inputs.length) {
			for (int i=0; i<consumer.consumeFlags.length; ++i) {
				consumer.consumeFlags[i] = false;
			}
			R joinResult = join(inputs, context, consumer, inputIndex);
			synchronized (resultCollector) { // For concurrent joins.
				resultCollector.add(joinResult);
			}
		} else if (currentIndex==inputIndex) {
			join(inputs, currentIndex+1, inputIndex, context, consumer, cache, resultCollector);
		} else {			
			if (outerJoin) {
				inputs[currentIndex]=null;
				join(inputs, currentIndex+1, inputIndex, context, consumer, cache, resultCollector);
				if (consumer.consumeFlags[inputIndex]) { // Input was consumed
					return;
				}
				for (int i=0; i<currentIndex; ++i) { // Lower index input was consumed - get out of here.
					if (consumer.consumeFlags[i]) {
						return;
					}
				}
			}
			
			Iterator<T> it = inputCollectors[currentIndex].iterator();
			try {
				while (it.hasNext()) {
					inputs[currentIndex]=it.next();
					if (isValidInput(currentIndex, inputs[currentIndex])) {
						join(inputs, currentIndex+1, inputIndex, context, consumer, cache, resultCollector);
						if (consumer.consumeFlags[currentIndex]) { // This input was consumed.
							it.remove();					
							return;
						}
						
						if (consumer.consumeFlags[inputIndex]) { // Input was consumed
							return;
						}
						
						for (int i=0; i<currentIndex; ++i) { // Lower index input was consumed - get out of here.
							if (consumer.consumeFlags[i]) {
								return;
							}
						}
					} else {
						if (consumer.consumeFlags[currentIndex]) { // This input is not valid anymore.
							it.remove();					
							return;
						}
					}
				}
			} finally {
				if (it instanceof AutoCloseable) {
					((AutoCloseable) it).close();
				}
			}
		}		
	}
	
	/**
	 * This method is invoked for every combination of valid inputs.
	 * @param inputs Inputs
	 * @param context Join context passed from addInput
	 * @param consumer callback interface to consume inputs.
	 * @param activator Index of join activator.
	 * @return Join result. 
	 * @throws Exception
	 */
	protected abstract R join(T[] inputs, C context, InputConsumer consumer, int activator) throws Exception;
	
	/**
	 * This method is invoked for all inputs before joining. It is useful for situation when 
	 * inputs in internal collections may become invalid over the course of life of joiner.
	 * @param index
	 * @param input
	 * @return
	 */
	protected boolean isValidInput(int index, T input) {
		return true;
	}
	
	/**
	 * This method is invoked to validate already joined inputs.
	 * If this method returns false, further joining is abandoned.
	 * @param inputs Inputs array.
	 * @param index Index of last already joined input.
	 * @param consumer Callback interface to consume inputs. 
	 * @return
	 */
	protected boolean partialJoin(T[] inputs, C context, InputConsumer consumer, int index, int activator) throws Exception {
		return true;
	}
	
	/**
	 * Clears join collections.
	 */
	public void reset() {
		for (Collector<T> collector: inputCollectors) {
			collector.clear();
		}
	}

}
