/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright (c) 2000-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.
 */

/*
 * @(#)PacketReference.java	1.201 08/28/07
 */ 

package com.sun.messaging.jmq.jmsserver.core;

import java.util.*;
import java.lang.ref.*;
import java.io.*;
import java.util.concurrent.ConcurrentHashMap;
import com.sun.messaging.jmq.io.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.DMQ;
import com.sun.messaging.jmq.jmsserver.GlobalProperties;
import com.sun.messaging.jmq.jmsserver.core.ConsumerUID;
import com.sun.messaging.jmq.jmsserver.resources.BrokerResources;
import com.sun.messaging.jmq.jmsserver.core.Consumer;
import com.sun.messaging.jmq.jmsserver.data.TransactionUID;
import com.sun.messaging.jmq.jmsserver.data.TransactionState;
import com.sun.messaging.jmq.jmsserver.data.TransactionBroker;
import com.sun.messaging.jmq.jmsserver.service.ConnectionUID;
import com.sun.messaging.jmq.jmsserver.service.imq.IMQConnection;
import com.sun.messaging.jmq.jmsserver.service.ConnectionManager;
import com.sun.messaging.jmq.jmsserver.util.*;
import com.sun.messaging.jmq.jmsserver.config.*;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.service.Connection;
import com.sun.messaging.jmq.util.DestType;
import com.sun.messaging.jmq.util.SizeString;
import com.sun.messaging.jmq.util.lists.*;
import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.jmsserver.util.memory.MemoryGlobals;

/**
 * This class contains the references for a packet which is passed
 * around the system.
 *
 * An optimization of this class would allow messages to be 
 * retrieved from the persistent store (if they are persistent)
 *
 */

public class PacketReference implements Sized, Ordered
{

    private static ConsumerUID queueUID = null;


    /**
     * Controls if we should enable the fix for bug 6196233 and prepend the
     * message ID with "ID:"
     */
    private static boolean PREPEND_ID =
        Globals.getConfig().getBooleanProperty(
            Globals.IMQ + ".fix.JMSMessageID", false);

    private boolean commit2pwait = Globals.getConfig().getBooleanProperty(
                                   Globals.IMQ+".cluster.2pcommitAckWaitReply", false);

    static {
        queueUID = new ConsumerUID(true /* empty */);
        queueUID.setShouldStore(true);
        
    }


    private static boolean DEBUG = false;

    private static boolean DEBUG_CLUSTER_TXN =
        Globals.getConfig().getBooleanProperty(
                            Globals.IMQ + ".cluster.debug.txn") || DEBUG;

    /**
     * set once store() is called
     */
    private boolean isStored = false;

    private boolean neverStore = false;

    /**
     * ets once store + interests are saved
     */
    private boolean isStoredWithInterest = false;

    private HashMap attachedData = null;

    /**
     * has packet been destroyed
     */
    private boolean destroyed = false;


    /**
     * message id for the packet
     */
    private SysMessageID msgid;


    transient ConnectionUID con_uid;

    /**
     * time the packet reference was created (entered the system)
     */
    private long creationtime;

    /**
     * time the packet reference was last accessed
     */
    private long lastaccesstime;

    /**
     * sequence # (used for sorting queue browsers)
     */
    private int sequence;

    /**
     * if the message is persisted
     */
    private boolean persist;


    /**
     * determines if the message should remain even if its expired
     * (if the lbit count is > 0)
     */
    private Set lbit_set = null;

    private boolean sendMessageDeliveredAck = false;

    /**
     * size of the packet data 
     */
    private long size;

    /**
     * message properties (null if swapped)
     */
    private Hashtable props;

    private HashMap headers;

    /**
     * original packet or SoftReference (null if swapped)
     */
    private Object pktPtr;

    /**
     * priority of the packet
     */
    private int priority = 4;

    /**
     * isQueue setting
     */
    boolean isQueue = false;


    /**
     * flag called when a packet "should" be removed
     * but is is remaining in memory (because the lbit
     * is set on the packet
     */ 
    boolean invalid = false;


    /** 
     * For remote message reference and its replacement ref only 
     */
    boolean overrided = false;
    Object lockObject = new Object();
    boolean canLock = false;
    boolean locked = false;
    boolean overriding = false;
    Thread lockOwner = null;

    /**
     * destination uid
     */
    DestinationUID destination = null;

    ExpirationInfo expire = null;

    /**
     * expired flag
     */
    boolean isExpired = false;

    /**
     * timestamp on the packet (for sorting)
     */
    long timestamp = 0;

    /**
     * transaction id
     */
    TransactionUID transactionid;


    /**
     * override redeliver
     */
    transient boolean overrideRedeliver = false;


    transient BrokerAddress addr = null;


    transient Destination d = null;

    transient String clientID = null;


    int interestCnt;
    int deliveredCnt;
    int ackCnt;
    int deadCnt;

    Map ackInfo = null;

    private static final int INITIAL = 0;
    private static final int ROUTED = 1;
    private static final int DELIVERED = 2;
    private static final int CONSUMED = 3;
    private static final int ACKED = 4;
    private static final int DEAD = 5;

    /**
     * Indicates if this message is in delivery or has been delivered to client
     */
    Map inDelivery = new ConcurrentHashMap(); 

    /**
     * true if the message is to be removed and need to be prevented delivering to client
     */
    boolean inRemoval = false; 



    ConsumerUID lastDead = null;
    
    long order = 0;

    public long getOrder() {
        return order;
    }
    public void setOrder(long order) {
        this.order = order;
    }

    static class ConsumerMessagePair
    {
        private ConsumerUID uid;
        private int state = INITIAL;
        private int redeliverCnt = 0;
        private boolean stored;
        private String deadComment = null;
        private Reason deadReason = null;
        private Throwable deadException = null;
        private long timestamp = 0;
        private String deadBroker = null;


        public ConsumerMessagePair(ConsumerUID uid,
                  boolean stored) {
            this.uid = uid;
            this.stored = stored;
        }

        public synchronized int incrementRedeliver(){
            redeliverCnt ++;
            return redeliverCnt;
        }

        public synchronized int decrementRedeliver() {
            redeliverCnt --;
            return redeliverCnt;
        }
        
        public synchronized boolean setState(int state) {
            if (this.state == state) return false;
            this.state = state;
            timestamp = System.currentTimeMillis();
            return true;
        }

        public synchronized int getState() {
            return state;
        }

        public synchronized boolean compareAndSetState(int state, int expected) {
            if (this.state != expected) {
                   return false;
            }
            timestamp = System.currentTimeMillis();
            this.state = state;
            return true;
        }

        public synchronized boolean compareState(int expected) {
            return this.state == expected;
        }

        public synchronized boolean compareStateGT(int expected) {
            return this.state > expected;
        }

        public synchronized boolean compareStateLT(int expected) {
            return this.state < expected;
        }

        public synchronized boolean setStateIfLess(int state, int max) {
            if (compareStateLT(max))
                return setState(state);
            return false;
        }

        public boolean isStored() {
            return stored;
        }

        public int getRedeliverCount() {
            return redeliverCnt;
        }

        public void setRedeliverCount(int cnt) {
            redeliverCnt = cnt;
        }

        public void setDeadComment(String str) {
             deadComment = str;
        }
        public void setDeadReason(Reason r) {
             deadReason = r;
        }
        public String getDeadComment() {
             return deadComment;
        }
        public Reason getDeadReason() {
            return deadReason;
        }
        public Throwable getDeadException() {
            return deadException;
        }

        public void setDeadException(Throwable thr) {
             deadException = thr;
        }
        public void setDeadBroker(String bkr) {
             deadBroker = bkr;
        }
        public String getDeadBroker() {
            return deadBroker;
        }

        public String toString() {
            return ": CMPair["+ uid.longValue()  + ":" + stateToString(state) 
                    + ":" + stored + "] - " + timestamp;
        }
    }

    public static ConsumerUID getQueueUID() {
        return queueUID;
    }

    public static PacketReference createReference(Packet p, 
             DestinationUID duid, 
             Connection con) throws BrokerException {
        PacketReference pr = new PacketReference(p,duid, con);
        if (con != null && pr.getExpiration() != null) {
            con.checkClockSkew(pr.getTime(), pr.getTimestamp(),
                pr.getExpireTime());
        }
            
        return pr;
    }

