/*
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
 *
 * Copyright 2000-2009 Sun Microsystems, Inc. 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.html
 * or mq/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 mq/legal/LICENSE.txt.  Sun designates
 * this particular file as subject to the "Classpath" exception as provided by
 * Sun in the GPL Version 2 section of the License file that accompanied this
 * code.  If applicable, add the following below the License Header, with the
 * fields enclosed by brackets [] replaced by your own identifying information:
 * "Portions Copyrighted [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. 
 */

/*
 * @(#)FalconProtocol.java	1.43 07/23/07
 */ 

package com.sun.messaging.jmq.jmsserver.multibroker.falcon;

import java.util.*;
import java.io.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.jmsserver.Broker;
import com.sun.messaging.jmq.jmsservice.BrokerEvent;
import com.sun.messaging.jmq.util.log.*;
import com.sun.messaging.jmq.io.*;
import com.sun.messaging.jmq.util.*;
import com.sun.messaging.jmq.util.selector.*;
import com.sun.messaging.jmq.jmsserver.core.*;
import com.sun.messaging.jmq.jmsserver.config.*;
import com.sun.messaging.jmq.jmsserver.resources.*;
import com.sun.messaging.jmq.jmsserver.util.*;
import com.sun.messaging.jmq.jmsserver.persist.Store;
import com.sun.messaging.jmq.jmsserver.service.ConnectionUID;
import com.sun.messaging.jmq.jmsserver.multibroker.MessageBusCallback;
import com.sun.messaging.jmq.jmsserver.multibroker.ClusterBrokerInfoReply;
import com.sun.messaging.jmq.jmsserver.multibroker.ClusterGlobals;
import com.sun.messaging.jmq.jmsserver.multibroker.Protocol;
import com.sun.messaging.jmq.jmsserver.multibroker.Cluster;
import com.sun.messaging.jmq.jmsserver.multibroker.CallbackDispatcher;
import com.sun.messaging.jmq.jmsserver.multibroker.BrokerInfo;
import com.sun.messaging.jmq.jmsserver.multibroker.raptor.RaptorProtocol;

public class FalconProtocol implements Protocol
{
    protected static final long ConsumerVersionUID = 99353142765567461L;

    private static boolean DEBUG = false;
    protected Logger logger = Globals.getLogger();
    protected BrokerResources br = Globals.getBrokerResources();

    protected static final boolean DEBUG_CLUSTER_ALL = 
        Globals.getConfig().getBooleanProperty(
                            Globals.IMQ + ".cluster.debug.all");

    protected static final boolean DEBUG_CLUSTER_LOCK = 
        Globals.getConfig().getBooleanProperty(
                            Globals.IMQ + ".cluster.debug.lock");

    protected static final boolean DEBUG_CLUSTER_CONN =
        Globals.getConfig().getBooleanProperty(
                            Globals.IMQ + ".cluster.debug.conn");

    protected MessageBusCallback cb = null;
    protected Cluster c = null;
    protected BrokerAddress selfAddress = null;
    protected CallbackDispatcher cbDispatcher = null;
    
    protected int version = ClusterGlobals.VERSION_300;
    protected HashMap brokerList = null;

    protected int lockTimeout = Globals.getConfig().getIntProperty(
            Globals.IMQ + ".cluster.locktimeout", 60);

    protected HashMap resTable = null;
    protected Random r = null;

    protected long startTime = 0;

    protected boolean configSyncComplete = false;
    protected boolean storeDirtyFlag = false;
    protected boolean isMasterBroker = false;

    protected Object eventLogLockObject;
    protected int eventLogStatus;
    protected long eventLogXid;
    protected long eventLogTimestamp;

    protected Object cfgSrvWaitObject = null;
    protected int cfgSrvRequestCount = 0;
    protected boolean cfgSrvRequestErr = false;

    protected Store store = null;

    // Falcon HA - The broker failover watchdog thread.
    protected HAWatchdog haWatchdog = null;

    public FalconProtocol(MessageBusCallback cb, Cluster c, 
            BrokerAddress myaddress)
       throws BrokerException
    {
        this.cb = cb;
        this.c = c;
        this.selfAddress = myaddress;
        this.cbDispatcher = new CallbackDispatcher(cb);
        store = Globals.getStore();
        resTable = new HashMap();
        r = new Random();
        brokerList = new HashMap();
        cfgSrvWaitObject = new Object();
        eventLogLockObject = new Object();
    }


    public int getHighestSupportedVersion() {
        return ClusterGlobals.VERSION_300;
    }

    /**
     * Get the message bus (cluster) protocol version used by this
     * broker.
     */
    public int getClusterVersion() {
        return version;
    }

    public Hashtable getDebugState() {
        Hashtable ht = new Hashtable();
        ht.put("protocol", "falcon");
        return ht;
    }

    public void handleGPacket(MessageBusCallback mbcb, BrokerAddress sender, GPacket pkt) {
        logger.logStack(Logger.ERROR, BrokerResources.E_INTERNAL_BROKER_ERROR, "Unexpected call",
                        new Exception("Unexpected call"));
    }

    public void preTakeover(String brokerID, UID storeSession, 
           String brokerHost, UID brokerSession) throws BrokerException {
        logger.logStack(Logger.ERROR, BrokerResources.E_INTERNAL_BROKER_ERROR, "Unexpected call",
                        new Exception("Unexpected call"));
        throw new BrokerException("Internal Error: not supported");
    }

    public void postTakeover(String brokerID, UID storeSession, boolean aborted) {
        logger.logStack(Logger.ERROR, BrokerResources.E_INTERNAL_BROKER_ERROR, "Unexpected call",
                        new Exception("Unexpected call"));
    }

    public void sendClusterTransactionInfo(long tid,
                com.sun.messaging.jmq.jmsserver.core.BrokerAddress to) {
        logger.logStack(logger.ERROR, "Internal error: unexpected call - not supported", new Exception(""));
    }

    public com.sun.messaging.jmq.jmsserver.core.BrokerAddress lookupBrokerAddress(String brokerid) {
        logger.logStack(logger.ERROR, "Internal error: unexpected call - not supported", new Exception(""));
        return null;
    } 

    public void receiveUnicast(BrokerAddress sender, GPacket pkt) {
        // This should never happen.
        logger.log(Logger.WARNING, "Protocol Mismatch. sender = " + sender);
        Thread.dumpStack();
    }

    public void receiveBroadcast(BrokerAddress sender, GPacket pkt) {
        logger.log(Logger.WARNING, "Protocol Mismatch. sender = " + sender);
        Thread.dumpStack();
    }

    /**
     * Receive a unicast packet.
     */
    public void receiveUnicast(BrokerAddress sender, int destId, byte []pkt)
    {
        switch (destId) {
        case ClusterGlobals.MB_MESSAGE_DATA:
            try {
                receiveMessage(sender, pkt);
            }
            catch (OutOfMemoryError oom) {
                Globals.handleGlobalError(oom,
                    br.getKString(br.M_LOW_MEMORY_CLUSTER));
                receiveMessage(sender, pkt);
            }
            break;

        case ClusterGlobals.MB_MESSAGE_ACK:
            receiveMessageAck(sender, pkt);
            break;

        case ClusterGlobals.MB_INTEREST_UPDATE:
            receiveInterestUpdate(sender, pkt, false);
            break;

        case ClusterGlobals.MB_LOCK_RESPONSE:
            receiveLockResponse(sender, pkt);
            break;

        case ClusterGlobals.MB_CONFIG_CHANGE_EVENT:
            receiveConfigChangeEvent(sender, pkt);
            break;

        case ClusterGlobals.MB_CONFIG_CHANGE_EVENT_ACK:
            receiveConfigChangeEventAck(sender, pkt);
            break;

        case ClusterGlobals.MB_GET_CONFIG_CHANGES_REQUEST:
            receiveConfigChangesRequest(sender, pkt);
            break;

        case ClusterGlobals.MB_GET_CONFIG_CHANGES_RESPONSE:
            receiveConfigChangesResponse(sender, pkt);
            break;

        case ClusterGlobals.MB_REQUEST_INTEREST_UPDATE:
            receiveRequestInterestUpdate(sender, pkt);
            break;

        default:
            logger.log(Logger.WARNING, br.W_MBUS_UNKNOWN_DESTID1,
                Integer.toString(destId));
            break;
        }
    }

    public void receiveBroadcast(BrokerAddress sender,
        int destId, byte []pkt) {
        receiveBroadcast(sender, destId, pkt, false);
    }

    /**
     * Receive a broadcast packet.
     */
    public void receiveBroadcast(BrokerAddress sender,
        int destId, byte []pkt, boolean configSyncResponse)
    {
        switch (destId) {
        case ClusterGlobals.MB_INTEREST_UPDATE:
            receiveInterestUpdate(sender, pkt, configSyncResponse);
            break;

        case ClusterGlobals.MB_LOCK_REQUEST:
            receiveLockRequest(sender, pkt);
            break;

        case ClusterGlobals.MB_CLIENT_CLOSED:
            receiveClientClosedUpdate(sender, pkt);
            break;

        case ClusterGlobals.MB_DESTINATION_UPDATE:
            receiveDestinationUpdate(sender, pkt);
            break;

        case ClusterGlobals.MB_RESET_PERSISTENCE:
            receiveResetPersistence(sender, pkt);
            break;

        case ClusterGlobals.MB_RESTART_CLUSTER:
            receiveRestartCluster(sender, pkt);
            break;

        case ClusterGlobals.MB_HA_ACTIVE_UPDATE:
            if (haWatchdog != null)
                haWatchdog.handleActiveBrokerUpdate(sender, pkt);
            // else
                // TBD: Warning.
            break;

        default:
            logger.log(Logger.WARNING, br.W_MBUS_UNKNOWN_DESTID2,
                Integer.toString(destId));
            break;
        }
    }


    /**
     * Construct a BrokerInfo object that describes this broker.
     * This object is exchanged during initial handshake between
     * brokers.
     * @return BrokerInfo object describing the current state of the broker.
     */
    public BrokerInfo getBrokerInfo() {
        BrokerInfo selfInfo = new BrokerInfo();
        selfInfo.setBrokerAddr(selfAddress);
        selfInfo.setStartTime(startTime);
        selfInfo.setStoreDirtyFlag(storeDirtyFlag);

        return selfInfo;
    }

	public ClusterBrokerInfoReply getBrokerInfoReply(BrokerInfo remote) throws Exception {
        throw new BrokerException("Unexpected call", Status.NOT_IMPLEMENTED);  
    }

    /**
     * Set the matchProps for the cluster.
     */
    public void setMatchProps(Properties matchProps) {
        c.setMatchProps(matchProps);
    }

    /**
     * Start the I/O operations on the cluster, now that the upper
     * layers are ready to receive notifications.
     */
    public void startClusterIO() {
        // If this broker is the config server, make sure it's own
        // persistent store is in sync with the config server
        // change record.
        try {
            BrokerAddress configServer = c.getConfigServer();
            if (configServer == null) { // No config server...
                configSyncComplete = true;
            }
            else if (configServer.equals(selfAddress)) {
                // I am the config server...
                initConfigServer();

                long timestamp = -1;
                timestamp = getLastRefreshTime();
                sendConfigChangesRequest(selfAddress, timestamp);

                // At this point the persistent store is in sync..
                //
                // Obviously, this is not very efficient - we can
                // avoid all the marshalling / unmarshalling since
                // the target is selfAddress.
            }
        }
        catch (Exception e) {}

        // Falcon HA -
        if (Globals.getHAEnabled())
            haWatchdog = new HAWatchdog(cb, cbDispatcher, c, this);
    }

