package com.sun.messaging.jmq.jmsserver.persist.file;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.data.BaseTransaction;
import com.sun.messaging.jmq.jmsserver.data.TransactionState;
import com.sun.messaging.jmq.jmsserver.data.TransactionUID;
import com.sun.messaging.jmq.jmsserver.data.TransactionWork;
import com.sun.messaging.jmq.jmsserver.data.TransactionWorkMessage;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;
import com.sun.messaging.jmq.util.log.Logger;

public abstract class BaseTransactionManager {

	public static final Logger logger = Globals.getLogger();

	PreparedTxnStore preparedTxnStore;
	TransactionLogManager transactionLogManager;

	Set<TransactionUID> playingToMessageStore = Collections.synchronizedSet(new HashSet<TransactionUID>());
	Map<TransactionUID, BaseTransaction> completeStored = new Hashtable<TransactionUID, BaseTransaction>();
	Map<TransactionUID, BaseTransaction> incompleteUnstored = new Hashtable<TransactionUID, BaseTransaction>();
	Map<TransactionUID, BaseTransaction> incompleteStored = new Hashtable<TransactionUID, BaseTransaction>();

	BaseTransactionManager(TransactionLogManager transactionLogManager) {
		this.transactionLogManager = transactionLogManager;
		preparedTxnStore = transactionLogManager.preparedTxnStore;
	}

	abstract String getPrefix();

