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);
        }

}