package com.hammurapi.common.concurrent.transactions;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.transaction.NotSupportedException;
import javax.transaction.Status;
import javax.transaction.SystemException;
import javax.transaction.Transaction;
import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;

import com.hammurapi.common.Context;
import com.hammurapi.common.MutableContext;
import com.hammurapi.common.SimpleMutableContext;
import com.hammurapi.common.concurrent.CompositeLogger;
import com.hammurapi.common.concurrent.LocalTrackingExecutorService;
import com.hammurapi.common.concurrent.LoggerWrapper;
import com.hammurapi.common.concurrent.PropertySet;
import com.hammurapi.extract.Predicate;

/**
 * This class performs concurrent transactional computations. 
 * @author Pavel Vlasov
 *
 * @param <KP>
 */
public abstract class AbstractTransactionalProcessor<KP> {
	
	/**
	 * Interface to propagate transactions between threads.
	 * @author Pavel Vlasov
	 *
	 */
	protected interface TransactionPropagation {
		
		/**
		 * This method is invoked in the thread where transaction shall be propagated to.
		 * @return Propagated transaction.
		 */
		Transaction getTransaction();
		
		/**
		 * Disassociates transaction from the caller thread.
		 */
		void dispose();
	}
	
	/**
	 * Propagates transaction object AS-IS
	 * @author Pavel Vlasov
	 *
	 */
	protected class SimpleTransactionPropagation implements TransactionPropagation {
		
		private Transaction transaction;

		public SimpleTransactionPropagation(Transaction transaction) {
			this.transaction = transaction;
		}

		@Override
		public Transaction getTransaction() {
			return transaction;
		}

		@Override
		public void dispose() {
			// NOP			
		}
		
	}
	
	/**
	 * Creates transaction propagation. This implementation creates SimpleTransactionPropagation
	 * @param transaction
	 * @return
	 */
	protected TransactionPropagation createTransactionPropagation(Transaction transaction) {
		return new SimpleTransactionPropagation(transaction);
	}
	
	protected ThreadLocal<Transaction> threadTransaction = new ThreadLocal<Transaction>();
	
	/**
	 * @return Thread's context transaction.
	 */
	public Transaction getCurrentTransaction() {
		return threadTransaction.get();
	}
		
	private class TransactionalRunnable implements Runnable {
		
		private TransactionPropagation transactionPropagation;
		private Runnable master;

		public TransactionalRunnable(Runnable master) {
			this.master = master;
			this.transactionPropagation = createTransactionPropagation(threadTransaction.get());
		}

		@Override
		public void run() {
			Transaction transaction = transactionPropagation.getTransaction();
			Transaction prevTransaction = threadTransaction.get();
			threadTransaction.set(transaction);
			try {
				master.run();
			} catch (RuntimeException e) {
				LOGGER.log(Level.SEVERE, "Task "+master+" execution failed, setting transaction to rollback only: "+e, e);
				if (transaction!=null) {
					try {
						transaction.setRollbackOnly();
					} catch (SystemException se) {
						LOGGER.log(Level.SEVERE, "Transactional execution failed: "+se, se);
						throw new ProcessingException(se);
					}
				}
				throw e;
			} finally {
				threadTransaction.set(prevTransaction);
				transactionPropagation.dispose();
			}
		}								
	}
	
	
	ExecutorService executorService;

	protected abstract PropertySet<KP> createPropertySet(ExecutorService executorService, Context context, PropertySet<KP>... chain);
	
	public PropertySet<KP> createPropertySet(Context context, PropertySet<KP>... chain) {
		return createPropertySet(executorService, context, chain);
	}
	
	static final Logger LOGGER = Logger.getLogger(AbstractTransactionalProcessor.class.getName());
	
	static final ProgressMonitor NOP_PROGRESS_MONITOR = new ProgressMonitor() {
		
		@Override
		public void onProgress(int completed, int active, int pending,	String message) {
			// NOP
			
		}
	};
	
	private ProgressMonitorFactory progressMonitorFactory;
	
	public void setProgressMonitorFactory(ProgressMonitorFactory progressMonitorFactory) {
		this.progressMonitorFactory = progressMonitorFactory;
	}

	protected ProgressMonitor createProgressMonitor(String name) {
		if (progressMonitorFactory==null) {
			return NOP_PROGRESS_MONITOR;
		}
		return progressMonitorFactory.createProgressMonitor(name);
	}
		