    public void stopClusterIO(boolean requestTakeover) {
        cbDispatcher.shutdown();
    }

    private void initConfigServer() {
        isMasterBroker = true;
        logger.log(Logger.INFO, br.I_MBUS_I_AM_MASTER);

        // The first change record must be ' ClusterGlobals.MB_RESET_PERSISTENCE'.
        try {
            Object lists[] = store.getAllConfigRecords();
            ArrayList records = (ArrayList) lists[1];
            // TBD: There should be a persistence store method to
            // get the size directly. Even though getAllConfigRecords
            // does a shallow clone, this is still quite ugly..
            // Fix this after 2.0 FCS (RTM)...

            if (records.size() == 0) {
                logger.log(Logger.INFO, br.I_MBUS_MASTER_INIT);

                byte[] resetEvent = prepareResetPersistenceRecord();
                store.storeConfigChangeRecord(System.currentTimeMillis(),
                    resetEvent, false);
            }
        }
        catch (Exception e) {
            // TBD: log error.
        }
    }

    /**
     * Handle jmq administration command to reload and update
     * the cluster configuration.
     */
    public void reloadCluster() {
        logger.log(Logger.INFO, br.I_MBUS_RELOAD_CLS);

        sendRestartCluster(); // Tell other brokers.
        c.reloadCluster();
    }

    /**
     * Stop the cluster message inflow.
     */
    public void stopMessageFlow() throws IOException {
        c.stopMessageFlow();
    }

    /**
     * Resume the cluster message inflow.
     */
    public void resumeMessageFlow() throws IOException {
        c.resumeMessageFlow();
    }


    /**
     *
     */
    public boolean waitForConfigSync() {
        BrokerAddress configServer = null;
        try {
            configServer = c.getConfigServer();
        }
        catch (Exception e) {
            return true; // There is config server but it's unreachable.
        }

        if (configServer == null)
            return false; // There is no config server.

        if (configServer.equals(selfAddress))
            return false; // I am the config server.

        return (! configSyncComplete); // Waiting for sync complete..
    }

    /**
     * Deliver the message.
     *
     * Constructs and sends  ClusterGlobals.MB_MESSAGE_DATA packets.
     */
    public void sendMessage(PacketReference pkt, Collection targets,
        boolean sendMsgDeliveredAck) {
        HashMap h = new HashMap();

        if (DEBUG) {
            logger.log(Logger.DEBUGMED,
                "MessageBus: sending message {0} to {1} targets.",
                pkt.getSysMessageID(),
                Integer.toString(targets.size()));
        }

        String debugString = "\n";

        Iterator itr = targets.iterator();
        while (itr.hasNext()) {
            // TBD: Revisit - Send the  ClusterGlobals.MB_MSG_SENT ack
            // instead of calling Interest.messageSent() ???
            Consumer target = (Consumer)itr.next();

            BrokerAddress baddr =
                target.getConsumerUID().getBrokerAddress();

            ArrayList v = (ArrayList) h.get(baddr);
            if (v == null) {
                v = new ArrayList();
                h.put(baddr, v);
            }
            v.add(target);
        }

        if (DEBUG) {
            logger.log(Logger.DEBUGHIGH,
                "MessageBus: Local Targets = {0}", debugString);
        }

        // Now deliver message to other target brokers. The target
        // list may contain multiple entries with same BrokerAddress,
        // Combine all such entries and send the packet only once.

        Packet roPkt = null;
        Iterator brokers = h.entrySet().iterator();
        while (brokers.hasNext()) {
            Map.Entry entry = (Map.Entry)brokers.next();
            BrokerAddress b = (BrokerAddress) entry.getKey();
            ArrayList v = (ArrayList) entry.getValue();
            
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            DataOutputStream dos = new DataOutputStream(bos);

            debugString = "\n";

            try {
                dos.writeInt(version);
                dos.writeInt(v.size());
                dos.writeBoolean(sendMsgDeliveredAck);

                for (int i = 0; i < v.size(); i++) {
                    ConsumerUID intid = ((Consumer) v.get(i)).getConsumerUID();
                    ConsumerUID storeid = ((Consumer) v.get(i)).getStoredConsumerUID();
                    try {
                        pkt.delivered(intid, storeid,  
                           intid.isUnsafeAck(), true);
                    } catch (Exception ex) {
                        logger.logStack(Logger.WARNING, BrokerResources.E_INTERNAL_BROKER_ERROR,
                           "saving redeliver flag for " +
                           pkt.getSysMessageID() + " to " + intid, ex);
                    }
                    writeConsumerUID(intid, dos);
                    if (DEBUG) {
                        debugString = debugString + "\t" + intid + "\n";
                    }
                }
                if (roPkt == null)
                    roPkt = pkt.getPacket();
                roPkt.generateTimestamp(false);
                roPkt.generateSequenceNumber(false);
                roPkt.writePacket(dos);
                dos.flush();
                bos.flush();
            }
            catch (Exception e) {
                logger.logStack(Logger.DEBUG,"Exception writing packet ", e);
            }

            byte[] buf = bos.toByteArray();

            try {
                c.unicast(b,  ClusterGlobals.MB_MESSAGE_DATA, buf, true);
                if (DEBUG) {
                    logger.log(Logger.DEBUGHIGH,
                        "MessageBus: Broker {0} Targets = {1}",
                        b, debugString);
                }
            }
            catch (IOException e) {
                // This exception means that there is no way to
                // deliver this message to the target. If there was
                // an alternate path, the cluster implementation would
                // have automatically used it...
                for (int i = 0; i < v.size(); i++) {
                    ConsumerUID intid = ((Consumer) v.get(i)).getConsumerUID();
                    try {
                    cb.processRemoteAck(pkt.getSysMessageID(), intid,
                                        ClusterBroadcast.MSG_IGNORED, null);
                    } catch (Exception ex) { //XXX
                    logger.log(logger.WARNING, ex.getMessage(), ex);
                    }
                }
                if (DEBUG) {
                    logger.log(Logger.DEBUGHIGH,
    "FalconProtocol: Could not deliver message to broker {0}", b);
                }
            }
        }
    }
    protected void writeConsumer(Consumer consumer, 
                DataOutputStream dos)
          throws IOException
    {
        String destName = consumer.getDestinationUID().getName();
        ConsumerUID id = consumer.getConsumerUID();
        String durableName = null;
        String clientID = null;
        String selstr = consumer.getSelectorStr();
        boolean noLocalDelivery = consumer.getNoLocal();
        boolean isQueue = consumer.getDestinationUID().isQueue();
        boolean isReady = true;
 
        if (consumer.getSubscription() != null) {
            durableName = consumer.getSubscription().getDurableName();
            clientID = consumer.getSubscription().getClientID();
            if (!consumer.getSubscription().isActive())
                isReady = false;
        }
        dos.writeLong(ConsumerVersionUID); // version
        dos.writeUTF(destName);
        dos.writeBoolean(id != null);
        if (id != null) {
            writeConsumerUID(id, dos);
        }
        dos.writeBoolean(clientID != null);
        if (clientID != null) {
            dos.writeUTF(clientID);
        }
        dos.writeBoolean(durableName != null);
        if (durableName != null) {
            dos.writeUTF(durableName);
        }
        dos.writeBoolean(selstr != null);
        if (selstr != null) {
            dos.writeUTF(selstr);
        }
        dos.writeBoolean(isQueue);
        dos.writeBoolean(noLocalDelivery);
        dos.writeBoolean(isReady);
        dos.writeBoolean(true);
    }

    public  void writeConsumerUID(ConsumerUID uid,
                DataOutputStream dos)
          throws IOException 
    {
        dos.writeLong(uid.longValue()); // UID write
        dos.writeLong(uid.getConnectionUID().longValue());
        BrokerAddress brokeraddr= uid.getBrokerAddress();
        if (brokeraddr == null)
            brokeraddr = Globals.getMyAddress();
        brokeraddr.writeBrokerAddress(dos); // UID write
    }


    public static Consumer readConsumer(DataInputStream dis,
        CallbackDispatcher cbDispatcher)
          throws IOException
    {
        Logger logger = Globals.getLogger();
        ConsumerUID id = null;
        String destName = null;
        String clientID = null;
        String durableName = null;
        String selstr = null;
        boolean isQueue;
        boolean noLocalDelivery;
        boolean consumerReady;

        long ver = dis.readLong(); // version
        if (ver != ConsumerVersionUID) {
            throw new IOException("Wrong Consumer Version " + ver + " expected " + ConsumerVersionUID);
        }
        destName = dis.readUTF();
        boolean hasId = dis.readBoolean();
        if (hasId) {
            id = readConsumerUID(dis);
        }
        boolean hasClientID = dis.readBoolean();
        if (hasClientID) {
            clientID = dis.readUTF();
        }
        boolean hasDurableName = dis.readBoolean();
        if (hasDurableName) {
            durableName = dis.readUTF();
        }

        boolean hasSelector = dis.readBoolean();
        if (hasSelector) {
            selstr = dis.readUTF();
        }

        isQueue = dis.readBoolean();
        noLocalDelivery = dis.readBoolean();
        consumerReady = dis.readBoolean();


        try {
            DestinationUID dest = DestinationUID.getUID(destName, isQueue);
            Consumer c = new Consumer(dest, selstr, noLocalDelivery,
                    id);
            if (durableName != null) {
                Subscription sub = Subscription.findCreateDurableSubscription
                       (clientID,durableName, dest, selstr, noLocalDelivery);

                int type = (dest.isQueue() ? DestType.DEST_TYPE_QUEUE :
                     DestType.DEST_TYPE_TOPIC);

                // Make sure that the destination exists...
                Destination tmpdest = Destination.getDestination(
                    dest.getName(),
                    type, true, true);

                if (cbDispatcher != null)
                    cbDispatcher.interestCreated(sub);

                sub.attachConsumer(c);
             }
             return c;
         } catch (SelectorFormatException ex) {
             logger.log(Logger.INFO,"L10N-XXX Got bad selector["+selstr + "] " , ex);
             throw new IOException("bad selector " + selstr);
         } catch (BrokerException ex) {
             logger.log(Logger.INFO,"L10N-XXX error creating consumer ", ex);
             throw new IOException("error creating consumer ");
         }

    }


    protected static ConsumerUID readConsumerUID(DataInputStream dis)
          throws IOException
    {
        long id = dis.readLong(); // UID write
        ConnectionUID conuid = new ConnectionUID(dis.readLong());
        BrokerAddress tempaddr = Globals.getMyAddress();
        if (tempaddr == null) {
            // XXX Revisit and cleanup : This method may be called
            // before cluster initialization only during persistent
            // store upgrade. i.e. from -
            // FalconProtocol.upgradeConfigChangeRecord()
            // At that time, Globals.getMyAddress() returns null.
            // Hence this kludge...
            try {
                tempaddr =
                    new com.sun.messaging.jmq.jmsserver.multibroker.fullyconnected.BrokerAddressImpl();
            }
            catch (Exception e) {}
        }


        BrokerAddress brokeraddr = (BrokerAddress)tempaddr.clone();
        brokeraddr.readBrokerAddress(dis); // UID write
        ConsumerUID cuid = new ConsumerUID(id);
        cuid.setConnectionUID(conuid);
        cuid.setBrokerAddress(brokeraddr);
        return cuid;
    }


    protected  void writeDestination(Destination d,
                DataOutputStream dos)
          throws IOException
    {
            dos.writeUTF(d.getDestinationName());
            dos.writeInt(d.getType());
    }