    public static PacketReference createReference(Packet p, 
             Connection con) throws BrokerException {
        return createReference(p, null, con);
    }

    public static void moveMessage(PacketReference oldLoc,
        PacketReference newLoc, Set targets)
            throws BrokerException, IOException
    {
        if (targets == null) {
            // get from the original message
            throw new RuntimeException("Internal error: moving message"
                      + " to null targets not supported");
        } 
        if (!oldLoc.isStored && oldLoc.persist) {
            newLoc.store(targets);
            return;
        }

        ConsumerUID[] uids = null;
        int[] states = null;
        ReturnInfo info = calculateConsumerInfo(targets, oldLoc.persist);
        newLoc.ackInfo = info.ackInfo;
        newLoc.ackCnt = info.ackInfo.size();
        if (info.uids == null || info.uids.length == 0) {
            // nothing to store XXXremove old? 
            newLoc.neverStore = true;
            newLoc.isStored = false;
            newLoc.isStoredWithInterest = false;
            return;
        }
        if (oldLoc.isStored && oldLoc.persist && !oldLoc.neverStore) {
            Globals.getStore().moveMessage(newLoc.getPacket(),
                 oldLoc.getDestinationUID(), 
                 newLoc.getDestinationUID(),
                 info.uids, info.states, Destination.PERSIST_SYNC);
            newLoc.isStored = true;
            newLoc.isStoredWithInterest = true;
            oldLoc.isStored = false;
            oldLoc.isStoredWithInterest = false;
        } else if (oldLoc.persist) {
            Globals.getStore().storeMessage(newLoc.getDestinationUID(),
                newLoc.getPacket(),
                info.uids, info.states, Destination.PERSIST_SYNC);
            newLoc.isStored = true;
            newLoc.isStoredWithInterest = true;
            newLoc.neverStore = false;
        } else {
            newLoc.isStored = false;
            newLoc.isStoredWithInterest = false;
            newLoc.neverStore = true;
        }
        return;
    }


    static class ReturnInfo
    {
        ConsumerUID uids[];
        int states[];
        int interestCnt;
        Map ackInfo;
    }

// LKS- BUG SOMEWHERE BELOW
/*
 * this method is called when a set of consumers are being routed
*/
    private static ReturnInfo calculateConsumerInfo(Collection consumers, boolean checkStored)
    {
 
        if (consumers.isEmpty())
            return null;

        ReturnInfo ri = new ReturnInfo(); 
        ri.ackInfo = Collections.synchronizedMap(new HashMap());
        ArrayList storedConsumers = (checkStored ? new ArrayList() : null);
        Iterator itr = consumers.iterator();

        int count = 0;
        while (itr.hasNext()) {
           Object o = itr.next();
           ConsumerUID cuid = null;
           if (o instanceof Consumer) {
               cuid = ((Consumer)o).getStoredConsumerUID();
           } else {
               cuid = (ConsumerUID)o;
           }
           ri.interestCnt ++;

           boolean store = false;
           if (storedConsumers != null && cuid.shouldStore()) {
               storedConsumers.add(cuid);
               store = true;
            }

           ConsumerMessagePair cmp = new ConsumerMessagePair(cuid,
                  store);
           ri.ackInfo.put(cuid, cmp);

        }
      
        if (storedConsumers != null) {
            ConsumerUID [] type = new ConsumerUID[0];
            ri.uids = (ConsumerUID[])
                   storedConsumers.toArray(type);
            ri.states = new int[ri.uids.length];
            for (int i=0; i < ri.states.length; i ++) {
                ri.states[i] = Store.INTEREST_STATE_ROUTED;
            }
         }
         return ri;
    }


    /**
     * create a new PacketReference object
     */
    private PacketReference(Packet pkt, DestinationUID duid,
             Connection con) throws BrokerException {
        this.creationtime = System.currentTimeMillis();
        this.lastaccesstime = creationtime;
        this.msgid = (SysMessageID)pkt.getSysMessageID().clone();
        this.isQueue = pkt.getIsQueue();
        this.persist = pkt.getPersistent();
        this.priority = pkt.getPriority();
        this.sequence = pkt.getSequence();
        this.timestamp = pkt.getTimestamp();
        if (pkt.getRedelivered())
            overrideRedeliver = true;

        if (con != null)
            this.clientID = (String)con.getClientData(IMQConnection.CLIENT_ID);
        setExpireTime(pkt.getExpiration());
        this.size = pkt.getPacketSize();
        String d  = pkt.getDestination();
        this.con_uid = (con == null ? null : con.getConnectionUID());
        if (duid != null) {
            destination = duid;
        } else {
            destination = DestinationUID.getUID(d, 
               isQueue);
        }
        long tid = pkt.getTransactionID();
        synchronized(this) {
            setPacketObject(false, pkt);
        }
        if (tid != 0) {
            transactionid = new TransactionUID(tid);
        } else {
            transactionid = null;
        }
    }

    public void setSequence(int seq) {
        this.sequence = seq;
    }


    public String getClientID() {
        return clientID;
    }


    private ConsumerMessagePair getAck(Object obj) {
        ConsumerUID cuid = null;
        if (obj instanceof ConsumerUID) {
            cuid = (ConsumerUID)obj;
        } else if (obj instanceof Consumer) {
            cuid = ((Consumer)obj).getConsumerUID();
        } else {
            throw new RuntimeException("Bogus ID");
        }
        if (ackInfo == null)  {
            throw new RuntimeException("Internal Error: No AckInfo for message "+ getSysMessageID());
        }
        return (ConsumerMessagePair) ackInfo.get(cuid);
    }




/*
 *-------------------------------------------------------------------
 * 
 *   ACCESS METHODS
 *
 *--------------------------------------------------------------------
 */

    public void setDestination(Destination d) {
        this.d = d;
    }

    public Destination getDestination() {
         if (d == null) {
             d = Destination.getDestination(destination);
         }
         return d;
    }

    boolean isStored() {
        return isStored;
    }


    public void setBrokerAddress(BrokerAddress a)
    {
        addr = a;
    }

    public BrokerAddress getAddress() {
        return addr;
    }

    public boolean isLocal() {
        return addr == null || addr == Globals.getMyAddress();
    }

    public String toString() {
        return "PacketReference["+msgid+"]";
    }

    Set deliveredMsgAcks = new HashSet();

    public synchronized boolean getMessageDeliveredAck(ConsumerUID uid) {
        return !deliveredMsgAcks.isEmpty() && 
              deliveredMsgAcks.contains(uid);
    }

    public synchronized ConsumerUID[] getConsumersForMsgDelivered()
    {
        return (ConsumerUID [])deliveredMsgAcks.toArray(
                   new ConsumerUID[0]);
    }

    public synchronized void removeMessageDeliveredAck(ConsumerUID uid) {
        deliveredMsgAcks.remove(uid);
    }

    public synchronized void addMessageDeliveredAck(ConsumerUID uid) {
        deliveredMsgAcks.add(uid);
    }

    Hashtable remoteConsumerUIDs = new Hashtable();

    public void addRemoteConsumerUID(ConsumerUID cuid, 
                                     ConnectionUID cnuid) {
        remoteConsumerUIDs.put(cuid, cnuid);
    }

    public Hashtable getRemoteConsumerUIDs() {
        return remoteConsumerUIDs;
    }

    private void setExpireTime(long time) {
        if (time == 0) {
            expire = null;
        } else {
            expire = new ExpirationInfo(msgid, time);
        }
    }
    public long getExpireTime() {
        if (expire == null) return 0;
        return expire.getExpireTime();
    }

    public ExpirationInfo getExpiration() {
        return expire;
    }

    public void overrided() {
        overrided = true;
    }
    public boolean isOverrided() {
        return overrided;
    }

    public void overriding() {
        overriding = true;
    }
    public boolean isOverriding() {
        return overriding;
    }

    /**
     * only called by the ref creator before put into MT access
     */
    public void lock() {
        canLock = true;
        locked = true;
        lockOwner = Thread.currentThread(); 
    }