	private class ExecutionContextImpl<V> implements ExecutionContext<KP>, Callable<V> {
		
		private Undoer undoer;
		private Command<KP, V> command;
		private PropertySet<KP> propertySet;
		private Object[] args;
		private ExecutorService executorService;
		private String resourceIdPrefix;
		private AtomicInteger resourceCounter;
		private Map<String, XAResource> resourceMap;
		private MutableContext context;
		private com.hammurapi.common.concurrent.Logger logger;

		ExecutionContextImpl(
				Undoer undoer, 
				Command<KP,V> command, 
				Context chain,
				PropertySet<KP> propertySet, 
				Object[] args,
				ExecutorService executorService, 
				String resourceIdPrefix, 
				AtomicInteger resourceCounter, 
				Map<String, XAResource> resourceMap) {
			
			this.undoer = undoer;
			this.command = command;
			this.context = AbstractTransactionalProcessor.this.createContext(this, chain);
			this.propertySet = propertySet;
			this.args = args;
			this.executorService = executorService;
			context.register(ExecutorService.class, executorService);
			this.resourceIdPrefix = resourceIdPrefix;
			this.resourceCounter = resourceCounter;
			this.resourceMap = resourceMap;
			
			final Collection<com.hammurapi.common.concurrent.Logger> contextLoggers = new ArrayList<com.hammurapi.common.concurrent.Logger>();
			Iterator<com.hammurapi.common.concurrent.Logger> lit = this.context.lookupAll(com.hammurapi.common.concurrent.Logger.class);
			while (lit.hasNext()) {
				contextLoggers.add(lit.next().getLogger(command.toString()));
			}
			this.logger = new CompositeLogger(contextLoggers.iterator());

		}	
		
		@Override
		public Transaction getTransaction() {
			return getCurrentTransaction();
		}

		@Override
		public Object lookup(String name) {
			return context.lookup(name);
		}

		@Override
		public <T> T lookup(Class<T> serviceClass, Predicate<T, Context>... selectors) {
			return context.lookup(serviceClass, selectors);
		}

		@Override
		public <T> Iterator<T> lookupAll(Class<T> serviceClass, Predicate<T, Context>... selectors) {
			return context.lookupAll(serviceClass, selectors);
		}

		@Override
		public void addUndo(Runnable undoTask) {
			try {
				undoer.addUndoTask(undoTask);
			} catch (XAException e) {
				getLogger().log(Level.SEVERE, "Failed to add undo task: "+e, e);
				try {
					getTransaction().setRollbackOnly();
				} catch (IllegalStateException ise) {
		        	getLogger().log(Level.SEVERE, "Sub-transaction rollback failed: "+ise, ise);
				} catch (SystemException se) {
		        	getLogger().log(Level.SEVERE, "Sub-transaction rollback failed: "+se, se);
				}
			}			
		}

		@Override
		public PropertySet<KP> createPropertySet(PropertySet<KP>... chain) {
			return AbstractTransactionalProcessor.this.createPropertySet(executorService, this, chain);
		}

		@Override
		public V call() throws Exception {
			return command.execute(this, propertySet, args);
		}
		
		@Override
		public <U> Future<U> execute(final Command<KP, U> command, Context context, PropertySet<KP> propertySet, Object... args) {
			ExecutionContextImpl<U> ec = new ExecutionContextImpl<U>(
					undoer, 
					command, 
					context, 
					propertySet, 
					args, 
					executorService, 
					resourceIdPrefix, 
					resourceCounter, 
					resourceMap);
			return lookup(ExecutorService.class).submit(ec);
		}

		@Override
		public void yield() {
			throw new UnsupportedOperationException();			
		}

		@Override
		public ProgressMonitor createProgressMonitor(String name) {
			return AbstractTransactionalProcessor.this.createProgressMonitor(name);
		}

		@Override
		public void registerXAResource(XAResource xaResource) {
			String uniqueName = resourceIdPrefix+"_"+Integer.toString(resourceCounter.incrementAndGet(), Character.MAX_RADIX);
			if (resourceMap.put(uniqueName, xaResource)!=null) {
				throw new ProcessingException("Duplicate resource name: "+uniqueName);
			}
			AbstractTransactionalProcessor.this.registerXAResource(uniqueName, xaResource);			
		}

		@Override
		public MutableContext createContext() {
			return new SimpleMutableContext(this);
		}