    protected Destination readDestination( DataInputStream dis)
        throws IOException,BrokerException
    {
        String name = dis.readUTF();
        int type = dis.readInt();
        DestinationUID duid = DestinationUID.getUID(name, type);

        Destination d = Destination.getDestination(duid);
        if (d == null) {
            d = Destination.createDestination(name,type, !DestType.isTemporary(type),
               false, selfAddress);
            cbDispatcher.notifyCreateDestination(d);
        }
        return d;
    }


    /**
     * Receive and process an  ClusterGlobals.MB_MESSAGE_DATA packet. This is where
     * JMS messages forwarded by other brokers are picked up.
     */
    private void receiveMessage(BrokerAddress sender, byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        ArrayList targetVector = new ArrayList();
        ArrayList ignoreVector = new ArrayList();

        Packet roPkt;
        PacketReference ref = null;
        boolean sendMsgDeliveredAck;

        if (DEBUG) {
            logger.log(Logger.DEBUGMED, "MessageBus: receiving message.");
        }

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING,  br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_MESSAGE_DATA));
                return;
            }

            int n = dis.readInt();
            sendMsgDeliveredAck = dis.readBoolean();

            for (int i = 0; i < n; i++) {
                ConsumerUID intid = readConsumerUID(dis);

                Consumer interest = Consumer.getConsumer(intid);

                if (interest != null) {
                    // we need the interest for updating the ref
                    targetVector.add(interest);
                } else {
                    ignoreVector.add(intid);
                }
            }

            roPkt = new Packet(false);
            roPkt.generateTimestamp(false);
            roPkt.generateSequenceNumber(false);
            roPkt.readPacket(dis);

            boolean exists = false;
            ref =Destination.get(roPkt.getSysMessageID());
            if (ref != null) {
                exists = true;
            } else {
                try {
                    ref = PacketReference.createReference(roPkt, null);
                } catch (BrokerException ex) {
                   // should never happen, UID should already be valid
                   Globals.getLogger().log(Logger.ERROR, "Internal error:"
                        +ex.getMessage(), ex);
                   return;
                }
                ref.setBrokerAddress(sender);
            }
            // XXX - note, we send a reply for all message delivered
            //       acks, that is incorrect
            //       really, we should have a sendMsgDeliveredAck per
            //       consumer
            if (sendMsgDeliveredAck) {
                for (int i=0; i < targetVector.size(); i ++) {
                    Consumer c = (Consumer)targetVector.get(i);
                    ref.addMessageDeliveredAck(c.getConsumerUID());
                }
            }
            try {
                if (ref == null)  {
                 
                    return;
                }
                Destination d = Destination.getDestination(ref.getDestinationUID().getName(), 
                     ref.getDestinationUID().isQueue() ? DestType.DEST_TYPE_QUEUE
                       : DestType.DEST_TYPE_TOPIC, true, true);
                if (d == null) {
                   ignoreVector.addAll(targetVector); 
                } else {
                    if (!exists) {
                        d.queueMessage(ref, false); // add to message count
                        ref.setNeverStore(true);
                        // OK .. we dont need to route .. its already happened
                        ref.store(targetVector);
                    } else {
                        ref.add(targetVector);
                    }
                }
            } catch (BrokerException ex) {
                // unable to store
                ignoreVector.addAll(targetVector); 
            }
        }
        catch (IOException e) {
            logger.log(Logger.INFO,"Internal Exception, unable to process message " +
                       ref, e);
            return;
        }

        String debugString = "\n";

        // Now deliver the message...
        int i;
        for (i = 0; i < targetVector.size(); i++) {
            Consumer interest = (Consumer)targetVector.get(i);

            interest.routeMessage(ref, false);


            if (DEBUG) {
                debugString = debugString +
                    "\t" + interest.getConsumerUID() + "\n";
            }
        }

        if (DEBUG) {
            logger.log(Logger.DEBUGHIGH,
                "MessageBus: Delivering message to : {0}",
                debugString);
        }

        debugString = "\n";
        // Finally, send  ClusterGlobals.MB_MSG_IGNORED acks if any...
        for (i = 0; i < ignoreVector.size(); i++) {
            sendMessageAck(sender, ref.getSysMessageID(),
                (ConsumerUID) ignoreVector.get(i),  ClusterGlobals.MB_MSG_IGNORED, null, false);

            if (DEBUG) {
                debugString = debugString +
                    "\t" + ignoreVector.get(i) + "\n";
            }
        }

        if (DEBUG) {
            if (ignoreVector.size() > 0)
                logger.log(Logger.DEBUGHIGH,
                    "MessageBus: Invalid targets : {0}",
                    debugString);
        }
    }

    public void sendMessageAck2P(BrokerAddress msgHome, SysMessageID[] sysids,
                                 ConsumerUID[] intids, int ackType, 
                                 Map optionalProps, Long txnID, boolean ackack, boolean async) 
                                 throws BrokerException {
        throw new BrokerException("Broker Internal Error: sendMessageAck2P not supported"); 
    }

    /**
     * Acknowledge a message to the message home broker. We will have to
     * somehow keep track of message home BrokerAddress for every
     * message coming from message bus. This may be done either using
     * the Packet trailer or by maintaining a SysMessageID to
     * message home (BrokerAddress) mapping.
     *
     * Constructs and sends  ClusterGlobals.MB_MESSAGE_ACK packets.
     */
    public void sendMessageAck(BrokerAddress msgHome, SysMessageID sysid,
                               ConsumerUID intid, int ackType, Map optionalProps, boolean ackack) {
        if (DEBUG) {
            String debugString =
                "\n\tackType = " + ackType +
                "\n\tSysMessageID = " + sysid+
                "\n\tConsumerUID = " + intid +
                "\n\tMessageHome = " + msgHome + "\n";
            logger.log(Logger.DEBUGHIGH,
                "MessageBus: Sending message acknowledgement : {0}",
                debugString);
        }


        // Else send the acknowledgement to the message home.
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);

            sysid.writeID(dos);
            writeConsumerUID(intid, dos);
            dos.writeInt(ackType);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {
            return;
        }

        byte[] buf = bos.toByteArray();

        try {
            c.unicast(msgHome,  ClusterGlobals.MB_MESSAGE_ACK, buf);
        }
        catch (IOException e) {}
    }

    /**
     * Receive and process an  ClusterGlobals.MB_MESSAGE_ACK packet.
     */
    private void receiveMessageAck(BrokerAddress sender, byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        SysMessageID sysid;
        ConsumerUID intid;
        int ackType;

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING,  br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_MESSAGE_ACK));
                return;
            }

            sysid = new SysMessageID();
            sysid.readID(dis);
            intid = readConsumerUID(dis);

            ackType = dis.readInt();
        }
        catch (Exception e) {
            logger.logStack(Logger.DEBUG,"Exception reading packet ", e);
            return;
        }

        if (DEBUG) {
            String debugString =
                "\n\tackType = " + ackType +
                "\n\tSysMessageID = " + sysid +
                "\n\tConsumerUID = " + intid +
                "\n\tSender = " + sender + "\n";

            logger.log(Logger.DEBUGHIGH,
                "MessageBus: Received message acknowledgement : {0}",
                debugString);
        }

        try {
        cb.processRemoteAck(sysid, intid, ackType, null);
        } catch (Exception e) { //XXX
        logger.log(logger.WARNING, e.getMessage(), e);
        }
    }

    /**
     * Prepares an  ClusterGlobals.MB_INTEREST_UPDATE packet.
     */
    private byte[] prepareInterestUpdate(Collection ints, int type) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeInt(type);
            dos.writeInt(ints.size());

            if (type ==  ClusterGlobals.MB_NEW_INTEREST) {
                Iterator itr = ints.iterator();
                while (itr.hasNext()) {
                    Consumer c = (Consumer)itr.next();
                    if (c instanceof Subscription || c.getSubscription()
                       != null) {
                       Subscription s = null;
                       if (c instanceof Subscription) {
                           s = (Subscription)c;
                       } else {
                           s = c.getSubscription();
                       }
                       dos.writeBoolean(true);
                       dos.writeUTF(s.getDurableName());
                       dos.writeUTF(s.getClientID());
                    } else {
                       dos.writeBoolean(false);
                    }
                }

                itr = ints.iterator();
                while (itr.hasNext()) {
                    Consumer c = (Consumer)itr.next();
                    writeConsumer(c, dos);
                }

            }
            else if (type ==  ClusterGlobals.MB_REM_DURABLE_INTEREST) {
                Iterator itr = ints.iterator();
                while (itr.hasNext()) {
                    Consumer c = (Consumer)itr.next();
                    Subscription s = null;
                    if (c instanceof Subscription) {
                        s = (Subscription)c;
                    } else {
                        s = c.getSubscription();
                    }
                    dos.writeUTF(s.getDurableName());
                    dos.writeUTF(s.getClientID());
                }
            }
            else {
                Iterator itr = ints.iterator();
                while (itr.hasNext()) {
                    Consumer c = (Consumer)itr.next();
                    writeConsumerUID(c.getConsumerUID(), dos);
                }
            }

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        return bos.toByteArray();
    }

    /**
     * Broadcast an interest update to the cluster.
     */
    public void sendInterestUpdate(Consumer interest, int type) {
        Set s = new HashSet();
        s.add(interest);
        sendInterestUpdate(s, type);
    }

    public void sendInterestUpdate(Collection consumers , int type) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
    "MessageBus: Broadcasting interest update. Count = {0}, Type = {1}",
                Integer.toString(consumers.size()),
                Integer.toString(type));
        }
        byte[] buf = prepareInterestUpdate(consumers, type);
        try {
            c.broadcast( ClusterGlobals.MB_INTEREST_UPDATE, buf);
        }
        catch (IOException e) {}
    }

    /**
     * Send an interest update to a specific broker. Usually called
     * to bring a new broker upto the speed.
     */
    private void sendInterestUpdate(BrokerAddress to,
        Collection consumers, int type) {

        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "MessageBus: Sending interest update to {0}. Count = {1}",
                to, Integer.toString(consumers.size()));
        }

        byte[] buf = prepareInterestUpdate(consumers, type);
        try {
            c.unicast(to,  ClusterGlobals.MB_INTEREST_UPDATE, buf);
        }
        catch (IOException e) {}
    }

    /**
     * Receive and process  ClusterGlobals.MB_INTEREST_UPDATE packet.
     */
    private void receiveInterestUpdate(BrokerAddress sender, byte[] pkt,
        boolean configSyncResponse) {

        ConsumerUID intid;

        if (configSyncComplete == false && !configSyncResponse) {
            // Do not accept the normal interest updates before
            // config sync is complete. Here is 
            if (DEBUG) {
                logger.log(Logger.DEBUG,
    "MessageBus: Dropping the  ClusterGlobals.MB_INTEREST_UPDATE. Not ready yet.");
            }
            return;
        }

        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING,  br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_INTEREST_UPDATE));
                return;
            }

            int type = dis.readInt();
            int count = dis.readInt();

            if (DEBUG) {
                logger.log(Logger.DEBUG,
    "MessageBus: Receiving interest update from {0}. Type = {1}",
                    sender, Integer.toString(type));
            }

            switch (type) {
            case  ClusterGlobals.MB_NEW_INTEREST:

                for (int i = 0; i < count; i++) {
                    if (dis.readBoolean()) { // is durable?
                        dis.readUTF(); // durable name
                        dis.readUTF(); // client id
                    }
                }

                for (int i = 0; i < count; i++) {
                    Consumer c  = readConsumer(dis, cbDispatcher);


                    // If this is a config sync response, and we already
                    // know about the interest, ignore this notification.
/* LKS-XXX ... do we need to worry ... if its a dup so what ?
                    if (configSyncResponse && interest.isDurable()) {
                        Consumer intr = im.getDurableInterest(
                            interest.getDurableName(),
                            interest.getClientID());

                        if (intr != null) {
                            continue;
                        }
                    }
*/

                    cbDispatcher.interestCreated(c);
                }
                break;

            case  ClusterGlobals.MB_REM_INTEREST:
                for (int i = 0; i < count; i++) {
                    intid  = readConsumerUID(dis);
                    Consumer c = Consumer.getConsumer(intid);
                    if (c != null) {
                        cbDispatcher.interestRemoved(c, null, false);
                    }
                }
                break;

            case  ClusterGlobals.MB_REM_DURABLE_INTEREST:
                for (int i = 0; i < count; i++) {
                    String dname = dis.readUTF();
                    String cname = dis.readUTF();
                    Subscription intr = Subscription.findDurableSubscription(cname, dname);
                    if (intr != null) {
                        cbDispatcher.interestRemoved(intr, null, false);
                    }
                }
                break;

            case  ClusterGlobals.MB_DURABLE_DETACH:
                for (int i = 0; i < count; i++) {
                    intid = readConsumerUID(dis);
                    Consumer interest = Consumer.getConsumer(intid);
                    cbDispatcher.interestRemoved(interest, null, false);
                }
                break;

            case  ClusterGlobals.MB_NEW_PRIMARY_INTEREST:
                for (int i = 0; i < count; i++) {
                    intid = readConsumerUID(dis);
                    Consumer interest = Consumer.getConsumer(intid);
                    if (interest != null)
                        notifyPrimaryInterestChanged(interest);
                    else {
                        logger.log(Logger.WARNING,
                            br.W_MBUS_BAD_PRIMARY_INT, intid);
                    }
                }
                break;
            }
        }
        catch (IOException e) { 
            logger.logStack(Logger.DEBUG,"Exception processing packet ", e);
        }
    }

    /**
     * Broadcast the client closed notification to all the
     */
    private void sendClientClosedUpdate(ConnectionUID conid) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "MessageBus: Client closed notification. Client = {0}",
                conid);
        }

        try {
            dos.writeInt(version);
            dos.writeLong(conid.longValue());
            dos.flush();
            bos.flush();
        }
        catch (Exception e) {
            return;
        }

        byte[] buf = bos.toByteArray();

        try {
            c.broadcast( ClusterGlobals.MB_CLIENT_CLOSED, buf);
        }
        catch (IOException e) {}
    }

    /**
     * Receive and process  ClusterGlobals.MB_CLIENT_CLOSED packet.
     */
    private void receiveClientClosedUpdate(BrokerAddress sender,
        byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        ConnectionUID conid;

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_CLIENT_CLOSED));
                return;
            }
            conid = new ConnectionUID(dis.readLong());
        }
        catch (Exception e) {
            return;
        }

        if (DEBUG) {
            logger.log(Logger.DEBUG,
    "MessageBus: Received Client closed notification from {0}. Client = {1}",
                sender, conid);
        }

        cbDispatcher.clientDown(conid);
    }

    /**
     * This method should be called when a client connection gets
     * closed. Following things need to happen when a client goes
     * away -
     * <OL>
     * <LI> All the resource locks owned by the client should be
     * released. </LI>
     * <LI> All the interests created by the client should be either
     * removed or detached in case of durables. </LI>
     * <LI> A notification should be sent to all the brokers so that
     * they can clean up any temporary destinations owned by the
     * client. </LI>
     * </OL>
     * @param conid The ConnectionUID representing the client.
     * @param notify If <code> true, </code> all the brokers in
     * the cluster (including 'this' broker) are notified.
     */
    public void clientClosed(ConnectionUID conid, boolean notify) {
        // First unlock all the resources.
        ArrayList tmpList = new ArrayList();
        synchronized(resTable) {
            Collection entries = resTable.keySet();
            Iterator itr = entries.iterator();
            while (itr.hasNext()) {
                String resId = (String) itr.next();
                Resource res = (Resource) resTable.get(resId);

                if (conid.equals(res.getOwner()))
                    tmpList.add(resId);
            }
        }
        for (int i = 0; i < tmpList.size(); i++) {
            String resId = (String) tmpList.get(i);
            unlockResource(resId);
        }

        // Tell interest manager to cleanup all the interests.
        if (notify) {
            sendClientClosedUpdate(conid); // Notify other brokers.
            // Notify self..
            cbDispatcher.clientDown(conid);
        }
    }

    /**
     * Obtain a cluster-wide "shared" lock on a resource.
     * Unlike the normal "exclusive" locks, the shared locks allow
     * more than one clients to access the same resource. This method
     * ensures that the resource cannot be locked as shared and
     * exclusive at the same time!
     *
     * @param resID Resource name. The caller must ensure that
     * there are no name space conflicts between different
     * types of resources. This can be achieved by simply using
     * resource names like -"durable:foo", "queue:foo",
     * "clientid:foo"...
     * @param owner The ConnectionUID object representing the
     * client that will own this resource if this method returns
     * successfully. When this client goes away, the resource
     * will be unlocked automatically.
     * @return  ProtocolGlobals.G_LOCK_SUCCESS if the resource was
     *          locked successfully.
     *          ProtocolGlobals.G_LOCK_FAILURE if the resource could
     *          not be locked.
     */
    public int lockSharedResource(String resId, ConnectionUID owner) {
        throw new UnsupportedOperationException(
            "Shared locks are not supported.");
    }

    /**
     * Obtain a cluster-wide lock on a resource. This method is
     * used to ensure mutual exclusion for durable subscriptions,
     * queue receivers, client IDs etc.
     *
     * @param resID Resource name. The caller must ensure that
     * there are no name space conflicts between different
     * types of resources. This can be achieved by simply using
     * resource names like -"durable:foo", "queue:foo",
     * "clientid:foo"...
     * @param owner The ConnectionUID object representing the
     * client that will own this resource if this method returns
     * successfully. When this client goes away, the resource
     * will be unlocked automatically.
     * @return  ClusterGlobals.MB_LOCK_SUCCESS if the resource was locked successfully.
     *          ClusterGlobals.MB_LOCK_FAILURE if the resource could not be locked.
     */
    public int lockResource(String resId, ConnectionUID owner) {
        return lockResource(resId, 0, owner);
    }

    /**
     * Obtain a cluster-wide lock on a resource. This method is
     * used to ensure mutual exclusion for durable subscriptions,
     * queue receivers, client IDs etc.
     *
     * @param resID Resource name. The caller must ensure that
     * there are no name space conflicts between different
     * types of resources. This can be achieved by simply using
     * resource names like -"durable:foo", "queue:foo",
     * "clientid:foo"...
     *
     * @param timestamp The creation time for the resource.
     * In case of a lock contention the older resource automatically
     * wins.
     *
     * @param owner The ConnectionUID object representing the
     * client that will own this resource if this method returns
     * successfully. When this client goes away, the resource
     * will be unlocked automatically.
     *
     * @return  ClusterGlobals.MB_LOCK_SUCCESS if the resource was locked successfully.
     *          ClusterGlobals.MB_LOCK_FAILURE if the resource could not be locked.
     */
    public int lockResource(String resId, long timestamp,
        ConnectionUID owner) {
        int b = 1;

        int i = 0;
        while (true) {
            int ret = tryLockResource(resId, timestamp, owner);
            if (ret ==  ClusterGlobals.MB_LOCK_SUCCESS || ret ==  ClusterGlobals.MB_LOCK_FAILURE)
                return ret;

            i++;
            if (i ==  ClusterGlobals.MB_LOCK_MAX_ATTEMPTS)
                break;

            // ret ==  ClusterGlobals.MB_LOCK_BACKOFF - Try to lock again after
            // binary exponential backoff.
            b = b * 2;
            int sleep = r.nextInt(b);
            try {
                Thread.sleep(sleep * 1000L);
            }
            catch (InterruptedException e) {
                // does nothing
            }
        }

        logger.log(Logger.WARNING, br.W_MBUS_LOCK_ABORTED, resId);

        return  ClusterGlobals.MB_LOCK_FAILURE;
    }

    /**
     * Attempts to lock a resource using election protocol.
     */
    private int tryLockResource(String resId, long timestamp,
        ConnectionUID owner) {
        Resource res;

        synchronized(resTable) {
            res = (Resource) resTable.get(resId);
            if (res != null)
                return  ClusterGlobals.MB_LOCK_FAILURE;

            res = new Resource(resId);
            res.setOwner(owner);
            if (timestamp != 0)
                res.setTimestamp(timestamp);
            resTable.put(resId, res);
        }

        res.setLockState( ClusterGlobals.MB_RESOURCE_LOCKING);

        while (true) {
            if (DEBUG) {
                logger.log(Logger.DEBUG,
                    "MessageBus: Trying to lock resource {0}", resId);
            }

            sendLockRequest(res);
            int status = res.waitForStatusChange(lockTimeout);

            if (DEBUG) {
                logger.log(Logger.DEBUG,
                    "MessageBus: Lock attempt status = {0} for resource {1}",
                    Integer.toString(status),
                    resId);
            }

            switch (status) {
            case  ClusterGlobals.MB_LOCK_SUCCESS:
                res.setLockState( ClusterGlobals.MB_RESOURCE_LOCKED);
                return status;

            case  ClusterGlobals.MB_LOCK_BACKOFF:
            case  ClusterGlobals.MB_LOCK_FAILURE:
                synchronized (resTable) {
                    resTable.remove(resId);
                }
                return status;

            case  ClusterGlobals.MB_LOCK_TIMEOUT:
                logger.log(Logger.WARNING, br.W_MBUS_LOCK_TIMEOUT,
                    res.getResId(),
                    res.showRecipients());
                synchronized (resTable) {
                    resTable.remove(resId);
                }
                return  ClusterGlobals.MB_LOCK_FAILURE;

            case  ClusterGlobals.MB_LOCK_TRY_AGAIN:
                if (DEBUG_CLUSTER_ALL || DEBUG_CLUSTER_LOCK) {
                    logger.log(Logger.DEBUG,
    "Active brokerlist changed. Restarting Lock election for {0}." +
    res.getResId());
                }
                break;
            }
        }
    }

    /**
     * Unlocks a resource.
     */
    public void unlockResource(String resId) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "MessageBus: Unlocking resource = {0}",
                resId);
        }
        synchronized (resTable) {
            Resource res = (Resource) resTable.remove(resId);
            if (res != null)
                res.impliedFailure();
        }
    }

    /**
     * Constructs and sends (broadcast) an  ClusterGlobals.MB_LOCK_REQUEST packet.
     */
    private void sendLockRequest(Resource res) {
        long xid = System.currentTimeMillis();

        BrokerAddress[] recipients = null;

        synchronized (brokerList) {
            recipients = new BrokerAddress[brokerList.size()];
            int n = 0;

            Collection values = brokerList.values();
            Iterator itr = values.iterator();
            while (itr.hasNext()) {
                BrokerInfo binfo = (BrokerInfo) itr.next();
                recipients[n++] = binfo.getBrokerAddr();
            }
        }

        if (DEBUG_CLUSTER_ALL || DEBUG_CLUSTER_LOCK) {
            String debugstr = "";
            for (int i = 0; i < recipients.length; i++) {
                debugstr = debugstr + "\n\t" + recipients[i];
            }
            logger.log(Logger.DEBUG,
                "Sending resource lock request for : " + res.getResId() +
                "\nExpecting responses from : " + debugstr);
        }

        res.prepareLockRequest(recipients, xid);

        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeUTF(res.getResId());
            dos.writeLong(res.getTimestamp());
            dos.writeLong(xid);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        byte[] buf = bos.toByteArray();

        try {
            c.broadcast( ClusterGlobals.MB_LOCK_REQUEST, buf);
        }
        catch (Exception e) {}
    }

    /**
     * Process an  ClusterGlobals.MB_LOCK_REQUEST packet.
     */
    private void receiveLockRequest(BrokerAddress sender, byte[] pkt) {
        String resId;
        long timestamp;
        long xid;

        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_LOCK_REQUEST));
                return;
            }

            resId = dis.readUTF();
            timestamp = dis.readLong();
            xid = dis.readLong();
        }
        catch (Exception e) {
            return;
        }

        int response;
        Resource res = null;
        
        synchronized (resTable) {
            res = (Resource) resTable.get(resId);
        }

        if (res == null)
            response =  ClusterGlobals.MB_LOCK_SUCCESS;
        else if (res.getLockState() ==  ClusterGlobals.MB_RESOURCE_LOCKED)
            response =  ClusterGlobals.MB_LOCK_FAILURE;
        else {
            if (timestamp < res.getTimestamp()) {
                res.impliedFailure();
                response =  ClusterGlobals.MB_LOCK_SUCCESS;
            }
            else if (timestamp > res.getTimestamp())
                response =  ClusterGlobals.MB_LOCK_FAILURE;
            else
                response =  ClusterGlobals.MB_LOCK_BACKOFF;
        }

        sendLockResponse(sender, resId, xid, response);
    }

    /**
     * Construct and send a  ClusterGlobals.MB_LOCK_RESPONSE packet.
     */
    private void sendLockResponse(BrokerAddress to, String resId,
        long xid, int response) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeUTF(resId);
            dos.writeLong(xid);
            dos.writeInt(response);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        byte[] buf = bos.toByteArray();
        try {
            c.unicast(to,  ClusterGlobals.MB_LOCK_RESPONSE, buf);
        }
        catch (Exception e) {}
    }

    /**
     * Process an  ClusterGlobals.MB_LOCK_RESPONSE packet.
     */
    private void receiveLockResponse(BrokerAddress sender, byte[] pkt) {
        String resId;
        long xid;
        int response;

        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_LOCK_RESPONSE));
                return;
            }

            resId = dis.readUTF();
            xid = dis.readLong();
            response = dis.readInt();
        }
        catch (Exception e) {
            return;
        }

        Resource res = null;
        synchronized (resTable) {
            res = (Resource) resTable.get(resId);
        }

        if (res == null)
            return;

        if (DEBUG_CLUSTER_ALL || DEBUG_CLUSTER_LOCK) {
            if (res.getXid() == xid)
                logger.log(Logger.DEBUG,
                    "Received Lock Response." +
                    "\n\tSender = " + sender +
                    "\n\tResource = " + res.getResId() +
                    "\n\tResponse = " + ClusterGlobals.lockResponseStrings[response]);
        }
        res.consumeResponse(xid, sender, response);
    }

    public int recordAddDurableEvent(Subscription intr)
        throws BrokerException {

        Set s = new HashSet();
        s.add(intr);
        byte[] buf = prepareInterestUpdate(s,  ClusterGlobals.MB_NEW_INTEREST);

        return recordConfigChangeEvent( ClusterGlobals.MB_INTEREST_UPDATE, buf);
    }

    public int recordRemDurableEvent(Subscription intr)
        throws BrokerException {
        Set s = new HashSet();
        s.add(intr);
        byte[] buf = prepareInterestUpdate(s,  ClusterGlobals.MB_REM_DURABLE_INTEREST);

        return recordConfigChangeEvent( ClusterGlobals.MB_INTEREST_UPDATE, buf);
    }

    /**
     * Log a cluster configuration event.
     */
    private int recordConfigChangeEvent(int eventDestId, byte[] eventData)
        throws BrokerException {
        if (c.getConfigServer() == null)
            return  ClusterGlobals.MB_EVENT_LOG_FAILURE;

        synchronized (eventLogLockObject) {
            eventLogStatus =  ClusterGlobals.MB_EVENT_LOG_WAITING;
            eventLogXid = System.currentTimeMillis();

            sendConfigChangeEvent(eventLogXid, eventDestId, eventData);

            // and wait for response.
            while (eventLogStatus ==  ClusterGlobals.MB_EVENT_LOG_WAITING) {
                try {
                    eventLogLockObject.wait();
                }
                catch (Exception e) {}
            }
            if (eventLogStatus ==  ClusterGlobals.MB_EVENT_LOG_SUCCESS) {
                storeDirtyFlag = true;
            }

            return eventLogStatus;
        }
    }

    /**
     * Send configuration change event to the config server.
     */
    private void sendConfigChangeEvent(long eventLogXid,
        int eventDestId, byte[] eventData) throws BrokerException {
        BrokerAddress configServer = c.getConfigServer();

        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeLong(eventLogXid);
            dos.writeInt(eventData.length + 4); // For eventDestId
            dos.writeInt(eventDestId);
            dos.write(eventData, 0, eventData.length);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        byte[] buf = bos.toByteArray();
        try {
            // Send the event to the config server.
            c.unicast(configServer,  ClusterGlobals.MB_CONFIG_CHANGE_EVENT, buf);
        }
        catch (Exception e) {
            throw new BrokerException(
                br.getKString(br.X_CFG_SERVER_UNREACHABLE),
                br.X_CFG_SERVER_UNREACHABLE, (Throwable) null,
                Status.FORBIDDEN);
        }
    }

    private void receiveConfigChangeEvent(BrokerAddress sender, byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        long xid = -1;
        byte[] eventData;

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_CONFIG_CHANGE_EVENT));
                return;
            }

            xid = dis.readLong();
            int len = dis.readInt();
            eventData = new byte[len];
            dis.readFully(eventData, 0, len);
        }
        catch (Exception e1) {
            // TBD: Log error
            return;
        }

        try {
            store.storeConfigChangeRecord(System.currentTimeMillis(),
                eventData, false);
            sendConfigChangeEventAck(sender, xid,  ClusterGlobals.MB_EVENT_LOG_SUCCESS);
        }
        catch (Exception e2) {
            sendConfigChangeEventAck(sender, xid,  ClusterGlobals.MB_EVENT_LOG_FAILURE);
            return;
        }
    }

    private void sendConfigChangeEventAck(BrokerAddress to, long xid,
        int status) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeLong(xid);
            dos.writeInt(status);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        byte[] buf = bos.toByteArray();
        try {
            c.unicast(to,  ClusterGlobals.MB_CONFIG_CHANGE_EVENT_ACK, buf);
        }
        catch (Exception e) {}
    }

    private void receiveConfigChangeEventAck(BrokerAddress sender,
        byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_CONFIG_CHANGE_EVENT_ACK));
                return;
            }

            long xid = dis.readLong();
            int status = dis.readInt();

            // Process the ack
            synchronized (eventLogLockObject) {
                if (xid != eventLogXid) {
                    // TBD: Log error
                    return;
                }
                eventLogStatus = status;
                eventLogLockObject.notify();
            }
        }
        catch (Exception e) {
            return;
        }
    }

    private void sendConfigChangesRequest(BrokerAddress to, long timestamp) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeLong(timestamp);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        byte[] buf = bos.toByteArray();
        synchronized (cfgSrvWaitObject) {
            try {
                c.unicast(to,  ClusterGlobals.MB_GET_CONFIG_CHANGES_REQUEST, buf);
                cfgSrvRequestCount++;
                cfgSrvRequestErr = false;
            }
            catch (Exception e) {
                cfgSrvRequestCount = 0;
                cfgSrvRequestErr = true;
                cfgSrvWaitObject.notifyAll();
            }
        }
    }

    private void receiveConfigChangesRequest(BrokerAddress sender,
        byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_GET_CONFIG_CHANGES_REQUEST));
                return;
            }

            long timestamp = dis.readLong();
            long now = System.currentTimeMillis();

            ArrayList records = store.getConfigChangeRecordsSince(timestamp);
            sendConfigChangesResponse(sender, now, records);
        }
        catch (Exception e) {
            // TBD: Log error
            return;
        }
    }

    private void sendConfigChangesResponse(BrokerAddress to,
        long currentTime, ArrayList records) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeLong(currentTime);
            dos.writeInt(records.size());

            for (int i = 0; i < records.size(); i++) {
                byte[] rec = (byte[]) records.get(i);
                dos.writeInt(rec.length);
                dos.write(rec, 0, rec.length);
            }

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        byte[] buf = bos.toByteArray();
        try {
            c.unicast(to,  ClusterGlobals.MB_GET_CONFIG_CHANGES_RESPONSE, buf);
        }
        catch (Exception e) {
            logger.log(Logger.DEBUG,"Error getting configChanges ", e);
        }
    }

    private void receiveConfigChangesResponse(BrokerAddress sender,
        byte[] pkt) {
        //
        // First, check if we are talking to the same config server
        // as last time. Usually this is true, except in following
        // scenarios -
        //
        // 1. The administrator has changed the config server for
        // this cluster, typically using backup and restore. In this
        // case the new config server will ensure that all the brokers
        // will get the  ClusterGlobals.MB_RESET_PERSISTENCE command, so there is
        // no problem.
        //
        // 2. When a broker is moved from one cluster to another.
        // In this case, we cannot allow the broker to join in
        // with the old destinations / durable state information...
        //

        boolean resetRequired = false;
        BrokerAddress lastConfigServer = getLastConfigServer();
        if (lastConfigServer != null &&
            ! lastConfigServer.equals(sender)) {
            resetRequired = true;
        }

        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_GET_CONFIG_CHANGES_RESPONSE));
                return;
            }

            long refreshTime = dis.readLong();
            long n = dis.readInt();

            if (resetRequired && n == 0) {
                logger.log(Logger.ERROR, br.E_MBUS_CLUSTER_JOIN_ERROR);

                // Since all the client connections were blocked,
                // it should be safe to terminate the VM here.
                Broker.getBroker().exit(1,
                    br.getString( br.E_MBUS_CLUSTER_JOIN_ERROR),
                    BrokerEvent.Type.FATAL_ERROR);
            }

            boolean resetFlag = false;
            ArrayList l = null;

            for (int i = 0; i < n; i++) {
                int len = dis.readInt();

                if (i == 0) {
                    // Check if the first record is
                    //  ClusterGlobals.MB_RESET_PERSISTENCE ...
                    dis.mark(len);

                    byte[] rec = new byte[len];
                    dis.readFully(rec, 0, len);

                    ChangeRecord cr = ChangeRecord.makeChangeRecord(rec);
                    if (cr.eventDestId ==  ClusterGlobals.MB_RESET_PERSISTENCE) {
                        resetFlag = true;
                        l = new ArrayList();
                    }

                    if (resetRequired && resetFlag == false) {
                        logger.log(Logger.ERROR,
                            br.E_MBUS_CLUSTER_JOIN_ERROR);

                        // Since all the client connections were blocked,
                        // it should be safe to terminate the VM here.
                        Broker.getBroker().exit(1,
                            br.getString( br.E_MBUS_CLUSTER_JOIN_ERROR),
                            BrokerEvent.Type.FATAL_ERROR);
                    }

                    dis.reset(); // jump back in the input stream.
                }

                if (resetFlag) {
                    byte[] rec = new byte[len];
                    dis.readFully(rec, 0, len);
                    l.add(rec);
                }
                else {
                    int eventDestId = dis.readInt();
                    byte[] eventData = new byte[len - 4]; // Skip eventDestId
                    dis.readFully(eventData, 0, len - 4);

                    // Treat the event as if it was just received
                    // via broadcast!
                    receiveBroadcast(sender, eventDestId, eventData, true);
                }
            }

            if (resetFlag)
                applyPersistentStateChanges(sender, l);

            if (configSyncComplete == false) {
                configSyncComplete = true;
                cbDispatcher.configSyncComplete(); // Do this only once.
                // So far, this broker was not capable of accepting
                // any interest updates (because it's destination list
                // was potentially stale) and hence receiveInterestUpdate
                // was just throwing them away. However note that such
                // interest updates can only originate from the master
                // broker itself, because before this point we were not
                // even talking to any other brokers...
                //
                // So - This now it the time to ask the master broker
                // to send it's full interest udpate again!!!
                //
                sendRequestInterestUpdate(sender);

                logger.log(Logger.INFO,
                    br.I_MBUS_SYNC_COMPLETE);
            }

            synchronized (cfgSrvWaitObject) {
                cfgSrvRequestCount--;
                if (cfgSrvRequestCount == 0) {
                    cfgSrvWaitObject.notifyAll();
                }
            }

            storeLastRefreshTime(refreshTime -
                 ClusterGlobals.MB_EVENT_LOG_CLOCK_SKEW_TOLERANCE);
            storeLastConfigServer(sender);
        }
        catch (Exception e) {
            // TBD: Log error
            return;
        }
    }

    /**
     * Wait for config server's response to the 
     *  ClusterGlobals.MB_GET_CONFIG_CHANGES_REQUEST.
     *
     * @return true if response received and consumed successfully,
     * false if config server failure.
     */
    private boolean waitConfigChangesResponse() {
        synchronized (cfgSrvWaitObject) {
            while (cfgSrvRequestCount > 0) {
                try {
                    cfgSrvWaitObject.wait();
                }
                catch (Exception e) {}
            }

            return (! cfgSrvRequestErr);
        }
    }

    private void applyPersistentStateChanges(BrokerAddress sender,
        ArrayList list) throws Exception {
        HashMap oldInts = new HashMap();
        HashMap oldDests = new HashMap();

        Set ints = Subscription.getAllSubscriptions(null);
        Iterator itr = ints.iterator();
        while (itr.hasNext()) {
            Subscription sub = (Subscription)itr.next();
            com.sun.messaging.jmq.jmsserver.core.ConsumerUID intid = sub.getConsumerUID();
            String key = sub.getDurableName() + ":" + sub.getClientID();
            oldInts.put(key, intid);
        }
        ints = null;

        itr = Destination.getAllDestinations(); 
        while (itr.hasNext()) {
            Destination d = (Destination) itr.next();
            if (d.isAutoCreated() || d.isInternal())
                continue;

            oldDests.put(d.getDestinationUID(), d);
        }

        for (int i = 0; i < list.size(); i++) {
            byte[] rec = (byte[]) list.get(i);
            ChangeRecord cr = ChangeRecord.makeChangeRecord(rec);

            if (cr.eventDestId ==  ClusterGlobals.MB_RESET_PERSISTENCE)
                continue;

            if (cr.eventDestId ==  ClusterGlobals.MB_INTEREST_UPDATE) {
                receiveBroadcast(sender, cr.eventDestId, cr.rec, true);
                InterestUpdateChangeRecord icr =
                    (InterestUpdateChangeRecord) cr;

                String key = icr.dname + ":" + icr.cid;
                oldInts.remove(key);
            }
            else if (cr.eventDestId ==  ClusterGlobals.MB_DESTINATION_UPDATE) {
                DestinationUpdateChangeRecord dcr =
                    (DestinationUpdateChangeRecord) cr;

                Object key = DestinationUID.getUID(dcr.name,
                           DestType.isQueue(dcr.type));

                Destination old = (Destination) oldDests.get(key);

                if (dcr.isAddOp()) {
                    if (old != null && old.getType() != dcr.type) {
                        DestinationUID duid = DestinationUID.getUID(
                               old.getDestinationName(),
                               DestType.isQueue(old.getType()));
                        cbDispatcher.notifyDestroyDestination(duid);
                    }
                }

                receiveBroadcast(sender, cr.eventDestId, cr.rec, true);
                oldDests.remove(key);
            }
        }

        for (itr = oldInts.values().iterator(); itr.hasNext();) {
            ConsumerUID intid = (ConsumerUID) itr.next();
            Consumer c = Consumer.getConsumer(intid);
            if (c != null) {
                cbDispatcher.interestRemoved(c, null, false);
            }
        }

        for (itr = oldDests.keySet().iterator(); itr.hasNext();) {
            DestinationUID d = (DestinationUID) itr.next();
            cbDispatcher.notifyDestroyDestination(d);
        }
    }

    public static byte[] upgradeConfigChangeRecord(byte[] olddata)
        throws IOException {

        ChangeRecord cr = ChangeRecord.makeChangeRecord(olddata);

        if (cr.eventDestId == ClusterGlobals.MB_RESET_PERSISTENCE) {
            return RaptorProtocol.prepareResetPersistenceRecord();
        }
        else if (cr.eventDestId == ClusterGlobals.MB_INTEREST_UPDATE) {
            ByteArrayInputStream bis = new ByteArrayInputStream(cr.rec);
            DataInputStream dis = new DataInputStream(bis);

            int pktVersion = dis.readInt();
            int type = dis.readInt();
            int count = dis.readInt();

            if (type == ClusterGlobals.MB_NEW_INTEREST) {
                dis.readBoolean(); // is durable?
                dis.readUTF(); // durable name
                dis.readUTF(); // client id
                Consumer c = readConsumer(dis, null);
                Subscription sub = c.getSubscription();
                return RaptorProtocol.generateAddDurableRecord(sub);
            }

            if (type == ClusterGlobals.MB_REM_DURABLE_INTEREST) {
                String dname = dis.readUTF();
                String cname = dis.readUTF();
                Subscription sub = Subscription.findDurableSubscription(
                    cname, dname);
                return RaptorProtocol.generateRemDurableRecord(sub);
            }
        }
        else if (cr.eventDestId == ClusterGlobals.MB_DESTINATION_UPDATE) {
            ByteArrayInputStream bis = new ByteArrayInputStream(cr.rec);
            DataInputStream dis = new DataInputStream(bis);

            int pktVersion = dis.readInt();
            int operation = dis.readInt();
            String name = dis.readUTF();
            int type = dis.readInt();

            if (operation == ClusterGlobals.MB_NEW_DESTINATION) {
                DestinationUID duid = null;
                try {
                    duid = DestinationUID.getUID(name, type);
                } catch (BrokerException ex) {
                    // should never happen, destination should already be verified
                    return olddata;
                }
                Destination d = Destination.getDestination(duid);
                return RaptorProtocol.generateAddDestinationRecord(d);
            }

            if (operation == ClusterGlobals.MB_REM_DESTINATION) {
                DestinationUID duid = null;
                try {
                    duid = DestinationUID.getUID(name, type);
                } catch (BrokerException ex) {
                    // should never happen, destination should already be verified
                    return olddata;
                }
                Destination d = Destination.getDestination(duid);
                return RaptorProtocol.generateRemDestinationRecord(d);
            }
        }

        Globals.getLogger().log(Logger.INFO,
            "Internal error : upgradeConfigChangeRecord conversion failed.");

        return olddata;
    }

    private byte[] prepareDestinationUpdate(String name, int type,
        int operation) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);
            dos.writeInt(operation);

            dos.writeUTF(name);
            dos.writeInt(type);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        return bos.toByteArray();
    }

    public void recordUpdateDestination(Destination d)
        throws BrokerException {
        if (DEBUG) {
            logger.log(logger.DEBUG,
                "FalconProtocol.recordUpdateDestination");
        }

        recordAddDestinationEvent(d.getDestinationName(),
             d.getType());
    }

    public void recordRemoveDestination(Destination d)
        throws BrokerException {
        if (DEBUG) {
            logger.log(logger.DEBUG,
                "FalconProtocol.recordRemoveDestination");
        }

        recordRemDestinationEvent(d.getDestinationName(),
             d.getType());
    }



    public void sendNewDestination(Destination d) 
        throws BrokerException 
    {
        logger.log(Logger.DEBUG,"Sending New Destination " + d);
        sendDestinationUpdate(d.getDestinationName(),
             d.getType(),
             ClusterGlobals.MB_NEW_DESTINATION);
    }

    public void sendUpdateDestination(Destination d)
        throws BrokerException 
    {
        // does nothing, unsupported
    }

    public void sendRemovedDestination(Destination d)
        throws BrokerException 
    {
        logger.log(Logger.DEBUG,"Sending New Destination " + d);
        sendDestinationUpdate(d.getDestinationName(),
             d.getType(),
             ClusterGlobals.MB_REM_DESTINATION);
    }

    public void recordCreateSubscription(Subscription sub)
        throws BrokerException {
        recordAddDurableEvent(sub);
    }

    public void recordUnsubscribe(Subscription sub)
        throws BrokerException {
        recordRemDurableEvent(sub);
    }

    public void sendNewSubscription(Subscription sub, Consumer cons,
        boolean active) throws BrokerException {
        sendNewConsumer(sub, active);
    }

    public void sendNewConsumer(Consumer c, boolean active)
        throws BrokerException 
    {
        sendInterestUpdate(c, ClusterGlobals.MB_NEW_INTEREST);

        if (active) {
            sendInterestUpdate(c, ClusterGlobals.MB_NEW_PRIMARY_INTEREST);
        }
    }

    public void sendRemovedConsumer(Consumer c, Map pendingMsgs, boolean cleanup) 
        throws BrokerException 
    {
        if (c instanceof Subscription) {
            sendInterestUpdate(c, ClusterGlobals.MB_REM_DURABLE_INTEREST);
        } else if (c.getSubscription() != null) { // detatching
            sendInterestUpdate(c, ClusterGlobals.MB_DURABLE_DETACH);
        } else {
            sendInterestUpdate(c, ClusterGlobals.MB_REM_INTEREST);
        }
    }

    protected int recordAddDestinationEvent(String name, int type)
        throws BrokerException {
        byte[] buf = prepareDestinationUpdate(name, type,  ClusterGlobals.MB_NEW_DESTINATION);
        return recordConfigChangeEvent( ClusterGlobals.MB_DESTINATION_UPDATE, buf);
    }

    protected int recordRemDestinationEvent(String name, int type)
        throws BrokerException {
        byte[] buf = prepareDestinationUpdate(name, type,  ClusterGlobals.MB_REM_DESTINATION);
        return recordConfigChangeEvent( ClusterGlobals.MB_DESTINATION_UPDATE, buf);
    }

    protected void sendDestinationUpdate(String name, int type, int operation) {
        if (DEBUG) {
            logger.log(Logger.DEBUG,
    "MessageBus: Broadcasting destination update. Name = {0}, Type = {1}",
                name,
                DestType.toString(type));
        }

        byte[] buf = prepareDestinationUpdate(name, type, operation);
        try {
            c.broadcast( ClusterGlobals.MB_DESTINATION_UPDATE, buf);
        }
        catch (IOException e) {}
    }

    private void receiveDestinationUpdate(BrokerAddress sender,
        byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING,  br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_DESTINATION_UPDATE));

                return;
            }

            int operation = dis.readInt();
            String name = dis.readUTF();
            int type = dis.readInt();

            if (DEBUG) {
                logger.log(Logger.DEBUG,
    "MessageBus: Receiving destination update from {0}. Name = {1}",
                    sender, name);
            }

            switch (operation) {
            case  ClusterGlobals.MB_NEW_DESTINATION:
                Destination d = Destination.createDestination(name,type, !DestType.isTemporary(type),
                       false, selfAddress);
                cbDispatcher.notifyCreateDestination(d);
                break;

            case  ClusterGlobals.MB_REM_DESTINATION:
                DestinationUID duid = DestinationUID.getUID(name,DestType.isQueue(type));
                cbDispatcher.notifyDestroyDestination(duid);
                break;

            default:
                logger.log(Logger.ERROR, br.E_MBUS_DEST_UPDATE_ERROR);
                break;
            }
        }
        catch (Exception e) {}
    }

    private byte[] prepareResetPersistenceRecord() {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt( ClusterGlobals.MB_RESET_PERSISTENCE);
            dos.writeInt(version);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {}

        return bos.toByteArray();
    }

    private void receiveResetPersistence(BrokerAddress sender,
        byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING,  br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_RESET_PERSISTENCE));
                return;
            }
            if (DEBUG) {
                logger.log(Logger.DEBUG,
    "MessageBus: Receiving reset persistence command from {0}",
                    sender);
            }
        }
        catch (Exception e) {}
    }

    private void sendRestartCluster() {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {
            return;
        }

        byte[] buf = bos.toByteArray();

        try {
            c.broadcast( ClusterGlobals.MB_RESTART_CLUSTER, buf);
        }
        catch (IOException e) {}
    }

    private void receiveRestartCluster(BrokerAddress sender, byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_RESTART_CLUSTER));
                return;
            }
        }
        catch (Exception e) {
            return;
        }

        if (DEBUG) {
            logger.log(Logger.DEBUG,
    "MessageBus: Received reset cluster notification from {0}.",
                sender);
        }

        logger.log(Logger.INFO, br.I_MBUS_RELOAD_CLS);
        c.reloadCluster();
    }

    private void sendRequestInterestUpdate(BrokerAddress broker) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);

        try {
            dos.writeInt(version);

            dos.flush();
            bos.flush();
        }
        catch (Exception e) {
            return;
        }

        byte[] buf = bos.toByteArray();

        try {
            c.unicast(broker,  ClusterGlobals.MB_REQUEST_INTEREST_UPDATE, buf);
        }
        catch (IOException e) {}
    }

    private void receiveRequestInterestUpdate(BrokerAddress sender,
        byte[] pkt) {
        ByteArrayInputStream bis = new ByteArrayInputStream(pkt);
        DataInputStream dis = new DataInputStream(bis);

        try {
            int pktVersion = dis.readInt();
            if (pktVersion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_BAD_VERSION,
                    Integer.toString(pktVersion),
                    Integer.toString( ClusterGlobals.MB_REQUEST_INTEREST_UPDATE));
                return;
            }
        }
        catch (Exception e) {
            return;
        }

        if (DEBUG) {
            logger.log(Logger.DEBUG,
    "MessageBus: Received request for a full interest update from {0}.",
                sender);
        }

        forwardLocalInterests(sender);
    }


    /**
     * Add a new broker to the list of known brokers in this cluster.
     */
    public int addBrokerInfo(BrokerInfo brokerInfo) {
        Object old;
        synchronized (brokerList) {
            old = brokerList.put(brokerInfo.getBrokerAddr(),
                brokerInfo);
        }

        if (configSyncComplete == false) {
            try {
                BrokerAddress configServer = c.getConfigServer();
                if (configServer != null &&
                    configServer.equals(brokerInfo.getBrokerAddr())) {
                    // This is the config server! Initiate config sync.

                    long timestamp = -1;
                    timestamp = getLastRefreshTime();

                    sendConfigChangesRequest(configServer, timestamp);

                    if (DEBUG_CLUSTER_ALL || DEBUG_CLUSTER_CONN) {
                        logger.log(Logger.INFO,
                            br.I_MBUS_SYNC_INIT);
                    }
                }
                else {
                    // This is not the config server. Cannot
                    // accept connections from this broker at this
                    // stage...
                    return ADD_BROKER_INFO_RETRY;
                }
            }
            catch (Exception e) {}
        }

        if (old == null) {
            if (DEBUG) {
                logger.log(Logger.DEBUG,
                    "MessageBus: New Broker : {0}",
                    brokerInfo);
            }

            // If the new broker is not the config server AND
            // it has added some destinations / durables recently,
            // make sure that this broker knows about it.
            // Why? Consider the following sequence of events -
            //
            // 1. A syncs with the config server.
            // 2. B adds a destination and sends a broadcast.
            // 3. Five seconds later A connects with B...
            // At this point A must check with the config server again
            // to see if anything has changed in last five seconds...

            try {
                BrokerAddress configServer = c.getConfigServer();
                if (brokerInfo.getStoreDirtyFlag() &&
                    configServer != null &&
                    ! configServer.equals(brokerInfo.getBrokerAddr())) {

                    long timestamp = -1;
                    timestamp = getLastRefreshTime();
                    sendConfigChangesRequest(configServer, timestamp);

                    // The broker I/O thread blocks here...
                    if (waitConfigChangesResponse() == false)
                        return ADD_BROKER_INFO_BAN;
                }
            }
            catch (Exception e) {}

            forwardLocalInterests(brokerInfo.getBrokerAddr());
            restartElections(brokerInfo.getBrokerAddr());

            logger.log(Logger.INFO, br.I_MBUS_ADD_BROKER,
                brokerInfo.getBrokerAddr().toString());
        }

        updateActivelistProperty();

        // Falcon HA : Notify the HA watchdog.
        if (haWatchdog != null)
            haWatchdog.handleBrokerUp(brokerInfo.getBrokerAddr());

        return ADD_BROKER_INFO_OK;
    }

    private void forwardLocalInterests(BrokerAddress broker) {
        // BugID : 4451545
        // Advertize my (local) temporary destinations.
        // Note - temporary destinations are not recorded by
        // the master broker.

        Iterator itr = Destination.getTempDestinations(selfAddress);
        while (itr.hasNext()) {
            Destination d = (Destination) itr.next();
            sendDestinationUpdate(d.getDestinationName(), d.getType(),
                 ClusterGlobals.MB_NEW_DESTINATION);
        }

        // Advertize my local interests only to the new guy.
        Set localActiveInterests = new HashSet();
        Set primaryInterests = new HashSet();

        itr = Consumer.getAllConsumers();
        while (itr.hasNext()) {
            Consumer c = (Consumer)itr.next();
            if (! (c instanceof Subscription)) { // active consumer
                ConsumerUID uid = c.getConsumerUID();
                if (selfAddress == uid.getBrokerAddress()) { // local
                    localActiveInterests.add(c);
                }
                if (c.getIsActiveConsumer()) { // indicates active queue
                    DestinationUID duid = c.getDestinationUID();
                    Destination d = Destination.getDestination(duid);
                    if (d != null && d.getMaxActiveConsumers() == 1) {
                        // primary SQR or primary on failover
                        primaryInterests.add(c);
                    }
                }
                   
            }
        }

        if (!localActiveInterests.isEmpty()) {
            sendInterestUpdate(broker, localActiveInterests,  
                    ClusterGlobals.MB_NEW_INTEREST);
        }
        if (!primaryInterests.isEmpty()) {
            sendInterestUpdate(broker, primaryInterests,  
                    ClusterGlobals.MB_NEW_PRIMARY_INTEREST);
        }
    }

    private void restartElections(BrokerAddress broker) {
        // The new broker should participate in all the ongoing
        // lockResource() elections...
        synchronized(resTable) {
            Collection entries = resTable.keySet();
            Iterator itr = entries.iterator();
            while (itr.hasNext()) {
                String resId = (String) itr.next();
                Resource res = (Resource) resTable.get(resId);
                res.brokerAdded(broker);
            }
        }
    }

    /**
     * Remove a broker since it is no longer attached to this cluster.
     */
    public void removeBrokerInfo(BrokerAddress broker) {
        BrokerInfo brokerInfo;

        synchronized (brokerList) {
            brokerInfo = (BrokerInfo) brokerList.remove(broker);
        }
        if (DEBUG) {
            logger.log(Logger.DEBUG,
                "MessageBus: Broker down : {0}",
                brokerInfo);
        }

        /* If this was the configuration server, fail any pending
         * attempts to record configuration chage */
        try {
            if (c.getConfigServer().equals(broker)) {
                synchronized (eventLogLockObject) {
                    eventLogStatus =  ClusterGlobals.MB_EVENT_LOG_FAILURE;
                    eventLogLockObject.notify();
                }

                synchronized (cfgSrvWaitObject) {
                    if (cfgSrvRequestCount > 0) {
                        cfgSrvRequestCount = 0;
                        cfgSrvRequestErr = true;
                        cfgSrvWaitObject.notifyAll();
                    }
                }
            }
        }
        catch (Exception e) {} // Catches NullPointerException too.

        logger.log(Logger.INFO, br.I_MBUS_DEL_BROKER,
            broker.toString());

        cbDispatcher.brokerDown(broker);

        // Since the broker has gone down, don't wait for its
        // election responses.
        synchronized(resTable) {
            Collection entries = resTable.keySet();
            Iterator itr = entries.iterator();
            while (itr.hasNext()) {
                String resId = (String) itr.next();
                Resource res = (Resource) resTable.get(resId);
                res.brokerRemoved(broker);
            }
        }

        updateActivelistProperty();

        // Falcon HA : Notify the HA watchdog.
        if (haWatchdog != null)
            haWatchdog.handleBrokerDown(broker);
    }

    /**
     * Update the active broker list for admin query operations.
     * The "imq.cluster.brokerlist.active" property always reflects
     * the list currently connected brokers. Note - this is NOT
     * a normal "configuration" property. It is just a piece of
     * information provided by the broker for the administration.
     */
    public void updateActivelistProperty() {
        TreeSet ts = new TreeSet();
        ts.add(selfAddress.toString());

        synchronized (brokerList) {
            Collection values = brokerList.values();
            Iterator itr = values.iterator();
            while (itr.hasNext()) {
                BrokerInfo binfo = (BrokerInfo) itr.next();
                ts.add(binfo.getBrokerAddr().toString());
            }
        }

        String str = null;
        Iterator itr = ts.iterator();
        while (itr.hasNext()) {
            if (str == null)
                str = ((String) itr.next());
            else
                str = str + ", " + ((String) itr.next());
        }

        BrokerConfig config = Globals.getConfig();
        config.putOne(Globals.IMQ + ".cluster.brokerlist.active", str);
    }

    /**
     * This method is called by the InterestManager implementation
     * when a new interest is created.
     */
    public void notifyInterestCreated(Consumer intr) {
        cbDispatcher.interestCreated(intr);
    }

    /**
     * This method is called by the InterestManager implementation
     * when an interest is removed.
     */
    public void notifyInterestRemoved(Consumer intr) {
        cbDispatcher.interestRemoved(intr, null, false);
    }

    /**
     * This method is called by the InterestManager implementation
     * when an interest is removed.
     */
    public void notifyPrimaryInterestChanged(Consumer intr) {
        cbDispatcher.activeStateChanged(intr);
    }

    /**
     * Backup the config server data.
     */
    private void configServerBackup(String fileName) {
        BrokerAddress configServer = null;

        try {
            configServer = c.getConfigServer();
        }
        catch (Exception e) {
            logger.log(Logger.WARNING, br.W_MBUS_CANCEL_BACKUP1);
            return;
        }

        if (configServer == null ||
            ! configServer.equals(selfAddress)) {
                logger.log(Logger.WARNING, br.W_MBUS_CANCEL_BACKUP1);
            return;
        }

        try {
            // Make sure that the file does not exist.
            File f = new File(fileName);
            if (! f.createNewFile()) {
                logger.log(Logger.WARNING, br.W_MBUS_CANCEL_BACKUP2, fileName);
                return;
            }

            FileOutputStream fos = new FileOutputStream(f);
            DataOutputStream dos = new DataOutputStream(fos);

            Object lists[] = store.getAllConfigRecords();
            ArrayList records = (ArrayList) lists[1];

            ArrayList recordList = new ArrayList();
            HashMap recordMap = new HashMap();

            for (int i = 0; i < records.size(); i++) {
                byte[] rec = (byte []) records.get(i);
                ChangeRecord cr = ChangeRecord.makeChangeRecord(rec);

                recordList.add(cr);

                // Discard previous record with the same name.
                ChangeRecord prev = (ChangeRecord)
                    recordMap.get(cr.getUniqueKey());

                if (prev != null)
                    prev.discard = true;

                // Keep only the last add operation.
                if (cr.isAddOp() != true)
                    cr.discard = true;

                recordMap.put(cr.getUniqueKey(), cr);
            }

            dos.writeInt(version); // Version
            dos.writeUTF(ClusterGlobals.CFGSRV_BACKUP_PROPERTY); // Signature.

            // Write the RESET record here.
            byte[] rst = prepareResetPersistenceRecord();
            dos.writeInt(rst.length);
            dos.write(rst, 0, rst.length);

            for (int i = 0; i < recordList.size(); i++) {
                ChangeRecord cr = (ChangeRecord) recordList.get(i);

                if (! cr.discard) {
                    dos.writeInt(cr.rec.length + 4);
                    dos.writeInt(cr.eventDestId);
                    dos.write(cr.rec, 0, cr.rec.length);
                }
            }

            dos.writeInt(0);
        }
        catch (Exception e) {
            logger.logStack(Logger.WARNING,
                br.W_MBUS_BACKUP_ERROR, e);
        }
    }

    /**
     * Restore the config server database.
     */
    private void configServerRestore(String fileName) {
        try {
            // Make sure that the file does not exist.
            File f = new File(fileName);
            if (! f.exists()) {
                logger.log(Logger.WARNING, br.W_MBUS_CANCEL_RESTORE1,
                    fileName);
                return;
            }

            FileInputStream fis = new FileInputStream(f);
            DataInputStream dis = new DataInputStream(fis);

            int curversion = dis.readInt(); // Version
            String sig = dis.readUTF(); // Signature.

            if (! sig.equals(ClusterGlobals.CFGSRV_BACKUP_PROPERTY)) {
                logger.log(Logger.WARNING, br.W_MBUS_CANCEL_RESTORE2,
                    fileName);
                return;
            }

            if (curversion > version) {
                logger.log(Logger.WARNING, br.W_MBUS_CANCEL_RESTORE3);
                return;
            }

            store.clearAllConfigChangeRecords(false);

            while (true) {
                int recsize = dis.readInt();
                if (recsize == 0)
                    break;

                byte[] rec = new byte[recsize];
                dis.readFully(rec, 0, recsize);

                store.storeConfigChangeRecord(System.currentTimeMillis(),
                    rec, false);
            }

            dis.close();
            fis.close();
        }
        catch (Exception e) {
            logger.logStack(Logger.WARNING, br.W_MBUS_RESTORE_ERROR,
                e);
            return;
        }
    }

    private void storeLastConfigServer(BrokerAddress baddr)
        throws BrokerException {
        store.updateProperty(ClusterGlobals.STORE_PROPERTY_LASTCONFIGSERVER, baddr, false);
    }

    private BrokerAddress getLastConfigServer() {
        BrokerAddress baddr = null;

        try {
            baddr = (BrokerAddress)
                store.getProperty(ClusterGlobals.STORE_PROPERTY_LASTCONFIGSERVER);
        }
        catch (Exception e) {}

        return baddr;
    }

    private void storeLastRefreshTime(long timestamp)
        throws BrokerException {
        Long t = new Long(timestamp);
        store.updateProperty(ClusterGlobals.STORE_PROPERTY_LASTREFRESHTIME, t, false);
    }

    private long getLastRefreshTime() {
        Long t = null;

        try {
            t = (Long) store.getProperty(ClusterGlobals.STORE_PROPERTY_LASTREFRESHTIME);
        }
        catch (Exception e) {}

        if (t == null) {
            return -1;
        }

        return t.longValue();
    }

}