    public void unlock() {
        synchronized(lockObject) {
            locked = false;
            lockOwner = null;
            lockObject.notifyAll();
        }
    }

    public PacketReference checkLock(boolean wait) {
        if (!canLock) return this;
        synchronized(lockObject) {
            if (locked && (lockOwner != null && Thread.currentThread() == lockOwner)) return this;
            while (wait && locked) {
                Globals.getLogger().log(Logger.INFO, "Wait for reference : "+this);
                try {
                lockObject.wait(5000);
                } catch (InterruptedException e) {
                }
            }
            if (locked) return null;
        }
        return this;
    }

    public void setInvalid() {
        invalid = true;
    }

    private Packet getPacketObject() {

        assert (pktPtr ==  null ||
               pktPtr instanceof SoftReference
               || pktPtr instanceof Packet) : pktPtr;

        Object ptr = pktPtr;
        if (ptr == null) return null;
        if (ptr instanceof SoftReference) {
            return (Packet)((SoftReference)pktPtr).get();
        }
        return (Packet)pktPtr;
    }

    private void setPacketObject(boolean soft, Packet p) {

        assert Thread.holdsLock(this);
        if (soft) {
            pktPtr = new SoftReference(p);
        } else {
            pktPtr = p;
        }
    }

    private void makePacketSoftRef() {
        assert Thread.holdsLock(this);
        Object ptr = pktPtr;
        if (ptr != null && ptr instanceof Packet) {
            pktPtr = new SoftReference(ptr);
        }
    }

    public void setNeverStore(boolean s) {
        neverStore = s;
    }

    public long byteSize() {
        return size;
    }

    public void setStoredWithInterest(boolean withInterest) {
        isStoredWithInterest = withInterest;
    }

    public synchronized void setLoaded() {
        isStored = true;
        makePacketSoftRef();
    }

    public ConnectionUID getProducingConnectionUID()
    {
        return con_uid;
    }

    public int getPriority() {
        return priority;
    }

    public TransactionUID getTransactionID() {
        return transactionid;
    }

    public boolean getIsQueue() {
        return this.isQueue;
    }

    public long getTime() {
        return creationtime;
    }

    public long getSequence() {
        return sequence;
    }

    public synchronized void setLastBit(ConsumerUID id)
        throws IllegalStateException
    {
        if (isInvalid() && isDestroyed()) {
            throw new IllegalStateException(
                   Globals.getBrokerResources().getString(
                    BrokerResources.X_INTERNAL_EXCEPTION, 
                        "reference has been destroyed"));
        }
        if (lbit_set == null) {
            lbit_set = new HashSet();
        }
        lbit_set.add(id);

    }

    public Hashtable getDebugState() {
        Hashtable ht = new Hashtable();
        ht.put("AckCount", String.valueOf(ackCnt));
        ht.put("DeadCount", String.valueOf(deadCnt));
        ht.put("ackInfo[#]", String.valueOf( ackInfo.size()));
        ht.put("interestCount", String.valueOf(interestCnt));
        Vector vt = new Vector();
        synchronized(ackInfo) {
            Iterator itr = ackInfo.keySet().iterator();
            while (itr.hasNext()) {
                Object key = itr.next();
                ConsumerMessagePair cmp = getAck(key);
                vt.add(cmp.toString());
            }    
        }
        if (!vt.isEmpty())
            ht.put("Acks",vt);
        return ht;
    }
        

    public synchronized boolean isLast(ConsumerUID id) {
        if (lbit_set == null) {
            return false;
        }
        return lbit_set.contains(id);
    }

    public synchronized void removeIsLast(ConsumerUID id) {
        if (lbit_set == null) {
            return;
        }
        lbit_set.remove(id);
        if (lbit_set.isEmpty() && invalid) { // clean up, lbit is gone
            destroy(); // at next expiration check, we will be
                       // removed from the packet store
        }
    }

    public synchronized boolean getLBitSet() {
        return lbit_set != null && !lbit_set.isEmpty();
    }

    public DestinationUID getDestinationUID() {
        return destination;
    }

    public String getDestinationName() {
        return destination.getName();
    }

    public synchronized Packet getPacket()
    {
        Packet pkt = getPacketObject();
        if (pkt != null || destroyed) {
            return pkt;
        }

        assert persist;

        if (!persist) {
            return null;
        }

        pkt = recoverPacket();

        assert pkt != null;

        setPacketObject(true, pkt);
        return pkt;
    }

    private Packet recoverPacket() 
    {
        // recover from the database
        assert Thread.holdsLock(this);
        assert pktPtr == null || 
            (pktPtr instanceof SoftReference &&
              ((Reference)pktPtr).get() == null);

        try {
            Packet p = Globals.getStore().getMessage(destination, msgid);
            assert p != null;
            return p;

        } catch (BrokerException ex) {
            assert false :ex;
            Globals.getLogger().logStack(Logger.ERROR,
                 BrokerResources.E_LOAD_MSG_ERROR,
                 msgid.toString(), ex);
        }
            
        return null;
    }

    public synchronized Hashtable getProperties() 
        throws ClassNotFoundException
    {
        if (destroyed || invalid) {
            return new Hashtable();
        }
        this.lastaccesstime = System.currentTimeMillis();
        Packet pkt = getPacketObject();
        if (props == null && !destroyed) {
            if (pkt == null) {
                pkt = getPacket();
            }
            try {
                props = pkt.getProperties(); 
            } catch (IOException ex) {
                // no properties
                Globals.getLogger().log(Logger.INFO,"Internal Exception: " , ex);
                props = new Hashtable();
            } catch (ClassNotFoundException ex) {
                assert false; // should not happen
                throw ex;
            }
        }
        return props;
     }

    public synchronized HashMap getHeaders() {
        if (headers == null) {
            if (destroyed || invalid) {
                return new HashMap();
            }
            Packet pkt = getPacketObject();
            assert pkt != null;
            headers = new HashMap();
            if (pkt == null) {
				// Fix for CR 6891615
				// reload packet if it has been GCd
				pkt = getPacket();
				if (DEBUG) {
					Globals.getLogger().log(
							Logger.DEBUG, 
							"reloaded packet for non-destroyed message "
									+ msgid);
				}
				if (pkt == null) {
					Globals.getLogger().log(
							Logger.ERROR,
							"could not reload packet for non-destroyed message "
									+ msgid);
					return headers;
				}
			 }
            headers.put("JMSPriority", new Integer(priority));

            /*
             * XXX If the fix for bug 6196233 is enabled, then 
             * prepend the messageid with "ID:".
             */
            headers.put("JMSMessageID",
               (PREPEND_ID ? "ID:" : "") + msgid.toString());
            headers.put("JMSTimestamp", new Long(timestamp));
            headers.put("JMSDeliveryMode",
                 (pkt.getPersistent() ? "PERSISTENT" :
                     "NON_PERSISTENT"));
            headers.put("JMSCorrelationID", pkt.getCorrelationID());
            headers.put("JMSType", pkt.getMessageType());
        }
        return headers;
    }


    public  SysMessageID getSysMessageID() 
    {
        return msgid;
    }

    public String getSysMessageIDString() {
        // returns the header value
        Map headers = getHeaders();
        if (headers == null) return null; ;
        return (String)headers.get( "JMSMessageID");

    }

    public  long getCreateTime() 
    {
        return creationtime;
    }

    public  long getLastAccessTime() 
    {
        return lastaccesstime;
    }

    public  long getTimestamp() 
    {
        return timestamp;
    }

    public void setTimestamp(long time)
    {
        timestamp = time;
    }


    public  boolean isPersistent() 
    {
        return persist;
    }

    public  void overridePersistence(boolean persist) 
    {
        this.persist = persist;
    }

    public  long getSize() 
    {
        return size;
    }


    public boolean isDestroyed()
    {
        return destroyed;
    }


    public synchronized boolean isInvalid()
    {
        return invalid;
    }

    public synchronized boolean checkDeliveryAndSetInRemoval() {
        if (destroyed || invalid ||
            ackInfo == null /*not routed yet*/) {
            inRemoval = true;
            return true;
        }

        if (ackCnt + deadCnt >= interestCnt) {
            inRemoval = true;
            return true;
        }

        if (inDelivery.size() > 0) {
            return false;
        }
        inRemoval = true;
        return true;
    }

