/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright (c) 2010 Oracle and/or its affiliates. All rights reserved.
 *
 * The contents of this file are subject to the terms of either the GNU
 * General Public License Version 2 only ("GPL") or the Common Development
 * and Distribution License("CDDL") (collectively, the "License").  You
 * may not use this file except in compliance with the License.  You can
 * obtain a copy of the License at
 * https://glassfish.dev.java.net/public/CDDL+GPL_1_1.html
 * or packager/legal/LICENSE.txt.  See the License for the specific
 * language governing permissions and limitations under the License.
 *
 * When distributing the software, include this License Header Notice in each
 * file and include the License file at packager/legal/LICENSE.txt.
 *
 * GPL Classpath Exception:
 * Oracle designates this particular file as subject to the "Classpath"
 * exception as provided by Oracle in the GPL Version 2 section of the License
 * file that accompanied this code.
 *
 * Modifications:
 * If applicable, add the following below the License Header, with the fields
 * enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyright [year] [name of copyright owner]"
 *
 * Contributor(s):
 * If you wish your version of this file to be governed by only the CDDL or
 * only the GPL Version 2, indicate your decision by adding "[Contributor]
 * elects to include this software in this distribution under the [CDDL or GPL
 * Version 2] license."  If you don't indicate a single choice of license, a
 * recipient has the option to distribute your version of this file under
 * either the CDDL, the GPL Version 2 or to extend the choice of license to
 * its licensees as provided above.  However, if you add GPL Version 2 code
 * and therefore, elected the GPL Version 2 license, then the option applies
 * only if the new code is made subject to such option by the copyright
 * holder.
 */

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

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import com.sun.messaging.jmq.io.SysMessageID;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.config.BrokerConfig;
import com.sun.messaging.jmq.jmsserver.core.Destination;
import com.sun.messaging.jmq.jmsserver.core.DestinationUID;
import com.sun.messaging.jmq.jmsserver.core.Subscription;
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.TransactionWorkMessage;
import com.sun.messaging.jmq.jmsserver.data.TransactionWorkMessageAck;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.persist.TransactionInfo;
import com.sun.messaging.jmq.jmsserver.resources.BrokerResources;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;
import com.sun.messaging.jmq.util.FileUtil;
import com.sun.messaging.jmq.util.SizeString;
import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.util.txnlog.CheckPointListener;
import com.sun.messaging.jmq.util.txnlog.TransactionLogRecord;
import com.sun.messaging.jmq.util.txnlog.TransactionLogWriter;
import com.sun.messaging.jmq.util.txnlog.file.FileTransactionLogWriter;

/**
 * @author gsivewright
 * 
 * This class manages the transfer of data between the message store and the
 * transaction log.
 * 
 */

/*
 * Order of handling prepared txns
 * 
 * The following scenario describes a sequence of updates to the transaction log
 * and message store. This sequence is used to illustrate how consistent state
 * can be recreated on restart after a failure of the broker at various specific
 * processing points.
 * 
 * 1.write prepared txn to txnLog 2.checkpoint1 (see checkpoint sequence)
 * 3.write commit record to txnLog 4.write txn to store 5.write commit record to
 * preparedTxnStore 6.checkpoint2 (see checkpoint sequence)
 * 
 * checkpoint sequence:
 * 
 * 1.sync message store 2.write prepared txns (since last checkpoint) to
 * prepared store 3.sync prepareTxnStore 4.reset txn log 5.remove committed txns
 * (since last checkpoint) from prepared store
 * 
 * 
 * replay after failure at 1 �> 2.2
 * 
 * 1.read preparedTxnStore (empty) 2.read prepared txn from txnLog 3.check if
 * txn is in preparedTxnStore. (add). 4.checkpoint
 * 
 * replay after failure at 2.3 -> 2.4
 * 
 * 1.read txns from preparedTxnStore 2.read prepared txn from txnLog 3.check if
 * txn is in preparedTxnStore. (already there) 4.checkpoint
 * 
 * replay after failure at 2.5
 * 
 * 1.read preparedTxn from preparedTxnStore 2.checkpoint
 * 
 * replay after failure at 3 -> 5
 * 
 * 1.read txns from preparedTxnStore 2.read commit record from txnLog 3.write
 * txn to message store (replace if there) 4.write commit record to
 * preparedTxnStore (if not committed) 5.checkpoint
 * 
 * replay after failure at 6.4 1)read txns from preparedTxnStore (find committed
 * txn) 2)do not replay (not in reset txnLog) 3)checkpoint
 * 
 */