/**
 * Encapsulates a config server backup record.
 *
 * TBD : Note - there is some duplicate code here. The normal event
 * processing methods (receiveDestinationUpdate and
 * receiveInterestUpdate) also do the same thing - i.e. read and parse
 * the event data. They should be merged so that the event data gets
 * parsed by common methods.
 */
class ChangeRecord {
    protected byte[] rec;
    protected int eventDestId;
    protected boolean discard = false;

    protected static ChangeRecord makeChangeRecord(byte[] rec)
        throws IOException {
        ByteArrayInputStream bis = new ByteArrayInputStream(rec);
        DataInputStream dis = new DataInputStream(bis);

        int destid = dis.readInt();
        ChangeRecord cr = null;

        dis.mark(rec.length);

        if (destid == ClusterGlobals.MB_INTEREST_UPDATE)
            cr = new InterestUpdateChangeRecord(dis);
        else if (destid == ClusterGlobals.MB_DESTINATION_UPDATE)
            cr = new DestinationUpdateChangeRecord(dis);
        else if (destid == ClusterGlobals.MB_RESET_PERSISTENCE)
            cr = new ChangeRecord();

        dis.reset(); // jump back in the input stream.

        cr.rec = new byte[rec.length - 4];
        dis.readFully(cr.rec, 0, cr.rec.length);

        cr.eventDestId = destid;
        cr.discard = false;
        return cr;
    }