    public synchronized boolean checkRemovalAndSetInDelivery(ConsumerUID sid) {
        if (destroyed || invalid || inRemoval || isExpired()) {
            return false;
        }
        inDelivery.put(sid, sid);
        return true;
    }

    public synchronized void removeInDelivery(ConsumerUID sid) {
        inDelivery.remove(sid);
    }

    public boolean mayExpire() {
        return expire == null;
    }

    public boolean isExpired() {
        if (expire == null) {
            return false;
        }
        return isExpired(System.currentTimeMillis());

    }

    public void overrideExpireTime(long expire)
    {
        setExpireTime(expire);
    }

    public boolean isExpired(long curtime) {
        if (isExpired) {
            return true;
        }
        if (expire == null) {
            return false;
        }
        boolean expiring = expire.isExpired(curtime);
        if (expiring) {
           // change to soft reference
           synchronized(this) {
               makePacketSoftRef();
           }
        }
        return expiring;
    }

    void clearExpirationInfo() {
        isExpired = true;
        expire = null;
    }



    public boolean equals(Object obj) {
        if (msgid == null)
            return msgid == obj;
        if (obj instanceof SysMessageID)
            return msgid.equals(obj);
        if (obj instanceof Packet)
            return msgid.equals(((Packet)obj).getSysMessageID());
        if (obj instanceof PacketReference)
            return msgid.equals(((PacketReference)obj).msgid);
        return false;
    }

    public int hashCode() {
        return msgid == null ? 0 : msgid.hashCode();
    }
        
      

    public boolean matches(DestinationUID uid)
    {
        return true;
    }  

/*
 *-------------------------------------------------------------------
 * 
 *   Acknowledgement handling methods
 *
 *--------------------------------------------------------------------
 */


    public static String stateToString(int state) {
        switch (state) {
            case INITIAL:
                return "INITIAL";
            case ROUTED:
                return "ROUTED";
            case DELIVERED:
                return "DELIVERED";
            case CONSUMED:
                return "CONSUMED";
            case ACKED:
                return "ACKED";
            case DEAD:
                return "DEAD";
        }
        return "UNKNOWN";
    }


    /**
     * stores the persistent message (if necessary)
     * (may be called by transactions to store the 
     * message)
     */

    public synchronized void store()  
        throws BrokerException
    {
        if (!destroyed && persist && !neverStore && !isStored) {
            // persist indicated IF we want to
            // store the message and may be
            // different from the actual
            // state on the message
            assert pktPtr instanceof Packet;
            try {
                Globals.getStore().storeMessage(destination,
                    (Packet)getPacket(), Destination.PERSIST_SYNC);
                if(Globals.isNewTxnLogEnabled())
                {
                	if(getPacket().getTransactionID()>0)
                	{
                		//transacted so will realy be stored when routed
                		
                		// don't make a softRef yet as may get garbage collected
                        isStored = false;
                        return;
                	}
                	else
                	{
                		makePacketSoftRef();
                	}
                }
                else{
                	makePacketSoftRef();
                }
            } catch (IOException ex) {
                throw new BrokerException(
                        ex.toString(), ex);
            } catch (Exception ex) {
                Globals.getLogger().logStack(Logger.ERROR,
                     BrokerResources.W_MESSAGE_STORE_FAILED,
                     msgid.toString(), ex);
                throw new BrokerException(
                        ex.toString(), ex);
            }
            isStored = true;
            
        }
    }

    public void store(Collection consumers) 
        throws BrokerException
    {        
    	if (destroyed || pktPtr == null) {
            return;
        }
        
        if (isStoredWithInterest) {
            // done already
            return;
        }
        
        boolean botherToStore=!neverStore && persist;

        ReturnInfo info = calculateConsumerInfo(consumers, botherToStore);
        if (ackInfo != null)  {
            ackInfo.putAll(info.ackInfo);
        } else {
            ackInfo = info.ackInfo;
        }
        interestCnt = info.ackInfo.size();
        
        if (!botherToStore) return;
        if (info.uids == null || info.uids.length == 0) {
            // nothing to store
            neverStore = true;
            return;
        }
            
        try {
            if (isStored && !neverStore && persist) {
                Globals.getStore().storeInterestStates(
                    destination,
                    msgid, info.uids, info.states, 
                    Destination.PERSIST_SYNC, getPacket());
            } else {
                assert pktPtr instanceof Packet || (pktPtr == null && destroyed) : "PktPtr is " + pktPtr.getClass();

                Globals.getStore().storeMessage(destination,
                        (Packet)pktPtr,
                         info.uids, info.states, Destination.PERSIST_SYNC);
                synchronized(this) {
                    makePacketSoftRef();
                }
            }
        } catch (IOException ex) {
                throw new BrokerException(
                    ex.toString(), ex);
        } catch (Exception ex) {
                Globals.getLogger().logStack(Logger.ERROR,
                    BrokerResources.W_MESSAGE_STORE_FAILED,
                    msgid.toString(), ex);
                throw new BrokerException(
                    ex.toString(), ex);
        } 
        isStored = true;
        isStoredWithInterest = true;
        assert interestCnt != 0;
               
    }
    
    public ConsumerUID[] getRoutingForStore(Collection consumers)
			throws BrokerException {

		if (destroyed || pktPtr == null) {
			return null;
		}

		boolean botherToStore = !neverStore && persist;
		if (!botherToStore)
			return null;

		if (consumers.isEmpty())
			return null;

		List<ConsumerUID> storedConsumers = null;
		Iterator itr = consumers.iterator();

		while (itr.hasNext()) {
			Object o = itr.next();
			ConsumerUID cuid = null;
			if (o instanceof Consumer) {
				cuid = ((Consumer) o).getStoredConsumerUID();
			} else {
				cuid = (ConsumerUID) o;
			}

			if (cuid.shouldStore()) {
				if (storedConsumers == null) {
					storedConsumers = new ArrayList<ConsumerUID>();
				}
				storedConsumers.add(cuid);
			}
		}
		if (storedConsumers == null)
			return null;

		return storedConsumers.toArray(new ConsumerUID[0]);

	}

    /**
     * this method is called from ???
     */
    public void add(Collection uids) {
        Iterator itr = uids.iterator();
        while (itr.hasNext()) {
           Object o = itr.next();
           ConsumerUID cuid = null, uid = null;
           Session ss = null;
           boolean fixcnt = false;

           if (o instanceof ConsumerUID) {
               cuid = (ConsumerUID)o;
           } else {
               cuid = ((Consumer)o).getStoredConsumerUID(); 
               uid = ((Consumer)o).getConsumerUID();
               if (uid != null) {
                   ss = Session.getSession(uid);
               }

           }

           if (uid != null && 
               (uid.getAckType() == Session.NONE ||
                ss != null && ss.isTransacted())) {
               fixcnt = true;
           }

           if (!fixcnt) interestCnt ++;

           if (ackInfo == null) {
               ackInfo = Collections.synchronizedMap(new HashMap());
           }

           ConsumerMessagePair cmp = new ConsumerMessagePair(cuid, false);
           cmp.setState(ROUTED);
           if (fixcnt) {
               if (ackInfo.get(cuid) == null) {
                   interestCnt ++;
               } else {
                   overriding();
               }
           }
           ackInfo.put(cuid, cmp);
        }
    }

    /**
     * called for messages which have already been
     * loaded from the database
     */
    public void update(ConsumerUID[] uids, int[] states) {
        update(uids, states, false);
    }