	public List<BaseTransaction> getAllIncompleteTransactions() {
		List<BaseTransaction> result = new ArrayList<BaseTransaction>();
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " getAllIncompleteTransactions  "
					+ " num incompleteUnstored = " + incompleteUnstored.size()
					+ " num incompleteStored = " + incompleteStored.size();
			logger.log(Logger.DEBUG, msg);
		}
		result.addAll(incompleteUnstored.values());
		result.addAll(incompleteStored.values());
		return result;
	}
	
	public HashMap getAllTransactionsMap() {
		List<BaseTransaction> txns = getAllIncompleteTransactions();
		HashMap map = new HashMap(txns.size());
		Iterator<BaseTransaction> itr = txns.iterator();
		while (itr.hasNext()) {
			BaseTransaction txn = itr.next();
			TransactionState txnState = new TransactionState(txn
					.getTransactionState());
			map.put(txn.getTid(), txnState);
		}
		return map;
	}

	public void addToCompleteStored(BaseTransaction baseTxn) {
		completeStored.put(baseTxn.getTid(), baseTxn);
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " adding  " + baseTxn.getTid()
					+ "  to completeStored. Total = " + completeStored.size();
			logger.log(Logger.DEBUG, msg);
		}
	}

	public void removeFromCompleteStored(TransactionUID tid) {
		completeStored.remove(tid);
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " removing " + tid
					+ " from completeStored. Total = " + completeStored.size();
			logger.log(Logger.DEBUG, msg);
		}
	}

	void addToIncompleteStored(BaseTransaction baseTxn) {
		incompleteStored.put(baseTxn.getTid(), baseTxn);
		if (Store.getDEBUG()) {
			String msg = getPrefix()

			+ " adding  " + baseTxn.getTid()
					+ "  to incompleteStored. Total = "
					+ incompleteStored.size();
			logger.log(Logger.DEBUG, msg);
		}
	}

	BaseTransaction removeFromIncompleteStored(TransactionUID tid) {
		BaseTransaction result = incompleteStored.remove(tid);
		if (Store.getDEBUG()) {
			String msg = getPrefix()

			+ " removing  " + tid + " from incompleteStored. Total = "
					+ incompleteStored.size();
			logger.log(Logger.DEBUG, msg);
		}
		return result;
	}

	void addToIncompleteUnstored(BaseTransaction baseTxn) {
		incompleteUnstored.put(baseTxn.getTid(), baseTxn);
		if (Store.getDEBUG()) {
			String msg = getPrefix()

			+ " adding  " + baseTxn.getTid()
					+ "  to incompleteUnstored. Total = "
					+ incompleteUnstored.size();
			logger.log(Logger.DEBUG, msg);
		}
	}

	BaseTransaction removeFromIncompleteUnstored(TransactionUID tid) {
		BaseTransaction result = incompleteUnstored.remove(tid);
		if (Store.getDEBUG()) {
			String msg = getPrefix()

			+ " removing  " + tid + " from incompleteUnstored. Total = "
					+ incompleteUnstored.size();
			logger.log(Logger.DEBUG, msg);
		}
		return result;
	}

	abstract void processStoredTxnOnStartup(BaseTransaction baseTxn);

	public void waitForPlayingToMessageStoreCompletion() {
		synchronized (playingToMessageStore) {
			if (Store.getDEBUG()) {
				String msg = getPrefix() + " num playingToMessageStore ="
						+ playingToMessageStore.size();
				logger.log(Logger.DEBUG, msg);
			}
			try {
				while (playingToMessageStore.size() > 0) {
					if (Store.getDEBUG()) {
						String msg = getPrefix() + " waiting for "
								+ playingToMessageStore.size()
								+ " playingToMessageStore";
						logger.log(Logger.DEBUG, msg);
					}
					playingToMessageStore.wait(1000);
				}
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

	/*
	 * this method is called by transactionHandler to notify that 
	 * the committed transaction has been written to the message store.
	 * 
	 * Keeping track of committed transactions and when they have been written to the message store 
	 * allows the TransactionlogManager to know when it is safe to do a checkpoint. 
	 */
	public void playingToMessageStoreComplete(TransactionUID tid) {
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " playingToMessageStoreComplete " + tid;
			logger.log(Logger.DEBUG, msg);
		}
		synchronized (playingToMessageStore) {
			Object found = playingToMessageStore.remove(tid);
			if (found == null) {
				String msg = getPrefix()
						+ " playingToMessageStoreComplete(): could not find "
						+ tid;
				logger.log(Logger.WARNING, msg);
			}
			playingToMessageStore.notify();
		}
		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " remove transaction from  playingToMessageStore."
					+ " tid=" + tid + " size = " + playingToMessageStore.size();
			logger.log(Logger.DEBUG, msg);
		}
	}

	// on checkpoint, we need to store any newly logged prepared transactions to the preparedTxnStore
	// as the transaction log is about to be reset
	void writePreparedTransactionsToPreparedTxnStoreOnCheckpoint()
			throws BrokerException {
		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " writePreparedTransactionsToPreparedTxnStoreOnCheckpoint"
					+ " num incompleteUnstored=" + incompleteUnstored.size();
			logger.log(Logger.DEBUG, msg);
		}
		Iterator<BaseTransaction> iter = incompleteUnstored.values().iterator();
		while (iter.hasNext()) {
			BaseTransaction baseTxn = iter.next();
			if (!preparedTxnStore.containsTransaction(baseTxn.getTid())) {
				if (Store.getDEBUG()) {
					String msg = getPrefix()
							+ " transaction storing preparedTransaction "
							+ baseTxn;
					logger.log(Logger.DEBUG, msg);
				}
				try {					
					preparedTxnStore.storeTransaction(baseTxn, true);
					addToIncompleteStored(baseTxn);
				} catch (IOException ioe) {
					throw new BrokerException(
							"failed to store transaction in preparedTxnStore "
									+ baseTxn, ioe);
				}
			} else {

				
					String msg = getPrefix()
							+ " transaction already exists in preparedTxnStore "
							+ baseTxn;
					logger.log(Logger.INFO, msg);
				
			}
		}
		incompleteUnstored.clear();
	}

	void removeCompleteTransactionsAfterCheckpoint() throws BrokerException {
		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " removeCompleteTransactionsAfterCheckpoint"
					+ "num completeStored=" + completeStored.size();
			logger.log(Logger.DEBUG, msg);
		}

		Iterator<TransactionUID> iter = completeStored.keySet().iterator();
		while (iter.hasNext()) {
			TransactionUID tid = iter.next();
			if (preparedTxnStore.containsTransaction(tid)) {
				preparedTxnStore.removeTransaction(tid, true);

				if (Store.getDEBUG()) {
					String msg = getPrefix() + " removed transaction " + tid;
					logger.log(Logger.DEBUG, msg);
				}
			} else {

				if (Store.getDEBUG()) {
					String msg = getPrefix()
							+ " Could not find transaction in preparedTxnStore "
							+ tid;
					logger.log(Logger.DEBUG, msg);
				}
			}
		}
		completeStored.clear();

	}
	
	void rollbackAllTransactions() {
		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " rollbackAllTransactions"
					+ " num incompleteStored=" + incompleteStored.size()
					+ " num incompleteUnstored=" + incompleteUnstored.size();
			logger.log(Logger.DEBUG, msg);
		}
		Collection<BaseTransaction> storedTxns = incompleteStored.values();
		Collection<BaseTransaction> storedTxnsCopy = new ArrayList<BaseTransaction>(
				storedTxns);
		rollbackTransactions(storedTxnsCopy);
		
		Collection<BaseTransaction> unstoredTxns = incompleteUnstored.values();
		Collection<BaseTransaction> unstoredTxnsCopy = new ArrayList<BaseTransaction>(
				unstoredTxns);
		rollbackTransactions(unstoredTxnsCopy);

	}
	
	void rollbackTransactions(Collection<BaseTransaction> txns) {
		Iterator<BaseTransaction> iter = txns.iterator();
		while (iter.hasNext()) {
			BaseTransaction txn = iter.next();
			try {
				processTxnCompletion(txn.getTid(), TransactionState.ROLLEDBACK,
						true);
			} catch (IOException ioe) {
				logger.log(Logger.ERROR, "could not rollback " + txn, ioe);

			} catch (BrokerException be) {
				logger.log(Logger.ERROR, "could not rollback " + txn, be);
			}

		}

	}

	abstract void processTxn(BaseTransaction baseTxn) throws IOException,
			BrokerException;

	abstract TransactionEvent generateEvent(BaseTransaction baseTxn, boolean completion)
			throws IOException, BrokerException;

	abstract BaseTransaction processTxnCompletion(TransactionUID tid, int state)
			throws IOException, BrokerException;

	BaseTransaction processTxnCompletion(TransactionUID tid, int state,
			boolean fullyComplete) throws IOException, BrokerException {

		// We are committing a prepared entry.
		// Check if it is in the prepared transaction store
		// If it is mark it as committed, so that it can be cleaned up after next checkpoint 
		if(state==TransactionState.COMMITTED)
		{
			this.playingToMessageStore.add(tid);
			if (Store.getDEBUG()) {
				String msg = getPrefix()
						+ " add transaction to  playingToMessageStore."
						+ " tid=" + tid + " size = " + playingToMessageStore.size();
				logger.log(Logger.DEBUG, msg);
			}
		}
		
		BaseTransaction existingTxn = null;
		boolean stored = false;
		existingTxn = incompleteUnstored.get(tid);
		if (existingTxn == null) {
			existingTxn = incompleteStored.get(tid);
			if (existingTxn != null) {
				stored = true;
			} else {
				String msg = getPrefix() + " processTxnCompletion: Could not find txn for " + tid;
				logger.log(Logger.WARNING, msg);
				throw new BrokerException(msg);
			}
		}
		existingTxn.getTransactionDetails().setState(state);

		if (stored) {
			// update the stored transaction from prepared to committed/rolledback state			
			//update state on file by writing at fixed offset

			updateStoredState(tid, state);

			if (fullyComplete) {
				incompleteStored.remove(tid);
				completeStored.put(tid, existingTxn);
			}

		} else {
			if (fullyComplete) {
				removeFromIncompleteUnstored(tid);
			}
		}
		messageListLogged(existingTxn);
		

		return existingTxn;

	}
	
	protected void messageListLogged(BaseTransaction baseTxn) {
		TransactionWork txnWork = baseTxn.getTransactionWork();
		if (txnWork != null) {
			List<TransactionWorkMessage> sentMessages = txnWork
					.getSentMessages();
			if (sentMessages != null) {
				transactionLogManager.loggedMessageHelper
						.messageListLogged(sentMessages);
			}
		}
	}
	

	void updateStoredState(TransactionUID tid, int state) throws IOException,
			BrokerException {
		preparedTxnStore.updateTransactionState(tid, state, true);

		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " updated transaction state of stored prepared transaction to "
					+ TransactionState.toString(state);
			logger.log(Logger.DEBUG, msg);
		}
	}

	void updateStoredCompletion(TransactionUID tid, boolean complete)
			throws IOException, BrokerException {
		preparedTxnStore.updateTransactionCompletion(tid, complete, true);

		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " updated completion status of stored transaction to "
					+ complete;
			logger.log(Logger.DEBUG, msg);
		}
	}

	abstract void replayTransactionEvent(TransactionEvent localTxnEvent,
			HashSet dstLoadedSet) throws BrokerException, IOException;

}