public class TransactionLogManager implements CheckPointListener {

	static final String TXNLOG_PROP_PREFIX = Globals.IMQ
			+ ".persist.file.txnLog";
	static final String TXNLOG_FILE_SIZE_PROP = TXNLOG_PROP_PREFIX
			+ ".file.size";
	

	static final long DEFAULT_TXNLOG_FILE_SIZE = FileTransactionLogWriter.DEFAULT_MAX_SIZE;

	static final String MSG_LOG_FILENAME = "txnlog";

	
	static final String INCOMPLETE_TXN_STORE = "incompleteTxnStore";
	public static final BrokerResources br = Globals.getBrokerResources();
	Store store;
	MsgStore msgStore;
	File rootDir;	
	
	 // whether non-transacted persistent message sent should be logged
	// by default, log non transacted message sends and acks if sync is enabled.
	// user can override this behavior by setiing properties explicitly.
	private static boolean defaultLogNonTransactedMsgSend = Destination.PERSIST_SYNC;
	public static final String LOG_NON_TRANSACTED_MSG_SEND_PROP = Globals.IMQ
			+ ".persist.file.txnLog.logNonTransactedMsgSend";
	public static boolean logNonTransactedMsgSend = Globals.getConfig()
			.getBooleanProperty(LOG_NON_TRANSACTED_MSG_SEND_PROP,
					defaultLogNonTransactedMsgSend);

	private static boolean defaultLogNonTransactedMsgAck = Destination.PERSIST_SYNC;
	public static final String LOG_NON_TRANSACTED_MSG_ACK_PROP = Globals.IMQ
			+ ".persist.file.txnLog.logNonTransactedMsgAck";
	public static boolean logNonTransactedMsgAck = Globals.getConfig()
			.getBooleanProperty(LOG_NON_TRANSACTED_MSG_ACK_PROP,
					defaultLogNonTransactedMsgAck);
 
    /**
     * whether to use a separate thread for writing to the txn log
     */
   
    public static final String TXN_LOG_GROUP_COMMITS_PROP = Globals.IMQ
			+ ".persist.file.txnLog.groupCommits";

	public static boolean isTxnLogGroupCommits = Globals.getConfig()
			.getBooleanProperty(TXN_LOG_GROUP_COMMITS_PROP, false);

      
    private static boolean replayInProgress=false;

	private TransactionLogWriter msgLogWriter = null;
	
	TransactionLogReplayer transactionLogReplayer;
	
	LocalTransactionManager localTransactionManager;
	ClusterTransactionManager clusterTransactionManager;
	RemoteTransactionManager remoteTransactionManager;
	
	LoggedMessageHelper loggedMessageHelper;

	CheckpointManager checkpointManager;
	PreparedTxnStore preparedTxnStore;
	
		public static final Logger logger = Globals.getLogger();
		