    public void update(ConsumerUID[] uids, int[] states, boolean store) {
        assert isStored;
        assert uids != null;
        assert states != null;
        assert uids.length == states.length;
        assert ackInfo == null;
        assert uids.length != 0;


        synchronized (this) {
            interestCnt +=uids.length;
        }
          

        for (int i=0; i < uids.length; i ++) {
           ConsumerUID cuid = uids[i];

           assert uids[i] != null;

           assert states[i] >= Store.INTEREST_STATE_ROUTED &&
                  states[i] <= Store.INTEREST_STATE_ACKNOWLEDGED;

           if (states[i] == Store.INTEREST_STATE_ACKNOWLEDGED) {
               // left over
               continue;
           }

           if (ackInfo == null)
               ackInfo = Collections.synchronizedMap(new HashMap());

           ConsumerMessagePair cmp = new
               ConsumerMessagePair(cuid, true);

           ackInfo.put(cuid, cmp);
           cmp.setState(states[i] == Store.INTEREST_STATE_ROUTED
                 ? ROUTED : CONSUMED);

           if (store) {
                try {
                   Globals.getStore().storeInterestStates(destination,
                         msgid,
                         uids, states, Destination.PERSIST_SYNC,
                         (Packet)pktPtr);
                } catch (Exception ex) {
                }
            }

        }

        assert interestCnt != 0;
            
    }

    public void debug(String prefix)
    {
        if (prefix == null)
            prefix = "";
        Globals.getLogger().log(Logger.INFO,prefix +"Message " + msgid);
        Globals.getLogger().log(Logger.INFO,prefix + "size " + ackInfo.size());
        Iterator itr = ackInfo.values().iterator();
        while (itr.hasNext()) {
            ConsumerMessagePair ae = (ConsumerMessagePair)itr.next();
            Globals.getLogger().log(Logger.INFO,prefix + "\t " + ae);
        }
    }
    

    /**
     * called when a consumer received
     * the specific message for delivery
     */
    public void routed(ConsumerUID intid) 
        throws BrokerException, IOException
    {

        // NOTE: the caller is responsible for
        // passing the "right" intid - for
        // queues this is the generic ID, for
        // durable subscribers, it is the subscription
        // ID

        if (destroyed || invalid) {
            Globals.getLogger().log(Logger.DEBUG,"route on destroyed ref "
                 + msgid + ":" + intid);
            return; // destroyed
        }


        // nothing to store
        // this just indicates that we've passed
        // the message off to a consumer

        // the state of the message should have
        // already been stored w/ routeTable

        // additional logic may be added in the future

    }

    public int getCompleteCnt() {
        return ackCnt + deadCnt;
    }

    public int getDeliverCnt() {
        return deliveredCnt;
    }

    

    public SysMessageID replacePacket(Hashtable props, byte[] bytes)
        throws BrokerException, IOException
    {
        //no messages can be delivered

        if (deliveredCnt > 0)
            throw new BrokerException("Unable to replace already delivered message");
        
        if (ackCnt > 0)
            throw new BrokerException("Unable to replace partially acknowledged message");

        Packet oldp = getPacket();
        Packet newp = new Packet();
        newp.fill(oldp);
        newp.updateSequenceNumber();
        newp.updateTimestamp();
        newp.setMessageBody(bytes);

        headers = null;
        Hashtable oldprops = null;
        try {
            oldprops = getProperties();
            if (oldprops == null)
                oldprops = new Hashtable();
            if (props != null) {
                Globals.getLogger().log(Logger.DEBUG,"Warning although properties "
                    + "have been changed on the message it will "
                    + "not be rerouted");
                oldprops.putAll(props);
            }
            oldprops.put("JMSOrigMessageID",
                (PREPEND_ID ? "ID:" : "") + msgid.toString());
            if (getHeaders() != null) {
                Hashtable ht = new Hashtable();
                HashMap headers = getHeaders();
                Iterator keys = headers.keySet().iterator();
                while (keys.hasNext()) {
                    String key = (String)keys.next();
                    Object value = headers.get(key);
                    if (value != null)
                        ht.put(key, value);
                }
                
            }
            newp.setProperties(oldprops);
        } catch (Exception ex) {
            Globals.getLogger().logStack(Logger.INFO,"Internal Error updating properties", ex);
        }

        setPacketObject(true /* soft*/, newp);
        if (isStoredWithInterest) {
            int cnt = 0;
            ConsumerUID uids[] = null;
            int states[] = null;
            if (ackInfo != null) {
                synchronized(ackInfo) {
                    // ok count stored entries
                    Iterator itr = ackInfo.values().iterator();
                    while (itr.hasNext()) {
                        ConsumerMessagePair ae = (ConsumerMessagePair)itr.next();
                        if (ae.stored) cnt ++;
                    }
                    uids= new ConsumerUID[cnt];
                    states = new int[cnt];
                    //ok allocate arrays
                    int i =0;
                    itr = ackInfo.values().iterator();
                    while (itr.hasNext()) {
                        ConsumerMessagePair ae = (ConsumerMessagePair)itr.next();
                        if (ae.stored) {
                            uids[i] = ae.uid;
                            states[i] = ae.state;
                        }
                        i ++;
                    }
                }
                Globals.getStore().storeMessage(destination,
                    (Packet)newp,
                    uids, states, Destination.PERSIST_SYNC);
            } else {
                Globals.getStore().storeMessage(destination,
                    (Packet)newp, Destination.PERSIST_SYNC);
            }
        } else if (isStored) {
            Globals.getStore().storeMessage(destination,
                (Packet)newp, Destination.PERSIST_SYNC);
        } else /* not stored */ {
        }
        setPacketObject(persist, newp);
        headers = null;
        props = null;
        SysMessageID id = msgid;
        this.msgid = (SysMessageID)newp.getSysMessageID().clone();
        if (isStored || isStoredWithInterest) {
            Globals.getLogger().log(Logger.DEBUG,"Cleaning up the old replaced message");
            Globals.getStore().removeMessage(destination,
                     id,Destination.PERSIST_SYNC );
        }
        return msgid;
        
    }

    /**
     * called just before the message is written
     * to the wire for a consumer
     */
    public boolean delivered(ConsumerUID intid, ConsumerUID storedid,
             boolean sync, boolean store) 
        throws BrokerException, IOException
    {
        // NOTE: the caller is responsible for
        // passing the "right" storedid - for
        // queues this is the generic ID, for
        // durable subscribers, it is the subscription
        // ID
        if (destroyed || invalid) {
            Globals.getLogger().log(Logger.DEBUG,"delivered on destroyed ref "
                 + msgid + ":" + storedid);
            return true; // destroyed
        }

        if (intid.isNoAck()) {
            // immediately ack message when delivered
            return acknowledged(intid, 
                   storedid,
                   sync, store);
        }

        ConsumerMessagePair cmp = getAck(storedid);

        if (cmp == null) {
            // nothing to do
            Globals.getLogger().log(Logger.DEBUG,"Received Unknown delivered:" 
                   +"\n\tStoreUID: " +  storedid 
                   +"\n\tConsumerUID: " + intid 
                   +"\n\tConsumer: " +  Consumer.getConsumer(intid));
            //if (DEBUG)
                debug("\t- ");
            return false;
        }

        // if we are greater than delivered

        if (cmp.compareStateLT(DELIVERED)) {
            synchronized (this) {
                deliveredCnt ++;
             }
        }
        cmp.setStateIfLess(DELIVERED, DELIVERED);        
        if (cmp.isStored() && store) {
                Globals.getStore().updateInterestState(
                    destination,
                    msgid, storedid, 
                    Store.INTEREST_STATE_DELIVERED, 
                    Destination.PERSIST_SYNC && sync,
                    null, false);
        }

        synchronized (this) {
            // in low memory, free ref explicity
            if (deliveredCnt >= interestCnt &&
                isStored &&
                ((MemoryGlobals.MEM_FREE_P_ACKED 
                   && persist ) ||
                  (MemoryGlobals.MEM_FREE_NP_ACKED
                   && (!persist)))) {
                   unload();
            }
        }

        return false;

    }

    public int getRedeliverCount(ConsumerUID intid) {
        ConsumerMessagePair cmp = getAck(intid);
        return cmp == null ? 0 : cmp.getRedeliverCount();
    }