    protected String getUniqueKey() {
        return "???";
    }

    protected boolean isAddOp() {
        return false;
    }

    public String toString() {
        return getUniqueKey() + ", isAddOp() = " + isAddOp();
    }
}

class DestinationUpdateChangeRecord extends ChangeRecord {
    protected String name;
    protected int type;
    protected int operation;

    protected DestinationUpdateChangeRecord(DataInputStream dis)
        throws IOException {
        int version = dis.readInt();

        operation = dis.readInt();
        name = dis.readUTF();
        type = dis.readInt();
    }

    protected String getUniqueKey() {
        return name + ":" + type + ":" + eventDestId;
    }

    protected boolean isAddOp() {
        return (operation == ClusterGlobals.MB_NEW_DESTINATION);
    }
}

class InterestUpdateChangeRecord extends ChangeRecord {
    protected String dname;
    protected String cid;
    protected int operation;

    protected InterestUpdateChangeRecord(DataInputStream dis)
        throws IOException {
        int version = dis.readInt();
        operation = dis.readInt();

        int count = dis.readInt(); // must be 1

        if (operation == ClusterGlobals.MB_NEW_INTEREST) {
            if (dis.readBoolean()) { // is durable?
                dname = dis.readUTF(); // durable name
                cid = dis.readUTF(); // client id
            }
        }
        else if (operation == ClusterGlobals.MB_REM_DURABLE_INTEREST) {
            dname = dis.readUTF(); // durable name
            cid = dis.readUTF(); // client id
        }
    }