	public TransactionLogManager(Store store, MsgStore msgStore, File rootDir, boolean resetStore) throws BrokerException{
		
		this.rootDir = rootDir;
		this.store = store;
		File preparedTxnStoreDir = new File(rootDir, INCOMPLETE_TXN_STORE);
		if (resetStore) {
			clearPreparedTxnStore(preparedTxnStoreDir);
			resetTransactionLogOnStartUp();
		}

		preparedTxnStore = new PreparedTxnStore(msgStore, preparedTxnStoreDir,
				false);

		this.msgStore = msgStore;
		
		this.transactionLogReplayer = new TransactionLogReplayer(msgStore);
		this.localTransactionManager = new LocalTransactionManager(this);
		this.clusterTransactionManager = new ClusterTransactionManager(this);
		this.remoteTransactionManager = new RemoteTransactionManager(this);
		this.checkpointManager = new CheckpointManager(this);
		this.loggedMessageHelper = new LoggedMessageHelper(this);
		
	}
	
	 

	public void startup() throws BrokerException {
		// These are the stages on broker startup
		// 1) read in any prepared transactions held in preparedTxnStore.
		// 2) check if the transaction log needs replaying.
		// 3) replay committed transaction to message store
		// 4) after replay is complete, add any remaining transactions to
		// preparedTxnStore.
		// 5) sync preparedTxnStore
		// 6) sync message store.
		// 7) reset transaction log.
		// 8) remove any committed transactions from preparedTxnStore

		if (Store.getDEBUG()) {
			logger.log(Logger.DEBUG, getPrefix() + " startup");
		}
		// 1 read in any prepared transactions held in preparedTxnStore.
		processStoredTxnsOnStartup();

		// 2) check if the transaction log needs replaying.
		initTransactionLogOnStartUp();

		// 3) replay committed transaction to message store
		replayTransactionLogOnStartup();

	
		// 4. write prepared transactions to prepared transaction store
		localTransactionManager
				.writePreparedTransactionsToPreparedTxnStoreOnCheckpoint();
		clusterTransactionManager
				.writePreparedTransactionsToPreparedTxnStoreOnCheckpoint();
		remoteTransactionManager
				.writePreparedTransactionsToPreparedTxnStoreOnCheckpoint();

		// 5. sync prepareTxnStore
		preparedTxnStore.sync();

		// 6) checkpoint message store.
		store.syncDestination(null);

		try {
			// 7) reset transaction log.
			msgLogWriter.reset();
		} catch (IOException e) {
			throw new BrokerException("failed to reset transaction log", e);
		}

		// 8) remove any committed transactions from preparedTxnStore
		removeCommittedTransactionsOnStartup();

	}
	
	public void close()
	{
		try{
			if(msgLogWriter!=null)
				msgLogWriter.close(false);
			if(preparedTxnStore!=null)
				preparedTxnStore.close(true);
		}
		catch(IOException e)
		{
			logger.logStack(Logger.ERROR, "caught exception closing", e);
		}
	}
	
	public static void deleteAllFileState(File rootDir) throws BrokerException {
		// delete txnLog file first
		// as deleting directory will not be atomic, and will probably be bottom
		// up
		// (harder to detect if some elements are missing)
		logger.log(Logger.DEBUG, "deleteAllFileState "+rootDir);
		File txnLogFile = new File(rootDir, MSG_LOG_FILENAME);
		boolean deleted = false;
		if (txnLogFile.exists()) {
			deleted = txnLogFile.delete();
			if (!deleted) {
				throw new BrokerException("Could not delete txnLog file "
						+ txnLogFile);
			}
		}
		File incompleteTxnStore = new File(rootDir, INCOMPLETE_TXN_STORE);
		try {
			if (incompleteTxnStore.exists())
				FileUtil.removeFiles(incompleteTxnStore, true);
		} catch (IOException e) {
			String msg = "Can not delete incomplete txn store "
					+ incompleteTxnStore;
			logger.log(Logger.ERROR, msg, e);
			throw new BrokerException(msg, e);
		}

	}
	
	
	public static boolean txnLogExists(File rootDir) {
		File txnLogFile = new File(rootDir, MSG_LOG_FILENAME);
		return txnLogFile.exists();
	}

	public static boolean incompleteTxnStoreExists(File rootDir) {
		File file = new File(rootDir, INCOMPLETE_TXN_STORE);
		return file.exists();
	}