    /**
     * called when the client indicates that
     * the message has been consumed (delivered
     * to the specific client side consumer).
     * This method may be triggered by a specific
     * message from the consumer OR when an ack
     * is received in autoack mode
     */
    public void consumed(ConsumerUID intid, boolean sync, boolean delivered) 
        throws BrokerException, IOException
    {
        if (destroyed || invalid) {
            Globals.getLogger().log(Logger.DEBUG,"consumed on destroyed ref "
                 + msgid + ":" + intid);
            return; // destroyed
        }

        if (delivered) {
            delivered(intid, intid, sync, true);
        }

        assert ackInfo != null;

        ConsumerMessagePair cmp = getAck(intid);

        assert cmp != null;

        // something went wrong, ignore
        if (cmp == null) {
           // nothing to do
           Globals.getLogger().log(Logger.ERROR,"Internal Error: unknown interest for " 
           + " consumed on " + msgid + intid+", "+getDestination()+", ackInfo="+ackInfo);
           return;
        }

        cmp.incrementRedeliver();
        cmp.setStateIfLess(CONSUMED, CONSUMED);

    }


    public boolean matches(ConsumerUID id) {
        return getAck(id) != null;
    }

    public boolean isAcknowledged(ConsumerUID id) {
        ConsumerMessagePair cmp = getAck(id);
        if (cmp == null) return true;
        return cmp.compareState(ACKED);
    }

    public boolean isDelivered(ConsumerUID id) {
        ConsumerMessagePair cmp = getAck(id);
        if (cmp == null) return true;
        return cmp.compareState(DELIVERED) || cmp.compareState(CONSUMED);
    }

    public boolean removeDelivered(ConsumerUID storedid, boolean decrementCounter)
    {
        if (destroyed || invalid) {
            return true; // destroyed
        }

        ConsumerMessagePair cmp = getAck(storedid);

        assert cmp != null;

        // something went wrong, ignore
        if (cmp == null) {
           // nothing to do
           Globals.getLogger().log(Logger.ERROR,"Internal Error: unknown interest for " 
               + " remove consumed on " + msgid
               + storedid);
           return false;
        }

        if (decrementCounter)
            cmp.decrementRedeliver();

        cmp.compareAndSetState(ROUTED, DELIVERED);

        //XXX use destination setting
        return (cmp.getRedeliverCount() >= 20);

    }


    public boolean hasConsumerAcked(ConsumerUID storedid)
    {
        try {
            if (destroyed || invalid) {
                return true; // destroyed
            }    
            ConsumerMessagePair cmp = getAck(storedid);

            
            if ( cmp != null && cmp.getState() != ACKED) {
                return false;
            }
        } catch (Throwable ex) {
            Globals.getLogger().logStack(Logger.ERROR,"Internal Error checking ack" +
                         " on " + msgid + " for " + storedid, ex);
            return false;
        }
        return true;
    }

    /**
     * handle updating acknowledgement of the message
     *
     * @returns if the message should be removed from the persistent list
     */
    public boolean acknowledged(ConsumerUID intid,ConsumerUID storedid,
                   boolean sync, boolean notIgnored) 
        throws BrokerException, IOException
    {
        return acknowledged(intid, storedid, sync, notIgnored, null, null, true);
    }

    /**
     * @param ackack whether client requested ackack
     */
    public boolean acknowledged(ConsumerUID intid,ConsumerUID storedid,
                   boolean sync, boolean notIgnored, boolean ackack)
                                throws BrokerException, IOException
    {
        return acknowledged(intid, storedid, sync, notIgnored, null, null, ackack);
    }


    /**
     * @param ackack whether client requested ackack
     *
     * If the ack is a remote non-transacted ack, waits for remote ack reply only if
     *    sync == true && notIgnored && ackack
     */
    public boolean acknowledged(ConsumerUID intid,ConsumerUID storedid,
                                boolean sync, boolean notIgnored, 
                                TransactionUID tuid, HashMap remoteNotified,
                                boolean ackack) 
                                throws BrokerException, IOException
    {
        Long txn = (tuid == null ? null : new Long(tuid.longValue()));
        try {
            if (destroyed || invalid) {
                return true; // destroyed
            }    

            removeInDelivery(storedid);
    
            ConsumerMessagePair cmp = getAck(storedid);
        
            // something went wrong, ignore
            if (cmp == null) {
               // nothing to do
               Globals.getLogger().log(Logger.ERROR,"Internal Error: Received Unknown ack " 
                                       + intid+ " for message "+getSysMessageID());
               Globals.getLogger().log(Logger.ERROR, "AckInfo" + ackInfo.toString());
               synchronized (this) {
                   return (ackCnt+deadCnt) >= interestCnt;
               }
            }

            // ok ... if setState == false, we were already
            // acked so do nothing
            if (cmp.setState(ACKED)) {
                    if (cmp.isStored()) {
                        boolean acked = false;
                        Store store = Globals.getStore();
                        if (store.isJDBCStore()) {
                            acked = store.hasMessageBeenAcked(destination, msgid);
                        }
                        if (!acked) {
                            try {
							// This test may fail to spot that this really is
							// the last ack because another thread processing an
							// ack for the same message may not have yet
							// incremented ackCnt.
							// However, it should not return a false positive so
							// should be safe
							boolean isLastAck = false;
							synchronized (this) {
								isLastAck = (ackCnt + 1 + deadCnt) >= interestCnt;
							}

							store.updateInterestState(destination, msgid,
									storedid,
									Store.INTEREST_STATE_ACKNOWLEDGED,
									Destination.PERSIST_SYNC && sync, tuid,
									isLastAck);

						} catch (Exception ex) {
                                if (ex instanceof BrokerException &&
                                    ((BrokerException)ex).getStatusCode() != Status.NOT_ALLOWED &&
                                    ((BrokerException)ex).getStatusCode() != Status.NOT_FOUND) {
                                    Globals.getLogger().log(Logger.WARNING, 
                                    "Update consumer "+storedid+" state failed for message "+
                                     msgid+": "+ ex.getMessage());
                                    throw ex;
                                } else {
                                    Globals.getLogger().log(Logger.DEBUGHIGH, 
                                    "Update consumer "+storedid+" state failed for message "+
                                     msgid+": "+ ex.getMessage());
                                }
                           }
                       }
                    }
                    if (!isLocal() ) { // not local
                        if (notIgnored) {
                            if (Globals.getClusterBroadcast().getClusterVersion() <
                                ClusterBroadcast.VERSION_410) {
                                Globals.getClusterBroadcast().
                                    acknowledgeMessage(getAddress(),
                                    getSysMessageID(), intid, 
                                    ClusterBroadcast.MSG_ACKNOWLEDGED, null, false);
                            } else if (txn == null) { 
                                Globals.getClusterBroadcast().
                                    acknowledgeMessage(getAddress(),
                                    getSysMessageID(), intid, 
                                    ClusterBroadcast.MSG_ACKNOWLEDGED, null, (sync && ackack));
                            } else {
                                BrokerAddress raddr = getAddress();
                                if (remoteNotified == null || remoteNotified.get(raddr) == null) {
                               
                                SysMessageID[] mysysids = { this.getSysMessageID() };
                                ConsumerUID[] myintids = { intid };
                                try {
                                Globals.getClusterBroadcast().
                                acknowledgeMessage2P(raddr,
                                                     mysysids, myintids,
                                                     ClusterBroadcast.MSG_ACKNOWLEDGED,
                                                     null, txn, true, !commit2pwait);
                                remoteNotified.put(raddr, "");
                                if (commit2pwait) {
                                    Globals.getTransactionList().completeClusterTransactionBrokerState(
                                             tuid,
                                             TransactionState.COMMITTED, raddr, true);
                                }

                                } catch (Exception e) {

                                if (e instanceof BrokerDownException) {
                                    remoteNotified.put(raddr, "");
                                }
                                if (DEBUG_CLUSTER_TXN) {
                                Globals.getLogger().logStack(Logger.WARNING,
                                    "Notify commit transaction ["+this.getSysMessageID()+
                                    ", "+intid+"]TUID="+txn+" got response: "+e.toString(),  e);
                                } else {
                                Globals.getLogger().log(Logger.WARNING,
                                    "Notify commit transaction ["+this.getSysMessageID()+
                                    ", "+intid+"]TUID="+txn+" got response: "+e.toString(),  e);
                                }

                                }
                                }
                            }
                        } else {
                            try {
                            Globals.getClusterBroadcast().
                                    acknowledgeMessage(getAddress(),
                                                       getSysMessageID(), 
                                                       intid, 
                                                       ClusterBroadcast.MSG_IGNORED, 
                                                       null, false /*no wait ack*/);
                            } catch (BrokerException e) {
                            Globals.getLogger().log(Logger.DEBUG, e.getMessage());
                            }
                        }
                    }
                    
            } else {
                    Consumer c = Consumer.getConsumer(intid);
                    if (c == null || !c.isValid()) {
// do we want to add a debug level message here
                            // got it while cleaning up
                            // this can only happen on topics and is a
                            // non-fatal error
                            // fixing the timing problem through synchronization
                            // adds too much overhead
                           synchronized (this) {
                               return (ackCnt+deadCnt) >= interestCnt;
                           }
                    } else {
                        Exception e = new Exception("double ack " + cmp);
                        e.fillInStackTrace();
                        Globals.getLogger().logStack(Logger.ERROR,"Internal Error: received ack twice " +
                             " on " + msgid + " for " + intid + " state is = " + cmp, e );
                        synchronized (this) {
                           return (ackCnt+deadCnt) >= interestCnt;
                        }
                    }
            }

            synchronized (this) {
                ackCnt ++;
                return (ackCnt+deadCnt) >= interestCnt;
            }
        } catch (Throwable thr) {
            if (thr instanceof BrokerDownException) {
                Globals.getLogger().log(Logger.WARNING, Globals.getBrokerResources().getKString(
                                        BrokerResources.W_UNABLE_PROCESS_REMOTE_ACK_BECAUSE,
                                        "["+msgid+"]"+intid+":"+storedid,  thr.getMessage()));
            } else {
                Globals.getLogger().logStack(Logger.ERROR, 
                "Error in processing ack" +" on "+ msgid+ " for " + intid, thr);
            }
            if (thr instanceof BrokerException) throw (BrokerException)thr;
            throw new BrokerException("Unable to process ack", thr);
        }
        
    }
    
