/*
 * 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.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import com.sun.messaging.jmq.io.Packet;
import com.sun.messaging.jmq.io.SysMessageID;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.core.ConsumerUID;
import com.sun.messaging.jmq.jmsserver.core.Destination;
import com.sun.messaging.jmq.jmsserver.core.DestinationUID;
import com.sun.messaging.jmq.jmsserver.core.PacketReference;
import com.sun.messaging.jmq.jmsserver.data.TransactionAcknowledgement;
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.data.TransactionWorkMessageAck;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.resources.BrokerResources;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;
import com.sun.messaging.jmq.jmsserver.util.lists.RemoveReason;
import com.sun.messaging.jmq.util.DestType;
import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.util.selector.SelectorFormatException;

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

	MsgStore msgStore;

	public TransactionLogReplayer(MsgStore msgStore) {
		this.msgStore = msgStore;
	}

	String getPrefix() {
		return "TransactionLogReplayer: " + Thread.currentThread().getName();
	}
	
	void replayTransactionWork(TransactionWork txnWork, TransactionUID tid,
			Set dstLoadedSet) throws IOException, BrokerException {

		if (Store.getDEBUG()) {
			String msg = getPrefix() + " replayTransactionWork for tid " + tid
					+ " txnWork=" + txnWork;
			logger.log(Logger.INFO, msg);
		}
		replaySentMessages(txnWork.getSentMessages(), dstLoadedSet);
		replayAcknowledgedMessages(txnWork.getMessageAcknowledgments(),
				dstLoadedSet);

	}
	
	private void replaySentMessages(List<TransactionWorkMessage> sentMessages,
			Set dstLoadedSet) throws IOException, BrokerException {

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

		for (int i = 0; i < sentMessages.size(); i++) {
			TransactionWorkMessage workMessage = sentMessages.get(i);
			replaySentMessage(workMessage, dstLoadedSet);
		}
	}
	
	private void replaySentMessage(TransactionWorkMessage workMessage,
			Set dstLoadedSet) throws IOException, BrokerException {
		// Reconstruct the message
		Packet pkt = workMessage.getMessage();

		SysMessageID mid = pkt.getSysMessageID();
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " replaying sent message: "
					+ workMessage + " dest= "+pkt.getDestination();
			logger.log(Logger.INFO, msg);
		}
		// Make sure destination exists; auto-create if necessary
		// TODO
		// recreating destinations may cause a bug if a destination had been
		// deleted by a command after the message was logged.
		// It will now reappear - (as an auto destination?)
		
        String dname = pkt.getDestination();
        int dtype = (pkt.getIsQueue() ? DestType.DEST_TYPE_QUEUE : DestType.DEST_TYPE_TOPIC);

        List dlist = Destination.findMatchingIDs(DestinationUID.getUID(dname, dtype));

        DestinationUID did = null;
        Iterator itr = dlist.iterator();
        while (itr.hasNext()) {
            
        did = (DestinationUID)itr.next();
		Destination dst = Destination.getDestination(did, dtype, true, true); 
		did = dst.getDestinationUID();

		// Load all msgs in order to verify if any msgs are missing
		if (!dstLoadedSet.contains(dst)) {
			dst.load();
			dstLoadedSet.add(dst); // Keep track of what has been loaded
		}

		// Check to see if the msg is in the store

		boolean exists = msgStore.containsMessage(did, mid);
		if (exists) {
			if (Store.getDEBUG()) {
				String msg = getPrefix() + " stored message exists " + mid;
				logger.log(Logger.INFO, msg);
			}
			// check if stored interests are the same as logged interests
			HashMap storedInterests = msgStore.getInterestStates(did, mid);
			boolean matched = compareStoredConsumers(mid,workMessage.getStoredInterests(), storedInterests);
			if(!matched)
			{

				logger.log(Logger.FORCE, BrokerResources.I_REPLACE_MSG_TXNLOG, mid);
				msgStore.removeMessage(did, mid, false);
				rerouteMessage(pkt, workMessage.getStoredInterests(),mid,dst);
			}
			
		} else {
			if (Store.getDEBUG()) {
				String msg = getPrefix() + " stored message does not exist " + mid;
				logger.log(Logger.INFO, msg);
			}
			rerouteMessage(pkt, workMessage.getStoredInterests(),mid,dst);
			
		}

        } //while
		
	}
	
	private void rerouteMessage(Packet pkt, ConsumerUID[] storedInterests, SysMessageID mid, Destination dst) throws BrokerException
	{
		// TO DO 
		// This should use existing routing
		
		logger.log(Logger.FORCE, BrokerResources.I_RECONSTRUCT_MSG_TXNLOG,
				mid);
		PacketReference pr = PacketReference.createReference(pkt, dst.getDestinationUID(), null);
        if(pr.isExpired())
        {
        	String msg ="not routing expired message on transaction log replay "+mid;
        	logger.log(Logger.INFO, msg);
        	return;
        }
		try {
			dst.queueMessage(pr, false);
			dst.routeNewMessage(pr);
		} catch (SelectorFormatException e) {
			// shouldn't happens
			throw new BrokerException(Globals.getBrokerResources().getString(
					BrokerResources.E_ROUTE_RECONSTRUCTED_MSG_FAILED, mid), e);
		}
	}
	
	private boolean compareStoredConsumers(SysMessageID mid,
			ConsumerUID[] logged, HashMap storedmap) {
		boolean match = true;
		int loggedLength = 0;
		if (logged != null)
			loggedLength = logged.length;
		if (loggedLength < storedmap.size()) {
			match = false;
			logger.log(Logger.ERROR,
					"Mismatch in number of logged and stored consumers for "
							+ mid + " logged=" + loggedLength + " stored="
							+ storedmap.size());
		}
		if (loggedLength > storedmap.size()) {
			match = false;
			logger.log(Logger.WARNING,
					"Mismatch in number of logged and stored consumers for "
							+ mid + " logged=" + loggedLength + " stored="
							+ storedmap.size());
		}
		HashSet loggedSet = new HashSet(loggedLength);
		for (int i = 0; i < loggedLength; i++) {

			ConsumerUID c = logged[i];
			loggedSet.add(c);
			if (!storedmap.containsKey(c)) {
				logger.log(Logger.WARNING,
						"stored interest does not contain logged interest. sysid= "
								+ mid + " ConsumerUID=" + c);
				match = false;
			}
		}
		Iterator iter = storedmap.keySet().iterator();
		while (iter.hasNext()) {
			Object storedConsumer = iter.next();
			if (!loggedSet.contains(storedConsumer)) {
				logger.log(Logger.ERROR,
						"logged interests does not contain stored interst. sysid= "
								+ mid + " consumerid=" + storedConsumer);
				match = false;
			}
		}

		if (match) {
			if (Store.getDEBUG()) {
				String msg = getPrefix()
						+ " stored consumers match. numConsumers= "
						+ loggedLength;

				logger.log(Logger.INFO, msg);
			}
		}
		return match;
	}
	
	
	public void replayNonTxnMsgAck(NonTransactedMsgAckEvent event,
			Set dstLoadedSet) throws IOException, BrokerException {

		replayAcknowledgedMessage(event.messageAck, dstLoadedSet);
	}

	public void replayNonTxnMsg(NonTransactedMsgEvent event,
			Set dstLoadedSet) throws IOException, BrokerException {

		replaySentMessage(event.message, dstLoadedSet);

	}
	
	private void replayAcknowledgedMessages(
			List<TransactionWorkMessageAck> acknowledgedMessages,
			Set dstLoadedSet) throws IOException, BrokerException {

		for (int i = 0; i < acknowledgedMessages.size(); i++) {
			TransactionWorkMessageAck messageAck = acknowledgedMessages.get(i);
			replayAcknowledgedMessage(messageAck, dstLoadedSet);

		}
	}

	
	private void replayAcknowledgedMessage(
			TransactionWorkMessageAck messageAck, Set dstLoadedSet)
			throws IOException, BrokerException {

	
		DestinationUID did = messageAck.getDestUID();
		SysMessageID mid = messageAck.getSysMessageID();
		ConsumerUID iid = messageAck.getConsumerID();
		if (Store.getDEBUG()) {
			String msg = getPrefix() + " replaying acknowledged message "
					+ messageAck;
			logger.log(Logger.INFO, msg);
		}
		// Make sure dst exists; autocreate if possible
		Destination dst = Destination.getDestination(did.getName(), did
				.isQueue() ? DestType.DEST_TYPE_QUEUE
				: DestType.DEST_TYPE_TOPIC, true, true);

		// Load all msgs inorder to update consumer states
		if (!dstLoadedSet.contains(dst)) {
			dst.load();
			dstLoadedSet.add(dst); // Keep track of what has been loaded
		}

		if (msgStore.containsMessage(did, mid)) {
			logger.log(logger.FORCE, BrokerResources.I_UPDATE_INT_STATE_TXNLOG,
					iid, mid);
			// For Queue, ensure the stored ConsumerUID is 0 otherwise
			// use try using the correct value; see bug 6516160
			if (dst.isQueue() && iid.longValue() != 0) {
				msgStore.updateInterestState(did, mid, PacketReference
						.getQueueUID(), Store.INTEREST_STATE_ACKNOWLEDGED,
						false);
			} else {
				msgStore.updateInterestState(did, mid, iid,
						Store.INTEREST_STATE_ACKNOWLEDGED, false);
			}

			acknowledgeOnReplay(dst, mid, iid);

		} else {
			logger.log(logger.FORCE,
					BrokerResources.I_DISREGARD_INT_STATE_TXNLOG, iid, mid);
		}
	}
	
	

	public void replayMessageRemoval(MsgRemovalEvent event,
			Set dstLoadedSet) throws IOException, BrokerException {
		DestinationUID did = event.destUID;
		SysMessageID mid = event.sysMessageID;
		
		// Make sure dst exists; autocreate if possible
		Destination dst = Destination.getDestination(did.getName(), did
				.isQueue() ? DestType.DEST_TYPE_QUEUE
				: DestType.DEST_TYPE_TOPIC, true, true);

		// Load all msgs inorder to update consumer states
		if (!dstLoadedSet.contains(dst)) {
			dst.load();
			dstLoadedSet.add(dst); // Keep track of what has been loaded
		}

		logger.log(Logger.FORCE, Globals.getBrokerResources().getKString(
                   BrokerResources.I_RM_MSG_ON_REPLAY_MSG_REMOVAL, mid, dst));
        dst.removeMessage(mid, RemoveReason.REMOVED_OTHER);

		if (msgStore.containsMessage(did, mid)) {
			msgStore.removeMessage(did, mid, false);
		}
		
	}

	
	public void acknowledgeOnReplay(Destination dst, SysMessageID mid,
			ConsumerUID iid) throws IOException, BrokerException {
		PacketReference ref = dst.getMessage(mid);
		boolean allAcked = false;
		if (ref != null) {
			allAcked = ref.acknowledgedOnReplay(iid, iid);
			if (Store.getDEBUG()) {
				String msg = getPrefix() + " acknowledgedOnReplay  " + mid
						+ " allAcked =" + allAcked;
				logger.log(Logger.INFO, msg);
			}
		} else {
			if (Store.getDEBUG()) {
				String msg = getPrefix()
						+ " did not find packet in destination " + mid;
				logger.log(Logger.INFO, msg);
			}
		}
		if (allAcked) {
			dst.removeMessage(mid, RemoveReason.ACKNOWLEDGED);
			if (Store.getDEBUG()) {
				String msg = getPrefix() + " removed message from destination "
						+ mid;
				logger.log(Logger.INFO, msg);
			}
		}
	}
	
	public void replayRemoteAcks(TransactionAcknowledgement[] txnAcks, DestinationUID[] destIds,
			TransactionUID tid, HashSet dstLoadedSet) throws IOException,BrokerException {
		// TO DO
		// store destUID alongside acks so that we can replay more directly

		if (Store.getDEBUG()) {
			String msg = getPrefix() + " replayRemoteAcks ";
			logger.log(Logger.INFO, msg);
		}
		for (int i = 0; i < txnAcks.length; i++) {
			TransactionAcknowledgement txnAck = txnAcks[i];
			DestinationUID destId = destIds[i];
			SysMessageID mid = txnAck.getSysMessageID();
			ConsumerUID iid = txnAck.getStoredConsumerUID();
			
			//Destination dest = Destination.getDestination(destId);
			// make sure it is loaded
			
			int type = (destId.isQueue() ? DestType.DEST_TYPE_QUEUE
					: DestType.DEST_TYPE_TOPIC);
			Destination dest = Destination.getDestination(destId.getName(),
					type, true, true);
			dest.load();

			PacketReference pr = dest.getMessage(mid);
			if (pr == null) {
				// could have been acknowledged already?
				
				// TO DO check this further
				String msg = " could not find packet for replayed message ack "
						+ txnAck + " dest " + destId + " in transaction " + tid;
				logger.log(Logger.WARNING, msg);
			} else {
				if (msgStore.containsMessage(destId, mid)) {
					logger
							.log(logger.FORCE,
									BrokerResources.I_UPDATE_INT_STATE_TXNLOG,
									iid, mid);
					// For Queue, ensure the stored ConsumerUID is 0 otherwise
					// use try using the correct value; see bug 6516160
					if (dest.isQueue() && iid.longValue() != 0) {
						msgStore.updateInterestState(destId, mid, PacketReference
								.getQueueUID(),
								Store.INTEREST_STATE_ACKNOWLEDGED, false);
					} else {
						msgStore.updateInterestState(destId, mid, iid,
								Store.INTEREST_STATE_ACKNOWLEDGED, false);
					}
					
					acknowledgeOnReplay(dest,mid,iid);
				} else {
					logger.log(logger.FORCE,
							BrokerResources.I_DISREGARD_INT_STATE_TXNLOG, iid,
							mid);
				}
			}

		}

	}

}