	public static void assertAllFilesExist(File rootDir) throws BrokerException {

		if (!txnLogExists(rootDir)) {
			throw new BrokerException("assertion failure: " + MSG_LOG_FILENAME
					+ " file does not exist");
		}
		if (!incompleteTxnStoreExists(rootDir)) {
			throw new BrokerException("assertion failure: "
					+ INCOMPLETE_TXN_STORE + " file does not exist");
		}
	}
	
	
/*	
	List<BaseTransaction> getAllIncompleteTransactions()
	{
		List<BaseTransaction> local = localTransactionManager.getAllIncompleteTransactions();
		List<BaseTransaction> cluster = clusterTransactionManager.getAllIncompleteTransactions();
		List<BaseTransaction> remote = remoteTransactionManager.getAllIncompleteTransactions();
		
		List<BaseTransaction> all = new ArrayList<BaseTransaction>();
		all.addAll(local);
		all.addAll(cluster);
		all.addAll(remote);
		return all;
	}
	
*/
	
	
	
	
	HashMap getAllTransactionStates() throws IOException {
		
		HashMap localTxnMap = localTransactionManager.getAllTransactionsMap();
		HashMap clusterTxnMap = clusterTransactionManager.getAllTransactionsMap();
		
		HashMap txnMap = new HashMap(localTxnMap.size()+clusterTxnMap.size());
		txnMap.putAll(localTxnMap);
		txnMap.putAll(clusterTxnMap);
		
		return txnMap;
	}
	
	void rollbackAllTransactions()
	{
		localTransactionManager.rollbackAllTransactions();
		clusterTransactionManager.rollbackAllTransactions();
		remoteTransactionManager.rollbackAllTransactions();
		
	}
	

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

		preparedTxnStore.loadTransactions();
				