    /**
     * this method is only  called when replaying the transaction log
     */
    public boolean acknowledgedOnReplay(ConsumerUID intid, ConsumerUID storedid) 
    throws BrokerException, IOException {
		try {
			if (destroyed || invalid) {
				return true; // destroyed
			}

			ConsumerMessagePair cmp = getAck(storedid);

			// something went wrong, ignore
			if (cmp == null) {
				// nothing to do
				Globals.getLogger().log(
						Logger.ERROR,
						"Internal Error: Received Unknown ack " + intid
								+ " for message " + getSysMessageID());
				Globals.getLogger().log(Logger.ERROR,
						"AckInfo" + ackInfo.toString());
				synchronized (this) {
					return (ackCnt + deadCnt) >= interestCnt;
				}
			}

			cmp.setState(ACKED);

			ackCnt++;
			return (ackCnt + deadCnt) >= interestCnt;

		} catch (Throwable thr) {
			Globals.getLogger().logStack(
					Logger.ERROR,
					"Error in processing ack" + " on " + msgid + " for "
							+ intid, thr);
			if (thr instanceof BrokerException)
				throw (BrokerException) thr;
			throw new BrokerException("Unable to process ack", thr);
		}

	}


    public void overrideRedeliver() {
        if (!overrideRedeliver) overrideRedeliver = true;
    }

    public boolean getRedeliverFlag(ConsumerUID intid)
    {
        if (destroyed || invalid) {
            Globals.getLogger().log(Logger.DEBUG,"redeliver for destroyed "
                 + msgid + ":" + intid);
            return true; // doesnt matter
        }

        ConsumerMessagePair cmp = getAck(intid);

        assert cmp != null;

        // something went wrong, ignore
        if (cmp == null) {
            return false;
        }

        if (overrideRedeliver)
            return true;

        // return true if our state is greater or equal to
        // DELIVERED
        return !cmp.compareStateLT(DELIVERED);
    }


    public boolean getConsumed(ConsumerUID intid) {
        if (destroyed || invalid) {
            Globals.getLogger().log(Logger.DEBUG,"getConsumed for destroyed "
                 + msgid + ":" + intid);
            return true; // doesnt matter
        }

        ConsumerMessagePair cmp = getAck(intid);

        // something went wrong, ignore
        if (cmp == null) {
           // nothing to do
           Globals.getLogger().log(Logger.ERROR,"Internal Error: unknown interest for " 
               + " getConsumed on " + msgid
               + intid);
           return true;
        }

        return cmp.compareState(CONSUMED);
    }



    public synchronized void clear() {
        props = null;
        if (pktPtr instanceof Reference) {
            ((Reference)pktPtr).clear();
            ((Reference)pktPtr).enqueue();
        }
        pktPtr = null;
        msgid = null;
    }

    public void remove(boolean onRollback) {
        if (isStored && !neverStore && persist) {
            try {
                //
                // removes synchronization when we remove
                // the message (we really dont need it)
                //
                // in the future we may evaulate changing the
                // code be smarter about syncing (if two
                // threads both change the file within a short
                // period of time, we really only need to sync once

                Globals.getStore().removeMessage(destination,
                       msgid, 
                      false /*Destination.PERSIST_SYNC*/, onRollback );
                isStored = false;

            } catch (IOException ex) {
                Globals.getLogger().logStack(Logger.ERROR,
                     BrokerResources.E_REMOVE_MSG_ERROR,
                     msgid.toString(), ex);
            } catch (BrokerException ex) {
                Globals.getLogger().logStack(Logger.ERROR,
                     BrokerResources.E_REMOVE_MSG_ERROR,
                     msgid.toString(), ex);
            }
            isStored = false;
        }
    }

    public void destroy() {
        assert getLBitSet() == false;
        synchronized (this) {
            if (destroyed) return;
            destroyed = true;
        }
        if (isStored) {
            try {
            	if(Globals.isNewTxnLogEnabled()) {
            	// With txnLog enabled, message will not have been persisted e.g. 
                // for a topic message with no durable subscribers.
            	// With txnLog, message is only stored if routed ( this is checked with neverStore)
            	   if (!neverStore) {
						Globals.getStore().removeMessage(destination, msgid,
								false);
					} else {
						if (Store.getDEBUG()) {
							Globals.getLogger().log(Logger.DEBUG,
									"NOT removing msg marked as neverstore");
						}
					}
            	}
            	else {
                	
            	// initial fix for bug 5025164
                // removes synchronization when we remove
                // the message (we really dont need it)
                //
                // in the future we may evaluate changing the
                // code be smarter about syncing (if two
                // threads both change the file within a short
                // period of time, we really only need to sync once
                Globals.getStore().removeMessage(destination, msgid, 
                      false /*Destination.PERSIST_SYNC*/ );
            	}
                isStored = false;

            } catch (IOException ex) {
                Globals.getLogger().logStack(Logger.ERROR,
                     BrokerResources.E_REMOVE_MSG_ERROR,
                     msgid.toString(), ex);
            } catch (BrokerException ex) {
                Globals.getLogger().logStack(Logger.ERROR,
                     BrokerResources.E_REMOVE_MSG_ERROR,
                     msgid.toString(), ex);
            }
        }
        props = null;
        if (pktPtr instanceof Reference) {
            ((Reference)pktPtr).clear();
            ((Reference)pktPtr).enqueue();
        }
        pktPtr = null;
    }


    void unload() {
        // clears out the reference
        if (pktPtr instanceof SoftReference) {
            ((SoftReference)pktPtr).clear();
        }
    }
    

//------------------------------------------------------------------
//
//          DMQ methods
//------------------------------------------------------------------

    public String getDeadComment() {
        if (lastDead == null)
            return null;
        ConsumerMessagePair cmp = getAck(lastDead);
        return cmp == null ? null: cmp.getDeadComment();
    }
    public int getDeadDeliverCnt() {
        if (lastDead == null)
            return getDeliverCnt();
        ConsumerMessagePair cmp = getAck(lastDead);
        return cmp == null ? getDeliverCnt(): cmp.getRedeliverCount();
    }
    public Reason getDeadReason() {
        if (lastDead == null)
            return null;
        ConsumerMessagePair cmp = getAck(lastDead);
        return cmp == null ? null: cmp.getDeadReason();
    }
    public String getDeadBroker() {
        if (lastDead == null)
            return null;
        ConsumerMessagePair cmp = getAck(lastDead);
        return cmp == null ? null: cmp.getDeadBroker();
    }
    public Throwable getDeadException() {
        if (lastDead == null)
            return null;
        ConsumerMessagePair cmp = getAck(lastDead);
        return cmp == null ? null: cmp.getDeadException();
    }