    protected String getUniqueKey() {
        return dname + ":" + cid + ":" + eventDestId;
    }

    protected boolean isAddOp() {
        return (operation == ClusterGlobals.MB_NEW_INTEREST);
    }
}

/**
 * Represents a resource to be locked. E.g. durable name, client ID,
 * role of primary queue receiver.
 */
class Resource {
    private String resId = null;
    private ConnectionUID owner = null;
    private long timestamp;
    private long xid;
    private int lockState;

    private int status;
    private HashMap recipients;

    public Resource(String resId) {
        this.resId = resId;
        timestamp = 0;
        xid = 0;

        recipients = new HashMap();
    }

    public String getResId() {
        return resId;
    }

    public ConnectionUID getOwner() {
        return owner;
    }

    public void setOwner(ConnectionUID owner) {
        this.owner = owner;
    }

    public void setTimestamp(long timestamp) {
        this.timestamp = timestamp;
    }

    public long getXid() {
        return xid;
    }

    public int getLockState() {
        return lockState;
    }

    public void setLockState(int lockState) {
        this.lockState = lockState;
    }

    public long getTimestamp() {
        return timestamp;
    }

    public String showRecipients() {
        String ret = "";
        Iterator itr = recipients.keySet().iterator();
        while (itr.hasNext()) {
            BrokerAddress baddr = (BrokerAddress) itr.next();
            ret = ret + "\n\t" + baddr.toString();
        }

        return ret;
    }