		@Override
		public com.hammurapi.common.concurrent.Logger getLogger() {			
			return logger;
		}
		
	}		
		
	protected AbstractTransactionalProcessor(final ExecutorService executorService) {        		
		this.executorService = new AbstractExecutorService() {
			
			@Override
			public void execute(Runnable command) {
				executorService.execute(new TransactionalRunnable(command));				
			}
			
			@Override
			public List<Runnable> shutdownNow() {
				throw new UnsupportedOperationException();
			}
			
			@Override
			public void shutdown() {
				throw new UnsupportedOperationException();
			}
			
			@Override
			public boolean isTerminated() {
				return executorService.isTerminated();
			}
			
			@Override
			public boolean isShutdown() {
				return executorService.isShutdown();
			}
			
			@Override
			public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
				return executorService.awaitTermination(timeout, unit);
			}
		};
	}
		
	private static final AtomicInteger processCounter = new AtomicInteger();
	
	private static final AtomicInteger instanceCounter = new AtomicInteger();
	
	private final int instanceNo = instanceCounter.incrementAndGet();

	/**
	 * Concrete implementations begin transaction and return transaction instance.
	 * @return
	 * @throws NotSupportedException
	 * @throws SystemException
	 */
	protected abstract Transaction beginTransaction() throws NotSupportedException, SystemException;
	
	protected void registerXAResource(String uniqueId, XAResource xaResource) {
		
	}
	
	protected void unregisterXAResource(String uniqueId, XAResource xaResource) {
		
	}

	/**
	 * Transactionally processes the command.
	 * @param command
	 * @return
	 */
	public <V> V process(Command<KP, V> command, Context context, PropertySet<KP> propertySet, Object... args) throws Exception {
		Transaction transaction = beginTransaction();
		threadTransaction.set(transaction);
		
		String resourceIdPrefix = getClass().getName()+"_"+Integer.toString(instanceNo, Character.MAX_RADIX)+"_"+Integer.toString(processCounter.incrementAndGet(), Character.MAX_RADIX);

		Map<String, XAResource> resourceMap = new ConcurrentHashMap<String, XAResource>();
		
		try {		
			com.hammurapi.common.concurrent.Logger logger;
			if (context==null) {
				logger = new LoggerWrapper(LOGGER);
			} else {
				logger = new CompositeLogger(context.lookupAll(com.hammurapi.common.concurrent.Logger.class));
			}
			Undoer undoer = new Undoer(createProgressMonitor("Undoer"), logger.getLogger("Undoer"));
			
			String undoerId = resourceIdPrefix+"_Undoer";
			
			resourceMap.put(undoerId, undoer);
			registerXAResource(undoerId, undoer);
			
			transaction.enlistResource(undoer);
						
			AtomicInteger resourceCounter = new AtomicInteger();		
			
			LocalTrackingExecutorService ltes = new LocalTrackingExecutorService(executorService, false, resourceIdPrefix+"_TrackingExecutorService");
			
			ExecutionContextImpl<V> ec = new ExecutionContextImpl<V>(
					undoer, 
					command, 
					context, 
					propertySet, 
					args, 
					ltes, 
					resourceIdPrefix, 
					resourceCounter, 
					resourceMap);
			
			Future<V> resultFuture = ltes.submit(ec);

			ltes.join();
			// process pending in the loop			
			
			V result = resultFuture.get();
			
			transaction.commit();
			
			return result;
		} catch (Exception e) {
			LOGGER.log(Level.SEVERE, "Exception: "+e, e);
			if (transaction!=null && transaction.getStatus()==Status.STATUS_ACTIVE) {
				try {
					transaction.rollback();
				} catch (Exception ex) {
					LOGGER.log(Level.SEVERE, "Cannot rollback transaction: "+e, e);					
				}
			}
			throw e;
		} finally {
			for (Map.Entry<String, XAResource> entry: resourceMap.entrySet()) {
				unregisterXAResource(entry.getKey(), entry.getValue());
			}
		}									
	}

	/**
	 * Creates mutable context for execution context. Subclasses can override this method to inject specific values into the context, 
	 * e.g. hierarchical logger.
	 * @param owner
	 * @param chain
	 * @return
	 */
	protected MutableContext createContext(ExecutionContext<KP> owner, Context... chain) {
		return new SimpleMutableContext(chain);
	}

}