		// for now just read and sort into different types
		Enumeration<BaseTransaction> transactions = preparedTxnStore
				.txnEnumeration();
		while (transactions.hasMoreElements()) {
			BaseTransaction baseTxn = transactions.nextElement();
			// need to switch on all types
			int type = baseTxn.getType();
			BaseTransactionManager baseTxnMan = this
					.getTransactionManager(type);
			baseTxnMan.processStoredTxnOnStartup(baseTxn);
		}

	}

	void removeCommittedTransactionsOnStartup() throws BrokerException {

		if (Store.getDEBUG()) {
			String msg = getPrefix()
					+ " removeCommitedTransactionsInPreparedTxnStore";
			logger.log(Logger.DEBUG, msg);
		}
		// for now just report on any transactions
		Enumeration<BaseTransaction> transactions = preparedTxnStore
				.txnEnumeration();
		List<BaseTransaction> committed = new ArrayList<BaseTransaction>();
		while (transactions.hasMoreElements()) {
			BaseTransaction baseTxn = transactions.nextElement();
			if (baseTxn.getState() == TransactionState.COMMITTED) {
				if (baseTxn.getType() == BaseTransaction.CLUSTER_TRANSACTION_TYPE
						&& !baseTxn.getTransactionDetails().isComplete()) {
					if (Store.getDEBUG()) {
						String msg = getPrefix()
								+ " not removing incomplete cluster transaction "
								+ baseTxn.getTransactionDetails();
						logger.log(Logger.DEBUG, msg);
					}
				} else {
					if (Store.getDEBUG()) {
						String msg = getPrefix() + " removing transaction "
								+ baseTxn.getTransactionDetails();
						logger.log(Logger.DEBUG, msg);
					}
					committed.add(baseTxn);
				}
			}

		}
		Iterator<BaseTransaction> iter = committed.iterator();
		while (iter.hasNext()) {
			BaseTransaction baseTxn = iter.next();
			TransactionUID tid = baseTxn.getTid();
			preparedTxnStore.removeTransaction(tid, true);
			if (Store.getDEBUG()) {
				String msg = getPrefix()
						+ " removed committed transaction from preparedTxnStore. Tid="
						+ tid;
				logger.log(Logger.DEBUG, msg);
			}
		}

	}
	private void resetTransactionLogOnStartUp() {
		
		String filename = MSG_LOG_FILENAME;
		File file = new File(rootDir, filename);
		logger.log(Logger.INFO, "resetting txn Log file "+file);
		if(!file.exists())
		{
			logger.log(Logger.INFO, "nothing to reset. txn Log file "+file+ " does not exist");
			return;
		}
		boolean deleted = file.delete();
		if (!deleted) {
			String msg = getPrefix() + " could not delete " + file;

			logger.log(Logger.DEBUG, msg);
		}
	}

	private void clearPreparedTxnStore(File preparedTxnStoreDir)
			throws BrokerException {
		// delete all files under the preparedTxnStoreDir directory
		try {
			
			FileUtil.removeFiles(preparedTxnStoreDir, false);
		} catch (IOException e) {
			logger.log(Logger.ERROR, BrokerResources.X_RESET_MESSAGES_FAILED,
					preparedTxnStoreDir, e);
			throw new BrokerException(br.getString(BrokerResources.X_RESET_MESSAGES_FAILED,
					preparedTxnStoreDir), e);
		}
	}
	
	public static boolean transactionLogExists(File rootDir) {
		boolean result = false;
		String filename = MSG_LOG_FILENAME;
		File txnLogFile = new File(rootDir, filename);
		result = txnLogFile.exists();
		return result;
	}

	private void initTransactionLogOnStartUp() throws BrokerException {

		if (Store.getDEBUG()) {
			String msg = getPrefix() + " initTransactionLogOnStartUp";
			logger.log(Logger.DEBUG, msg);
		}
		logger.log(Logger.INFO, "new transaction log enabled");
		logger.log(Logger.INFO, "sync writes to disk = "+ Destination.PERSIST_SYNC);
		logger.log(Logger.INFO, "logNonTransactedMsgSend = "+ logNonTransactedMsgSend);
		logger.log(Logger.INFO, "logNonTransactedMsgAck = "+ logNonTransactedMsgAck);
		

		// create txn log writers
		String filename = null;
		try {
			BrokerConfig config = Globals.getConfig();
			SizeString filesize = config.getSizeProperty(TXNLOG_FILE_SIZE_PROP,
					DEFAULT_TXNLOG_FILE_SIZE);
					
			filename = MSG_LOG_FILENAME;

			String mode = "rwd";
			boolean synch = true;
			if (!Destination.PERSIST_SYNC) {
				mode = "rw";
				synch = false;
			}
			logger.log(Logger.INFO, br.getKString(BrokerResources.I_OPEN_TXNLOG,
						mode, Long.valueOf(filesize.getBytes())));

			FileTransactionLogWriter ftlw = new FileTransactionLogWriter(
					rootDir, filename, filesize.getBytes(), mode, synch,
					isTxnLogGroupCommits,
					BaseTransaction.CURRENT_FORMAT_VERSION);
			long existingFormatVersion = ftlw.getExistingAppCookie();
			
			// 
			// Check version here
			// if(existingFormatVersion!=BaseTransaction.CURRENT_FORMAT_VERSION) 
			// then file may need to be converted.
			//
			// Note this appCookie format specifies the format of the transactions 
			// being stored and not the format of the transaction log itself, which may 
			// change less frequently and should be checked separately
			// Note also that transaction format is also dependent on any changes to packet format etc.
			// 
			// No need to do convert yet as we are on the first version
			// so this is just a sanity check.
			if(existingFormatVersion!=BaseTransaction.CURRENT_FORMAT_VERSION)
			{
				throw new BrokerException("Unexpected transaction log format. Format on file = "+existingFormatVersion + 
						" Current software version = "+BaseTransaction.CURRENT_FORMAT_VERSION);
			}
			
			msgLogWriter = ftlw;
			
			msgLogWriter.setCheckPointListener(this);
			if (Store.getDEBUG()) {
				logger.log(Logger.DEBUG, "created txn log");
			}
		} catch (IOException ex) {
			logger.logStack(Logger.ERROR,
					BrokerResources.E_CREATE_TXNLOG_FILE_FAILED, filename, ex);
			throw new BrokerException(br.getString(
					BrokerResources.E_CREATE_TXNLOG_FILE_FAILED, filename), ex);
		}

	}
	
	

	public void replayTransactionLogOnStartup() throws BrokerException {

		if (Store.getDEBUG()) {
			logger.log(Logger.DEBUG, getPrefix()
					+ " replayTransactionLogOnStartup");
		}

		try {
			setReplayInProgress(true);			
			if (msgLogWriter.playBackRequired()) {
				if (Store.getDEBUG()) {
					String msg = getPrefix()
							+ " replayTransactionLogOnStartup: playBackRequired";
					logger.log(Logger.DEBUG, msg);
				}

				logger.log(Logger.FORCE, BrokerResources.I_PROCESS_MSG_TXNLOG);

				// All destinations need to be loaded
				Destination.init();
				Subscription.initSubscriptions();
				HashSet dstLoadedSet = new HashSet(); // Keep track of loaded
				// dst

				int count = 0;

				
				Iterator itr = msgLogWriter.iterator();
				while (itr.hasNext()) {
					count++; // Keep track the number of records processed

					// Read in the acks or msgs & acks
					TransactionLogRecord rec = (TransactionLogRecord) itr
							.next();

					byte[] data = rec.getBody();
					TransactionEvent txnEvent = readTransactionEvent(data);
					int type = txnEvent.getType();
					if (Store.getDEBUG()) {
						String msg = getPrefix()
								+ " replayTransactionLogOnStartup() recordSeq= " + rec.getSequence()+ " txnEvent= "
								+ txnEvent;
						logger.log(Logger.DEBUG, msg);
					}
					if (type == BaseTransaction.NON_TRANSACTED_MSG_TYPE) {
						transactionLogReplayer.replayNonTxnMsg((NonTransactedMsgEvent)txnEvent,dstLoadedSet);
					} else if (type == BaseTransaction.NON_TRANSACTED_ACK_TYPE) {
						transactionLogReplayer.replayNonTxnMsgAck((NonTransactedMsgAckEvent)txnEvent,dstLoadedSet);
					} else if (type == BaseTransaction.MSG_REMOVAL_TYPE) {
						transactionLogReplayer.replayMessageRemoval((MsgRemovalEvent)txnEvent,dstLoadedSet);
					} else {

						BaseTransactionManager tm = getTransactionManager(type);
						tm.replayTransactionEvent(txnEvent, dstLoadedSet);
					}

				}

			} else {
				if (Store.getDEBUG()) {
					logger.log(Logger.DEBUG, "no playBackRequired");
				}
			}
		}

		catch (IOException e) {
			logger.log(Logger.ERROR, "exception in playback",e);
			throw new BrokerException("exception in playback",e);

		}finally{
			setReplayInProgress(false);
		}
	}

	

	TransactionEvent readTransactionEvent(byte[] data) throws IOException,
			BrokerException {
		TransactionEvent event = TransactionEvent.createFromBytes(data);
		return event;
	}

	
	public void loggedCommitWrittenToMessageStore(TransactionUID tid, int type) {
		BaseTransactionManager txnMan = getTransactionManager(type);

		txnMan.playingToMessageStoreComplete(tid);

	}

	BaseTransactionManager getTransactionManager(int type) {
		BaseTransactionManager r = null;
		switch (type) {
		case BaseTransaction.LOCAL_TRANSACTION_TYPE:
			r = localTransactionManager;
			break;
		case BaseTransaction.CLUSTER_TRANSACTION_TYPE:
			r = clusterTransactionManager;
			break;
		case BaseTransaction.REMOTE_TRANSACTION_TYPE:
			r = remoteTransactionManager;
			break;
		case BaseTransaction.UNDEFINED_TRANSACTION_TYPE:
			throw new UnsupportedOperationException(
					"UNDEFINED_TRANSACTION_TYPE");
		default:
			throw new UnsupportedOperationException("unknown type:" + type);
		}
		return r;
	}
	
	
	public void logMsgRemoval(DestinationUID dstID, SysMessageID mid)
			throws BrokerException {
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " logMsgRemoval() dstID=" + dstID + " mid="+mid;
			logger.log(Logger.DEBUG, msg);
		}
		try {
			store.txnLogSharedLock.lock();

			MsgRemovalEvent txnEvent = new MsgRemovalEvent(dstID,mid);
			byte[] data = txnEvent.writeToBytes();

			TransactionLogRecord record = msgLogWriter
					.newTransactionLogRecord();
			record.setBody(data);
			msgLogWriter.write(record);

		} catch (IOException ioe) {
			throw new BrokerException("error logging transaction", ioe);
		} finally {
			store.txnLogSharedLock.unlock();
		}
	}
	
	
	public void logNonTxnMessage(TransactionWorkMessage twm)
			throws BrokerException {
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " logNonTxnMessage() " + twm;
			logger.log(Logger.DEBUG, msg);
		}
		try {
			store.txnLogSharedLock.lock();

			NonTransactedMsgEvent txnEvent = new NonTransactedMsgEvent(twm);
			writeTransactionEvent(txnEvent);
			loggedMessageHelper.messageLogged(twm);
		} finally {
			store.txnLogSharedLock.unlock();
		}
	}
	
	
	public void logNonTxnMessageAck(TransactionWorkMessageAck twma)
			throws BrokerException {
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " logNonTxnMessageAck() " + twma;
			logger.log(Logger.DEBUG, msg);
		}
		try {
			store.txnLogSharedLock.lock();
			NonTransactedMsgAckEvent txnEvent = new NonTransactedMsgAckEvent(twma);
			writeTransactionEvent(txnEvent);					
		} finally {
			store.txnLogSharedLock.unlock();
		}
	}

	public void logTxn(BaseTransaction baseTxn) throws BrokerException {
		if (Store.getDEBUG()) {
			logger.log(Logger.DEBUG, getPrefix() + " logTxn() " + baseTxn);
		}

		try {
			store.txnLogSharedLock.lock();
			int type = baseTxn.getType();
			BaseTransactionManager txnManager = getTransactionManager(type);
			txnManager.processTxn(baseTxn);
			TransactionEvent txnEvent = txnManager
					.generateEvent(baseTxn, false);
			writeTransactionEvent(txnEvent);
			
		} catch (IOException ioe) {
			throw new BrokerException("error logging transaction", ioe);
		} finally {
			store.txnLogSharedLock.unlock();
		}
	}

	public void logTxnCompletion(TransactionUID tid, int completionState,
			int type) throws BrokerException {
		if (Store.getDEBUG()) {
			logger
					.log(Logger.DEBUG, getPrefix() + " logTxnCompletion() "
							+ tid);
		}
		try {
			store.txnLogSharedLock.lock();
			BaseTransactionManager txnMan = getTransactionManager(type);
			BaseTransaction existing = txnMan.processTxnCompletion(tid,
					completionState);
			TransactionEvent txnEvent = txnMan.generateEvent(existing, true);

			writeTransactionEvent(txnEvent);
		} catch (IOException ioe) {
			throw new BrokerException("error logging transaction", ioe);

		} finally {
			store.txnLogSharedLock.unlock();
		}
	}
	
	
	public void writeTransactionEvent(TransactionEvent txnEvent) throws BrokerException
	{
		try {			
			byte[] data = txnEvent.writeToBytes();
			TransactionLogRecord record = msgLogWriter
					.newTransactionLogRecord();
			record.setBody(data);
			msgLogWriter.write(record);

		} catch (IOException ioe) {
			throw new BrokerException("error logging transaction", ioe);
		}
	}

	public void doCheckpoint() {

		if (Store.getDEBUG()) {
			logger.log(Logger.DEBUG, getPrefix() + " doCheckpoint");
		}
		// 1. get exclusive lock on transaction log
		// 2. wait for all logged commits to be written to message store
		// 3. sync message store
		// 4. write prepared txns (since last checkpoint) to prepared store
		// 5. sync prepareTxnStore
		// 6. reset txn log
		// 7. remove committed txns (since last checkpoint) from prepared store

		Store store = null;
		try {
			store = Globals.getStore();
		} catch (Throwable e) {
			logger.logStack(Logger.ERROR, "failed to getStore", e);

			// fatal
		}

		try {

			// 1. get an exclusive lock on txnLog to prevent any other updates
			// occurring
			store.txnLogExclusiveLock.lock();

			// 2. wait for all logged commits to be written to message store
			localTransactionManager.waitForPlayingToMessageStoreCompletion();
			clusterTransactionManager.waitForPlayingToMessageStoreCompletion();
			remoteTransactionManager.waitForPlayingToMessageStoreCompletion();
			
			//2.1 wait for pending removes to be completed for logged last ack on message 
			loggedMessageHelper.waitForPendingRemoveCompletion();

			// 3. sync the message store
			store.syncStoreOnCheckpoint();

			// 4. write prepared transactions to prepared transaction store
			localTransactionManager
					.writePreparedTransactionsToPreparedTxnStoreOnCheckpoint();
			clusterTransactionManager
					.writePreparedTransactionsToPreparedTxnStoreOnCheckpoint();
			remoteTransactionManager
					.writePreparedTransactionsToPreparedTxnStoreOnCheckpoint();

			// 5. sync prepareTxnStore
			preparedTxnStore.sync();

			// 6. reset txn log
			msgLogWriter.checkpoint();

			// 7. remove committed txns (since last checkpoint) from prepared
			// store
			localTransactionManager.removeCompleteTransactionsAfterCheckpoint();
			clusterTransactionManager
					.removeCompleteTransactionsAfterCheckpoint();
			remoteTransactionManager
					.removeCompleteTransactionsAfterCheckpoint();
			
			// 8. reset logged message list
			loggedMessageHelper.onCheckpoint();

		} catch (Throwable e) {
			String msg = getPrefix()
					+ "Failed to synchronize persistence store for transaction log checkpoint";
			logger.logStack(Logger.ERROR,
					BrokerResources.E_INTERNAL_BROKER_ERROR, msg, e);
		} finally {
			store.txnLogExclusiveLock.unlock();

			if (Store.getDEBUG()) {
				String msg = getPrefix() + " doCheckpoint complete";
				logger.log(Logger.DEBUG, msg);
			}
		}
	}

	// checkpoint handling

	public final void checkpoint() {
		if (Store.getDEBUG()) {
			logger.log(Logger.DEBUG, getPrefix() + " request a checkpoint");
		}

		checkpointManager.enqueueCheckpoint();
	}

	String getPrefix() {
		return "TransactionLogManager: " + Thread.currentThread().getName();
	}

	public LocalTransactionManager getLocalTransactionManager() {
		return localTransactionManager;
	}

	public ClusterTransactionManager getClusterTransactionManager() {
		return clusterTransactionManager;
	}

	public RemoteTransactionManager getRemoteTransactionManager() {
		return remoteTransactionManager;
	}

	public LoggedMessageHelper getLoggedMessageHelper() {
		return loggedMessageHelper;
	}



	public static boolean isReplayInProgress() {
		return replayInProgress;
	}



	public static void setReplayInProgress(boolean val) {
		TransactionLogManager.replayInProgress = val;
	}
}