    public boolean markDead(ConsumerUID intid, ConsumerUID storedid,
                            String comment, Throwable ex, Reason reason,
                            int redeliverCnt, String broker) 
                            throws BrokerException {
        removeInDelivery(storedid);

        ConsumerMessagePair cmp = getAck(storedid);

        if (cmp == null) {
            // nothing to do
            Globals.getLogger().log(Logger.DEBUG,"Received unknown dead message " 
                   + storedid);
            return false;
        }

        if (!isLocal()) { // send remotely and ack
            Hashtable props = new Hashtable();
            if (comment != null)
                props.put(DMQ.UNDELIVERED_COMMENT, comment);
            if (redeliverCnt != -1)
                props.put(Destination.TEMP_CNT, new Integer(redeliverCnt));
            if (ex != null)
                props.put(DMQ.UNDELIVERED_EXCEPTION, ex);
            if (reason != null)
                props.put("REASON", new Integer(reason.intValue()));
            if (broker != null)
                props.put(DMQ.DEAD_BROKER, broker);

            Globals.getClusterBroadcast().
                acknowledgeMessage(getAddress(),
                getSysMessageID(), intid, 
                ClusterBroadcast.MSG_DEAD, props, true);
            cmp.setState(ACKED);
        } else {
            lastDead = storedid;
            cmp.setState(DEAD);
            cmp.setDeadComment(comment);
            cmp.setDeadReason(reason);
            cmp.setDeadException(ex);
            cmp.setDeadBroker(broker);
            if (redeliverCnt > -1)
               cmp.setRedeliverCount(redeliverCnt);
        }
        synchronized (this) {
            deadCnt ++;
            return (ackCnt + deadCnt >= interestCnt);
        }
    }

    public boolean isDead() {
        synchronized( this ) {
            return deadCnt > 0 && (ackCnt + deadCnt >= interestCnt);
        }
    }
    

/**
 * TEST CASE FOR VERIFYING REPLACEMEMENT - MUST BE VALIDATED MANUALLY
 **/
 
    public static void ReplaceTest() {
try {
        DestinationUID duid = new DestinationUID("test", true);
        DestinationUID duid_t = new DestinationUID("test", false);
        ConsumerUID uid1 = new ConsumerUID(1);
        ConsumerUID uid2 = new ConsumerUID(2);
        ConsumerUID[] uids = {uid1,uid2};

        Consumer consumer1 = new Consumer(duid,null, false,queueUID);
        ArrayList queues = new ArrayList();
        queues.add(consumer1);

        Consumer tconsumer1 = new Consumer(duid_t,null, false,uid1);
        Consumer tconsumer2 = new Consumer(duid_t,null, false,uid2);
        // simple Non-persistent

        ArrayList topics = new ArrayList();
        topics.add(tconsumer1);
        topics.add(tconsumer2);
        try {
        Destination.createDestination("test",DestType.DEST_TYPE_QUEUE);
        Destination.createDestination("test",DestType.DEST_TYPE_TOPIC);
        } catch (Exception ex) {};
        
        Packet p1 = new Packet();
        Hashtable props = new Hashtable();
        props.put("MyName","MyValue");

        byte[] body1 = {1};
        p1.setProperties(props);
        p1.setMessageBody(body1);
        p1.updateTimestamp();
        p1.updateSequenceNumber();

        System.out.println("----------------------------------");
        System.out.println("Test1: non-persist No Properties");
        System.out.println("----------------------------------");

        System.out.print("Initial Packet " + p1.getSysMessageID());
        // USE CASE 1 basic replace
        //
        // first create original packet
        // then create the packet reference
        // then store it
        // then replace the body
        // then check the contents
        // then clear out the reference and reload
        // then check the contents
        PacketReference ref =  createReference(p1, duid, null);
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get() +"]");
        ref.store(queues);

        byte[] body2 = {2};
        try {
            SysMessageID newid = ref.replacePacket(null, body2);
            System.out.print("New ID " + newid);
        } catch (Exception ex) {
           ex.printStackTrace();
        }
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");

        System.out.println("----------------------------------");
        System.out.println("Test2: persist No Properties");
        System.out.println("----------------------------------");
        p1 = new Packet();
        props = new Hashtable();
        props.put("MyName","MyValue");
        byte[] body3 = {3};
        byte[] body4 = {4};
        p1.setMessageBody(body3);
        p1.setPersistent(true);
        p1.setProperties(props);
        p1.updateTimestamp();
        p1.updateSequenceNumber();

        System.out.print("Initial Packet " + p1.getSysMessageID());
        ref =  createReference(p1, duid, null);
        ref.store(queues);
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");

        try {
            SysMessageID newid = ref.replacePacket(null, body4);
            System.out.print("New ID " + newid);
        } catch (Exception ex) {
           ex.printStackTrace();
        }
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");

        System.out.println("----------------------------------");
        System.out.println("Test3: Topic and 2 consumerUID");
        System.out.println("----------------------------------");
        p1 = new Packet();
        props = new Hashtable();
        props.put("MyName","MyValue");
        byte[] body5 = {5};
        byte[] body6 = {6};
        p1.setMessageBody(body5);
        p1.setPersistent(true);
        p1.setIsQueue(false);
        p1.setProperties(props);
        p1.updateTimestamp();
        p1.updateSequenceNumber();

        System.out.print("Initial Packet " + p1.getSysMessageID());
        ref =  createReference(p1, duid, null);
        ref.store(topics);
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");

        try {
            SysMessageID newid = ref.replacePacket(null, body6);
            System.out.print("New ID " + newid);
        } catch (Exception ex) {
           ex.printStackTrace();
        }
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");
        System.out.println("----------------------------------");
        System.out.println("Test4: Updating Properties");
        System.out.println("----------------------------------");
        p1 = new Packet();
        props = new Hashtable();
        props.put("MyName","MyValue");
        byte[] body7 = {7};
        byte[] body8 = {8};
        p1.setMessageBody(body7);
        p1.setPersistent(true);
        p1.setIsQueue(false);
        p1.setProperties(props);
        p1.updateTimestamp();
        p1.updateSequenceNumber();

        Hashtable newProps = new Hashtable();
        newProps.put("MyNewName","MyNewValue");
        newProps.put("MyName","***REPLACE***");
        System.out.print("Initial Packet " + p1.getSysMessageID());
        ref =  createReference(p1, duid, null);
        System.out.println(ref.getProperties());
        ref.store(topics);
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");

        try {
            SysMessageID newid = ref.replacePacket(newProps, body8);
            System.out.print("New ID " + newid);
        } catch (Exception ex) {
           ex.printStackTrace();
        }
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");
        System.out.println(ref.getProperties());
        System.out.println("----------------------------------");
        System.out.println("Test5: Message Delivered");
        System.out.println("----------------------------------");
        p1 = new Packet();
        props = new Hashtable();
        props.put("MyName","MyValue");
        byte[] body9 = {9};
        byte[] body10 = {10};
        p1.setMessageBody(body9);
        p1.setPersistent(true);
        p1.setIsQueue(false);
        p1.setProperties(props);
        p1.updateTimestamp();
        p1.updateSequenceNumber();

        newProps = new Hashtable();
        newProps.put("MyNewName","MyNewValue");
        newProps.put("MyName","***REPLACE***");
        ref =  createReference(p1, duid, null);
        ref.store(topics);
        ref.delivered(uid2, uid2, false, false);
        System.out.print("Initial Packet " + p1.getSysMessageID());
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");

        try {
            SysMessageID newid = ref.replacePacket(newProps, body10);
            System.out.print("New ID " + newid);
        } catch (Exception ex) {
           ex.printStackTrace();
        }
        System.out.println("\t[Body="+ref.getPacket().getMessageBodyByteBuffer().get()+"]");
        System.out.println(ref.getProperties());
} catch (Exception ex) {
ex.printStackTrace();
}
    }


    public static void main(String args[]) {
        System.out.println("RUNNING TEST");
        ReplaceTest();
    }
    
/**/
    

}