    /**
     * Election protocol preparation. Remember the list of brokers
     * that need to vote on this lock request.
     */
    public synchronized void prepareLockRequest(
        BrokerAddress[] brokerList, long xid) {
        recipients.clear();
        for (int i = 0; i < brokerList.length; i++)
            recipients.put( brokerList[i], null);
        this.xid = xid;
        status = ClusterGlobals.MB_LOCK_SUCCESS;
    }

    /**
     * Wait for the conclusion of election protocol.
     */
    public synchronized int waitForStatusChange(long timeout) {
        long waittime = timeout * 1000;
        long endtime = System.currentTimeMillis() + waittime;

        while (status == ClusterGlobals.MB_LOCK_SUCCESS &&
            recipients.size() > 0) {
            try {
                wait(waittime);
            }
            catch (Exception e) {}

            long curtime = System.currentTimeMillis();
            if (curtime > endtime)
                return ClusterGlobals.MB_LOCK_TIMEOUT;

            waittime = endtime - curtime;            
        }

        return status;
    }

    /**
     * Process an election protocol 'vote' from a broker.
     */
    public synchronized void consumeResponse(long xid,
        BrokerAddress sender, int response) {
        if (xid != this.xid)
            return;

        if (status != ClusterGlobals.MB_LOCK_SUCCESS)
            return;

        switch (response) {
        case ClusterGlobals.MB_LOCK_SUCCESS:
            recipients.remove(sender);
            break;

        case ClusterGlobals.MB_LOCK_FAILURE:
        case ClusterGlobals.MB_LOCK_BACKOFF:
            status = response;
            break;
        }

        if (status != ClusterGlobals.MB_LOCK_SUCCESS ||
            recipients.size() == 0)
            notify();
    }

    /**
     * Forfeit an attempt to lock a resource.
     */
    public synchronized void impliedFailure() {
        status = ClusterGlobals.MB_LOCK_FAILURE;
        notify();
    }

    public synchronized void brokerAdded(BrokerAddress broker) {
        status = ClusterGlobals.MB_LOCK_TRY_AGAIN;
        notify();
    }

    public synchronized void brokerRemoved(BrokerAddress broker) {
        if (status != ClusterGlobals.MB_LOCK_SUCCESS)
            return;

        recipients.remove(broker);
        if (recipients.size() == 0)
            notify();
    }
}

/*
 * EOF
 */
