/*
 * 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. 
 */

/*
 * @(#)IMQIPConnection.java	1.10 11/06/07
 */ 

package com.sun.messaging.jmq.jmsserver.service.imq;

import java.net.*;
import java.util.*;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.io.*;

import com.sun.messaging.jmq.jmsserver.service.*;
import com.sun.messaging.jmq.jmsserver.Globals;
import com.sun.messaging.jmq.util.log.Logger;
import com.sun.messaging.jmq.jmsserver.resources.BrokerResources;

import java.security.Principal;
import com.sun.messaging.jmq.jmsserver.auth.AccessController;

import com.sun.messaging.jmq.io.*;
import com.sun.messaging.jmq.jmsserver.util.MetricManager;

import com.sun.messaging.jmq.util.admin.ConnectionInfo;

import com.sun.messaging.jmq.util.timer.MQTimer;

import com.sun.messaging.jmq.util.GoodbyeReason;

import com.sun.messaging.jmq.jmsserver.core.PacketReference;
import com.sun.messaging.jmq.jmsserver.core.Session;
import com.sun.messaging.jmq.jmsserver.util.memory.*;
import com.sun.messaging.jmq.jmsserver.data.PacketRouter;
import com.sun.messaging.jmq.net.IPAddress;
import com.sun.messaging.jmq.jmsserver.util.BrokerException;
import com.sun.messaging.jmq.jmsserver.net.*;
import com.sun.messaging.jmq.jmsserver.service.ConnectionManager;
import com.sun.messaging.jmq.util.lists.*;



public class IMQIPConnection extends IMQBasicConnection 
        implements Operation, MemoryCallback
{

    public static boolean expectPingReply = false;

    /**
     * a value of -1 or 0 turns off feature
     */
    public static final int closeInterval = Globals.getConfig().getIntProperty(Globals.IMQ + 
              ".ping.close.interval", 5);

    public static final boolean enablePingReply = Globals.getConfig().getBooleanProperty(Globals.IMQ + 
              ".ping.reply.enable", true);

    protected int ctrlPktsToConsumer = 0;

    boolean STREAMS = true; 
    boolean BLOCKING = false; 

    protected Set tmpDestinations = Collections.synchronizedSet(new HashSet());

    protected Object timerLock = new Object();

    public int packetVersion = NO_VERSION;

    ConvertPacket convertPkt = null; // for old to new pkts

    public static final int DEFAULT_INTERVAL = 180; // 180sec

   // XXX-CODE TO OVERRIDE BEHAVIOR OF PACKETS

   // to override the type of packet ... 
   //  jmq.packet.[ctrl|read|fill].override = [direct/heap/unset]
   //        direct -> always use direct packets
   //        heap -> always use heap packets
   //        unset -> current behavior
   //  fill is the "waitingForWrite packet"

    private static boolean OVERRIDE_CTRL_PACKET = false;
    private static boolean OVERRIDE_READ_PACKET = false;
    private static boolean OVERRIDE_FILL_PACKET = false;
    private static boolean O_CTRL_USE_DIRECT = false;
    private static boolean O_READ_USE_DIRECT = false;
    private static boolean O_FILL_USE_DIRECT = false;

    static {
        try {
            String ctrlover = Globals.getConfig().getProperty(
            Globals.IMQ + ".packet.ctrl.override");
            if (ctrlover != null && ctrlover.trim().length() > 0) {
                ctrlover = ctrlover.trim();
                if (ctrlover.equalsIgnoreCase("direct")) {
                    OVERRIDE_CTRL_PACKET = true;
                    O_CTRL_USE_DIRECT = true;
                    Globals.getLogger().log(Logger.DEBUG,
                        "DEBUG: Overriding ctrl message "
                        + " packet behavior to DIRECT BUFFERS");
                } else if (ctrlover.equalsIgnoreCase("heap")) {
                    OVERRIDE_CTRL_PACKET = true;
                    O_CTRL_USE_DIRECT = false;
                    Globals.getLogger().log(Logger.DEBUG,
                        "DEBUG: Overriding ctrl message "
                        + " packet behavior to HEAP BUFFERS");
                } else {
                    Globals.getLogger().log(Logger.ERROR, 
                        "DEBUG: Can not determine behavior from "
                        +" imq.packet.ctrl.override = "+ctrlover
                        +"  not one of the valid setting [heap,direct]");
                }
            }
            String readover = Globals.getConfig().getProperty(
                Globals.IMQ + ".packet.read.override");
            if (readover != null && readover.trim().length() > 0) {
                readover = readover.trim();
                if (readover.equalsIgnoreCase("direct")) {
                    OVERRIDE_READ_PACKET = true;
                    O_READ_USE_DIRECT = true;
                    Globals.getLogger().log(Logger.DEBUG,
                        "DEBUG: Overriding read packet"
                        + " behavior to DIRECT BUFFERS");
                } else if (readover.equalsIgnoreCase("heap")) {
                    OVERRIDE_READ_PACKET = true;
                    O_READ_USE_DIRECT = false;
                   Globals.getLogger().log(Logger.DEBUG,
                       "DEBUG: Overriding read packet "
                       + " behavior to HEAP BUFFERS");
                } else {
                    Globals.getLogger().log(Logger.ERROR, 
                        "DEBUG: Can not determine behavior from "
                        + " imq.packet.read.override = "+readover
                        +"  not one of the valid setting [heap,direct]");
                }
            }

            String fillover = Globals.getConfig().getProperty(
            Globals.IMQ + ".packet.fill.override");
            if (fillover != null && fillover.trim().length() > 0) {
                fillover = fillover.trim();
                if (fillover.equalsIgnoreCase("direct")) {
                    OVERRIDE_FILL_PACKET = true;
                    O_FILL_USE_DIRECT = true;
                   Globals.getLogger().log(Logger.DEBUG,
                       "DEBUG: Overriding fill packet "
                       + " behavior to DIRECT BUFFERS");
                } else if (fillover.equalsIgnoreCase("heap")) {
                    OVERRIDE_FILL_PACKET = true;
                    O_FILL_USE_DIRECT = false;
                    Globals.getLogger().log(Logger.DEBUG,
                            "DEBUG: Overriding fill packet "
                            + " behavior to HEAP BUFFERS");
                } else {
                    Globals.getLogger().log(Logger.ERROR, 
                      "DEBUG: Can not determine "
                       + " behavior from jmq.packet.fill.override = "
                       +fillover
                       +"  not one of the valid setting [heap,direct]");
                }
            }
        } catch (Exception ex) {
            Globals.getLogger().logStack(Logger.DEBUG,
                "DEBUG: error setting overrides", ex);
        }
    }

    byte[] empty = {0};

    Object ctrlEL = null;

    /**
     * Ping class
     */
    static class StateWatcher extends TimerTask {
        private int state;
        IMQIPConnection con = null;

        public StateWatcher(int state, IMQIPConnection con) {
            super();
            this.state = state;
            this.con = con;
        }
        public boolean cancel() {
            con = null;
            return super.cancel();
        }

        public void run() {
            con.checkConnection(state);
        }
    }    

    private StateWatcher stateWatcher = null;
    private long interval = DEFAULT_INTERVAL;


    protected ProtocolStreams ps = null;

    protected SocketChannel channel;
    protected InputStream is = null;
    protected OutputStream os = null;

    protected boolean critical = false;

    private boolean flush = false;
    private boolean flushCtrl = false;
    private Object flushLock = new Object();
    private Object flushCtrlLock = new Object();
    private boolean flushCritical = false;
    private boolean lockCritical = false;

    private OperationRunnable read_assigned;
    private OperationRunnable write_assigned;

    /**
     * constructor
     */


    public IMQIPConnection(Service svc, ProtocolStreams ps, 
             PacketRouter router) 
        throws IOException, BrokerException
    {
        super(svc, router);
        this.ps = ps;

        InetAddress ia = null;
        if (ps != null) {
            ia = ps.getRemoteAddress();
            if (ia != null) {
                this.setRemoteIP(ia.getAddress());
            }

            STREAMS = (ps.getChannel() == null);
            BLOCKING = ps.getBlocking();
            channel = (SocketChannel)ps.getChannel();
            is = ps.getInputStream();
            os = ps.getOutputStream();
        }
        accessController = AccessController.getInstance(svc.getName(),
                                                        svc.getServiceType());
        if (ia != null) {
            accessController.setClientIP(ia.getHostAddress());
        }

        // LKS - XXX we want notification, but it doesnt need to have
        // filters -> could created an orderedNLset

        this.control = new NFLPriorityFifoSet();
        ctrlEL = this.control.addEventListener(this,EventType.EMPTY,  null);

        setConnectionState(Connection.STATE_CONNECTED);
        waitingWritePkt = new Packet(OVERRIDE_FILL_PACKET 
                ? O_FILL_USE_DIRECT : !STREAMS);

        if (!isAdminConnection() && Globals.getMemManager() != null)
            Globals.getMemManager().registerMemoryCallback(this);
    }

// -------------------------------------------------------------------------
//   General connection information and metrics
// -------------------------------------------------------------------------
    public void dumpState() {
	super.dumpState();

        logger.log(Logger.INFO,
                "\tcontrol = " + control.size());
        logger.log(Logger.INFO,
                "\tread_assigned = " + read_assigned);
        logger.log(Logger.INFO,
                "\twrite_assigned = " + write_assigned);
        if (ninfo != null)
            ninfo.dumpState();
    }

    public int getLocalPort() {
        if (ps == null) return 0;
        return ps.getLocalPort();
    }

    /** 
     * The debug state of this object
     */
    public synchronized Hashtable getDebugState() {
        Hashtable ht = super.getDebugState();
        ht.put("pkts[TOTAL](in,out) ", "("+msgsIn + "," + (ctrlPktsToConsumer + 
                     msgsToConsumer) + ")");
        for (int i=0; i < pktsIn.length; i ++) {
            if (pktsIn[i] == 0 && pktsOut[i] == 0)
                continue;
            ht.put("pkts[" + PacketType.getString(i) + "] (in,out)",
                   "("+pktsIn[i] + "," + pktsOut[i] + ")");
        }
        ht.put("ctrlPktsToConsumer", String.valueOf(ctrlPktsToConsumer));
        ht.put("critical", String.valueOf(critical));
        ht.put("controlSize", String.valueOf(control.size()));
        if (control.size() > 0) {
             Vector v = new Vector();
             Iterator itr = control.iterator();
             while (itr.hasNext()) {
                Packet p = (Packet)itr.next();
                v.add(p.toString());
             }
             ht.put("control", v);
        }
        if (ps != null)
            ht.put("transport", ps.getDebugState());
        return ht;
    }


    public Vector getDebugMessages(boolean full) {
        Vector ht = new Vector();
        synchronized(control) {
            Iterator itr = control.iterator();
            while (itr.hasNext()) {
                PacketReference pr = (PacketReference)itr.next();
                ht.add((full ? pr.getPacket().dumpPacketString()
                   : pr.getPacket().toString()));
            }
        }
        return ht;
       
    }

    public ConnectionInfo getConnectionInfo() {
	    coninfo = super.getConnectionInfo();
        coninfo.remPort = (ps == null ? 0 : ps.getRemotePort());
        return coninfo;
    }

 
    public boolean useDirectBuffers() {
        return (OVERRIDE_CTRL_PACKET ? O_CTRL_USE_DIRECT : !STREAMS);
        // return !STREAMS;
    }

// -------------------------------------------------------------------------
//   Basic Operation implementation
// -------------------------------------------------------------------------

    NotificationInfo ninfo = null;

    public synchronized AbstractSelectableChannel getChannel() {
        if (ps == null) 
            return null;
        return ps.getChannel();
    }

    public boolean canKill() {
        return !critical;
    }

    public void setCritical(boolean critical) {
        this.critical = critical;
    }

    public boolean waitUntilDestroyed(long time) {
        long targettime = System.currentTimeMillis() + time;
        while (isValid() && System.currentTimeMillis() < targettime) {
            waitForWork(time);
        }
        return isValid();
    }

    Object releaseWaitLock = new Object();

    /**
     * called when the thread is not longer processing the
     * operation
     * this mask used when a thread is assigned or released,
     * which is returned from the notificationInfo object
     * (if any)
     */
    public synchronized void notifyRelease(
               OperationRunnable runner, int events) 
    {
        /* 
         * the thread is giving us UP ...
         */
        int release_events = 0;
        if ((events & SelectionKey.OP_WRITE) > 0 && runner == write_assigned) {
            release_events = release_events | SelectionKey.OP_WRITE;
            write_assigned = null;
        }
        if ((events & SelectionKey.OP_READ) > 0 && runner == read_assigned) {
            release_events = release_events | SelectionKey.OP_READ;
            read_assigned = null;
        }

        if (ninfo != null && release_events != 0) {
            ninfo.released(this, release_events);
        }
        notifyAll();
        return;
    }

    public synchronized void waitForRelease(long timeout) {
        long waitt = 5000;
        while (read_assigned != null || write_assigned != null) { 
            if (timeout <= 0) {
                Globals.getLogger().log(Logger.WARNING, 
                "Timeout in waiting for runnable threads release in "+this);
                return;
            }
            Globals.getLogger().log(Logger.INFO, "Waiting for runnable threads release in "+this);
            if (timeout < waitt) waitt = timeout;
            try {
                wait(waitt); 
            } catch (InterruptedException e) {
                Globals.getLogger().log(Logger.WARNING, 
                "Interrupted in waiting for runnable threads release in "+this);
                return;
            }
            timeout -= waitt;
        }
    }

    public synchronized void clearAssigned() {
        read_assigned = null;
        write_assigned = null;
    }

    public synchronized OperationRunnable getReadRunnable() {
        return read_assigned;
    }

    public synchronized OperationRunnable getWriteRunnable() {
        return write_assigned;
    }

    public synchronized void threadAssigned(
            OperationRunnable runner, int events) 
        throws IllegalAccessException 
    {
        int release_events = 0;
        if ((events & SelectionKey.OP_WRITE) > 0) {
            if (write_assigned != null)  {
                 // we havent released yet ... its a timing thing
                 release_events = release_events | SelectionKey.OP_WRITE;
            }
            write_assigned = runner;
         }
         if ((events & SelectionKey.OP_READ) > 0) {
            if (read_assigned != null)  {
                 // we havent released yet ... its a timing thing
                release_events = release_events | SelectionKey.OP_READ;
            }
            read_assigned = runner;
         }
 
         if (ninfo != null) {
            if (release_events != 0)
                ninfo.released(this, release_events);
            ninfo.assigned(this, events);
         }
    }

    public void attach(NotificationInfo obj) {
        ninfo = obj;
    }

    public NotificationInfo attachment() {
        return ninfo;
    }

    private String getKeyString(int events) {
        String str = "";
        if ((events & SelectionKey.OP_WRITE) > 0) {
            str += " WRITE ";
        }
        if ((events & SelectionKey.OP_READ) > 0) {
            str += " READ ";
        }
        return str;
    }


    public boolean process(int events, boolean wait) 
        throws IOException
    {
         boolean didSomething = false;
         boolean processedLastIteration = true;

         int readcount = 0;
         int writecount = 0;

         try {
             while (processedLastIteration) {
                 // process all writes
                 processedLastIteration = false;
                 if ((events & SelectionKey.OP_WRITE) > 0) {
                     while (true) {
    
                         if (writeData(wait) != 
                            Operation.PROCESS_PACKETS_REMAINING) 
                         {
                             // wasnt able to write anymore
                             // break out of the loop
                             break;
                         } 
                         processedLastIteration = true;
                         writecount++;
                     }
                 }
                 // process one write
                 if ((events & SelectionKey.OP_READ) > 0) {
                     int returnval = readData();
                     switch (returnval) {
                         case Operation.PROCESS_PACKETS_REMAINING:
                             processedLastIteration = true;
                         case Operation.PROCESS_PACKETS_COMPLETE:
                         case Operation.PROCESS_WRITE_INCOMPLETE:
                     }
                     readcount++;
                 }
                 didSomething |= processedLastIteration;
              }
         } catch (OutOfMemoryError ex) {
             logger.log(Logger.ERROR,
                 BrokerResources.E_FORCE_CON_CLOSE,
                 this, ex);
             int count = 0;
             boolean firstpass = true;
             while (true) {
                try {        
                    logger.log(Logger.ERROR,
                            "Received out of memory error trying to shutdown");
                    closeConnection(firstpass, GoodbyeReason.CON_FATAL_ERROR,
                            ex.toString());
                    firstpass = false;
                    break;
                } catch (OutOfMemoryError err1) {
                    logger.log(Logger.DEBUG,
                         "Connection could not be cleanly closed,"
                         + " trying again on " + 
                         this, ex);
                   count ++;
                   if (count >= 2) {
                       throw err1;
                   }
                }
             }
         } catch (IOException thr) {
             // rethrow
             throw thr;
         } catch (Throwable thr) {
             logger.logStack(Logger.ERROR, "Internal Error: "
                     + "Received unexpected exception processing connection "
                     + " closing connection", thr);
             // something went wrong, close connection
             closeConnection(true, GoodbyeReason.CON_FATAL_ERROR,
                     thr.toString());
         }

         return !didSomething;
    }

    public String getRemoteConnectionString() {
        if (remoteConString != null)
            return remoteConString;

        boolean userset = false;

        String remotePortString = "???";
        String userString = "???";

        if (state >= Connection.STATE_AUTHENTICATED) {
            try {
                Principal principal = getAuthenticatedName();
                if (principal != null) {
                    userString = principal.getName();
                    userset = true;
                }
            } catch (BrokerException e) { 
                if (IMQBasicConnection.DEBUG)
                    logger.log(Logger.DEBUG,"Exception getting authentication name "
                        + conId, e );
                        
            }
        }

        remotePortString = Integer.toString((ps == null 
                 ? 0 :ps.getRemotePort()));

        String retstr = userString + "@" +
            IPAddress.rawIPToString(getRemoteIP(), true, true) + ":" +
            remotePortString;
        if (userset) remoteConString = retstr;
        return retstr;
    }

    String localsvcstring = null;
    protected String localServiceString() {
        if (localsvcstring != null)
            return localsvcstring;
        String localPortString = "???";
        localPortString = Integer.toString((ps == null 
                           ? 0 : ps.getLocalPort()));
        localsvcstring = service.getName() + ":" + localPortString;
        return localsvcstring;
    }

// -------------------------------------------------------------------------
//   Basic Connection Management
// -------------------------------------------------------------------------

    public synchronized void closeConnection(
            boolean force, int reason, String reasonStr) 
    { 

        if (state >= Connection.STATE_CLOSED)  {
             logger.log(logger.DEBUG,"Requested close of already closed connection:"
                    + this);
             return;
        }
               
        stopConnection();
        if (Globals.getMemManager() != null)
             Globals.getMemManager().removeMemoryCallback(this);
        if (force) { // we are shutting it down, say goodbye
            sayGoodbye(false, reason, reasonStr);
            flushControl(1000);
        }
        
        // CR 6798464: Don't mark connection as closed until we've flushed the queue and sent the GOODBYE
        state = Connection.STATE_CLOSED;
        notifyConnectionClosed();
        
        // clean up everything 
        this.control.removeEventListener(ctrlEL);
        cleanup(reason == GoodbyeReason.SHUTDOWN_BKR);
        // OK - we are done with the flush, we dont need to be
        // notified anymore
        if (ninfo != null)
            ninfo.destroy(reasonStr);

        try {
            if (ps != null) 
                ps.close(); // close socket
            ps = null;
        } catch (IOException ex) {
            // its OK if we cant close the socket ..,
            // aother thread may have already done it
            // ignore
        }

        if (reason == GoodbyeReason.SHUTDOWN_BKR) {
            cleanupConnection(); // OK if we do it twice
        } else {
            cleanup(false);
        }
    }

    protected void cleanupControlPackets(boolean shutdown) {
        while (!control.isEmpty()) {
            Packet p = (Packet) control.removeNext();
            if (p == null) continue;
            p.destroy();
        }
    }


    int destroyRecurse = 0;
    /**
     * destroy the connection to the client
     * clearing out messages, etc
     */
    public void destroyConnection(boolean force, int reason, String reasonstr) { 
        int oldstate = 0;
        boolean destroyOK = false;
        try {

            synchronized (this) {
                oldstate = state;
                if (state >= Connection.STATE_DESTROYING)
                    return;
    
                if (state < Connection.STATE_CLOSED) {
                     closeConnection(force, reason, reasonstr);
                }
    
                setConnectionState(Connection.STATE_DESTROYING);
            }
            Globals.getConnectionManager().removeConnection(getConnectionUID(),
                   force, reason, reasonstr);
    
            if (accessController.isAuthenticated()) {
                accessController.logout();
            }

            // The connection is going away. Deposit our metric totals
            // with the metric manager
            MetricManager mm = Globals.getMetricManager();
            if (mm != null) {
                mm.depositTotals(service.getName(), counters);
            }

            // Clear, just in case we are called twice
            counters.reset();

            synchronized (timerLock) {

                if (stateWatcher != null) {
                    try {
                        stateWatcher.cancel();
                    } catch (IllegalStateException ex) {
                        logger.log(Logger.DEBUG,"Error destroying "+
                            " connection "  + this + " to state " +
                            state, ex);
                    }
                    stateWatcher = null;
                }
            }

	    if (this.getDestroyReason()!= null){
		logConnectionInfo(true, this.getDestroyReason());
            } else {
            	logConnectionInfo(true, reasonstr);
            }

            setConnectionState(Connection.STATE_DESTROYED);
            destroyOK = true;
            wakeup();
        } finally {
            if (!destroyOK && reason != GoodbyeReason.SHUTDOWN_BKR 
                    &&  (Globals.getMemManager() == null 
                    || Globals.getMemManager().getCurrentLevel() > 0)) {

                state = oldstate;
                if (destroyRecurse < 2) {
                    destroyRecurse ++;
                    destroyConnection(force, reason, reasonstr);
                }
            } 
                
            // free the lock
            Globals.getClusterBroadcast().connectionClosed(
                getConnectionUID(), isAdminConnection());
        }
    }

    /**
     * sets the connection state 
     * @return false if connection being destroyed
     */
    public boolean setConnectionState(int state) { 
        synchronized (timerLock) {
            this.state = state;
            if (this.state >= Connection.STATE_CLOSED) {
                if (stateWatcher != null) {
                    try {
                        stateWatcher.cancel();
                    } catch (IllegalStateException ex) {
                        logger.log(Logger.DEBUG,"Error setting state on "+
                            " connection "  + this + " to state " +
                            state, ex);
                    }
                    stateWatcher = null;
                }
                wakeup();
		return false;
            } else if (state == Connection.STATE_CONNECTED) {
                interval = Globals.getConfig().getLongProperty(
                   Globals.IMQ + ".authentication.client.response.timeout",
                   DEFAULT_INTERVAL);
                MQTimer timer = Globals.getTimer(true);
                stateWatcher = new StateWatcher(Connection.STATE_INITIALIZED, this);
                try {
                    timer.schedule(stateWatcher, interval*1000);
                } catch (IllegalStateException ex) {
                    logger.log(Logger.DEBUG,"InternalError: timer canceled ", ex);
                }

            } else if (state == Connection.STATE_INITIALIZED 
                   || state == Connection.STATE_AUTH_REQUESTED
                   || state == Connection.STATE_AUTH_RESPONSED) {
                if (stateWatcher != null) {
                    try {
                        stateWatcher.cancel();
                    } catch (IllegalStateException ex) {
                        logger.log(Logger.DEBUG,"Error setting state on "+
                            " connection "  + this + " to state " +
                            state, ex);
                    }
                    stateWatcher = null;
                }
                // if next state not from client, return 
                if (state == Connection.STATE_INITIALIZED) {
                    return true;
                }
                if (state == Connection.STATE_AUTH_RESPONSED) {
                    return true;
                }

                MQTimer timer = Globals.getTimer(true);
                stateWatcher = new StateWatcher(
                        Connection.STATE_AUTH_RESPONSED, this);
                try {
                    timer.schedule(stateWatcher, interval*1000);
                } catch (IllegalStateException ex) {
                    logger.log(Logger.DEBUG,"InternalError: timer canceled ", ex);
                }
            } else if (state >= Connection.STATE_AUTHENTICATED 
                    || state == Connection.STATE_UNAVAILABLE) 
            {
                if (stateWatcher != null) {
                    try {
                        stateWatcher.cancel();
                    } catch (IllegalStateException ex) {
                        logger.log(Logger.DEBUG,"Error setting state on "+
                            " connection "  + this + " to state " +
                            state, ex);
                    }
                    stateWatcher = null;
                }
                if (state == Connection.STATE_AUTHENTICATED) {
                    logConnectionInfo(false);
                }
            }
        }
        return true;
    }

    public void logConnectionInfo(boolean closing) {
        this.logConnectionInfo(closing,"Unknown");
    }

    public void logConnectionInfo(boolean closing, String reason) {

        String[] args = {
            getRemoteConnectionString(),
            localServiceString(),
            Integer.toString(Globals.getConnectionManager().size()),
            reason,
            String.valueOf(control.size()),
            Integer.toString(service.size())
        };

        if (!closing) {
            logger.log(Logger.INFO, BrokerResources.I_ACCEPT_CONNECTION, args);
        } else {
            logger.log(Logger.INFO, BrokerResources.I_DROP_CONNECTION, args);
        }
    }


// -------------------------------------------------------------------------
//   Queuing Messages
// -------------------------------------------------------------------------

    private NFLPriorityFifoSet control = null;
    boolean hasCtrl = true;


    /**
     * send a control (reply) message back to the client
     *
     * @param msg message to send back to the client
     */
    public void sendControlMessage(Packet msg) {
        if (!isValid() && msg.getPacketType() != PacketType.GOODBYE ) {
            logger.log(Logger.INFO,"Internal Warning: message " + msg
                  + "queued on destroyed connection " + this);
        }
            
        control.add(msg);
        synchronized (control) {
            hasCtrl = !control.isEmpty();
        }
    }

    protected void sendControlMessage(Packet msg, boolean priority)
    {
        if (IMQBasicConnection.DEBUG) {
            logger.log(Logger.DEBUGHIGH, 
                "IMQIPConnection[ {0} ] queueing Admin packet {1}", 
                this.toString(), msg.toString());
        }
        if (!isValid()) { // we are being destroyed
            return;
        }
        // I am assuming that there isnt any issue
        // with priority
        // if there is, I may need a new list type
        assert priority == false;
        sendControlMessage(msg);
    }


    /** 
     * Flush all control messages on this connection to
     * the client.
     * @param timeout the lenght of time to try and flush the
     *         messages (0 indicates wait forever)
     */
    public void flushControl(long timeout) {

        if (read_assigned == write_assigned && read_assigned != null) {
            localFlushCtrl();
            return;
        }

        synchronized (flushCtrlLock) { 
            if (IMQBasicConnection.DEBUG) {
                logger.log(Logger.DEBUG,
                        "Flushing Control Messages with timeout of " + timeout);
            }
            // dont worry about syncing here -> if we miss the
            // window we should still be woken up w/ the ctrl 
            // notify -> since that happens AFTER a message is
            // removed from the list
            if (ctrlpkt == null && control.isEmpty() && !flushCritical)
                return;
            if (!isValid()) {
                return;
            }
            long time = System.currentTimeMillis();
            flushCtrl = true;
            if (timeout < 0) {
                return;
            }
            while (flushCtrl && isValid()) {
                try {
                    if (timeout != 0) {
                        flushCtrlLock.wait(timeout);
                    } else {
                        flushCtrlLock.wait(1000 /* 1 second */);
                    }
                } catch (InterruptedException ex) {
                   // no reason to do anything w/ it
                }
                if (flushCtrl && timeout > 0 && 
                    System.currentTimeMillis() >= time+timeout)
                    break;
            }
            flushCtrl = false;
            if (IMQBasicConnection.DEBUG) {
                if (flush) {
                    logger.log(Logger.DEBUG,
                        "Control Flush did not complete in timeout of " 
                        + timeout);
                } else {
                    logger.log(Logger.DEBUG,
                            "Contrl Flush completed");
                }
            }
        }
    }

    private void localFlushCtrl() {
        // OK .. if we are in the SAME thread as write do it inline
        flushCtrl = true;
        try {
            while (writeData(false) != Operation.PROCESS_PACKETS_COMPLETE) {
            }
        } catch (Exception ex) {
            // got exception while flushing, connection is probably gone
            logger.log(Logger.DEBUG,"error in flush " + this , ex);
        }
        flushCtrl = false;

    }
    private void localFlush() {
        // OK .. if we are in the SAME thread as write do it inline
        flush = true;
        try {
            while (writeData(false) != Operation.PROCESS_PACKETS_COMPLETE) {}
        } catch (Exception ex) {
            // got exception while flushing, connection is probably gone
            logger.log(Logger.DEBUG,"error in flush " + this , ex);
        }
        flush = false;

    }

    /** 
     * Flush all control and JMS messages on this connection to
     * the client.
     * @param timeout the lenght of time to try and flush the
     *         messages (0 indicates wait forever)
     */
    public  void flush(long timeout) {
        if (read_assigned == write_assigned && read_assigned != null) {
            localFlush();
            return;
        }
        if ( !inCtrlWrite && control.isEmpty() 
            && !inJMSWrite && hasBusySessions()
            && !flushCritical && !lockCritical) 
        {
            // nothing to do
            return;
        }
        synchronized (flushLock) { 
            if (IMQBasicConnection.DEBUG) {
                logger.log(Logger.DEBUG,
                        "Flushing Messages with timeout of " + timeout);
            }
            if (!isValid())
                return;
            // dont worry about syncing here -> if we miss the
            // window we should still be woken up w/ the ctrl 
            // notify -> since that happens AFTER a message is
            // removed from the list
            if ( !inCtrlWrite && control.isEmpty() 
                && !inJMSWrite && hasBusySessions()
                && !flushCritical && !lockCritical)
            {
                // nothing to do
                return;
            }
            long time = System.currentTimeMillis();
            flush = true;
            while (flush && isValid()) {
                try {
                    if (timeout != 0) {
                        flushLock.wait(timeout);
                    } else {
                        flushLock.wait(1000 /* 1 second */);
                    }
                } catch (InterruptedException ex) {
                    // valid, no reason to check
                }
                if (flush && timeout > 0 && 
                    System.currentTimeMillis() >= time+timeout)
                    break;
            }
            if (IMQBasicConnection.DEBUG) {
                if (flush) {
                    logger.log(Logger.DEBUG,
                            "Flush did not complete in timeout of " + timeout);
                } else {
                    logger.log(Logger.DEBUG,
                            "Flush completed");
                }
            }
        }
    }




    void dumpConnectionInfo() {
        if (ninfo != null) {
            logger.log(Logger.INFO,"Connection Information [" +this +
              "]" + ninfo.getStateInfo());
        }
    }

    void checkConnection(int state) {

        synchronized(timerLock) {
            try {
                stateWatcher.cancel();
            } catch (IllegalStateException ex) {
                logger.log(Logger.DEBUG,"Error destroying "+
                    " connection "  + this + " to state " +
                    state, ex);
            }
            stateWatcher = null;
        }
        String[] args = {toString(), getConnectionStateString(this.state),
                         getConnectionStateString(state),
                         String.valueOf(interval)};
        synchronized (this) { 
            if (this.state >= state) 
            { 
                return;
            }
            if (this.state >= Connection.STATE_CLOSED 
                || this.state == Connection.STATE_UNAVAILABLE) 
            {
                return;
            }

            logger.log(Logger.WARNING, 
                 Globals.getBrokerResources().getKString(
				   BrokerResources.W_CONNECTION_TIMEOUT, args));
        }

        // FOR DEBUG ... add additional state information
        if (IMQBasicConnection.DEBUG) {
            dumpConnectionInfo();
        }

       // dont bother being nice
        destroyConnection(false, GoodbyeReason.CON_FATAL_ERROR,
            Globals.getBrokerResources().getKString(
            BrokerResources.W_CONNECTION_TIMEOUT, args)); 
    }




// -------------------------------------------------------------------------
//   Receiving Messages
// -------------------------------------------------------------------------

    Packet readpkt = null;


    // flag used with assertions to make sure
    // that the thread pool never assigns this
    // thread twice at one time

    private boolean inReadProcess = false;

    // info of the last good packet read on a connection for diagnostic
    private int lastPacketType = 0;
    private int lastPacketSize = 0;


    // new method to handle how we get the packet
    // this is overridden in Embedded more to get it from a queue
    protected boolean readInPacket(Packet p) 
           throws IllegalArgumentException, StreamCorruptedException, BigPacketException, IOException
    {
        boolean OK = true;
            
        if (STREAMS) {
           assert is!= null;
           readpkt.readPacket(is);
        } else {
           assert channel != null;
           OK = readpkt.readPacket(channel, BLOCKING);
         }
         return OK;
    }

    // used for subclasses if we don't want to recreate a packet
    protected Packet clearReadPacket(Packet p) {
        return null;
    }


    public int readData()
         throws IOException, BrokerException /* operation incomplete */
    {
        assert inReadProcess == false;


        try {
            inReadProcess = true;

            if (IMQBasicConnection.DEBUG || DUMP_PACKET || IN_DUMP_PACKET) {
                logger.log(Logger.DEBUG,
                        "Reading from " + getClass() + "{0} ", 
                        this.toString() 
                        + Thread.currentThread());
            }
    
           if (!isValid()) {
                if (IMQBasicConnection.DEBUG) {
                    logger.log(Logger.DEBUG,
                            "Invalid Connection {0} ", 
                            this.toString() + 
                            Thread.currentThread());
                }
               throw new IOException(
                    "Connection has been closed " + this);
           }
    
    
           try {
               if (readpkt == null) { // heck its a new packet
                  readpkt = new Packet(OVERRIDE_READ_PACKET 
                                 ? O_READ_USE_DIRECT 
                                 : !STREAMS);
                  readpkt.generateSequenceNumber(false);
                  readpkt.generateTimestamp(false);
                  if (IMQBasicConnection.DEBUG) {
                      logger.log(Logger.DEBUG,  
                          "IMQIPConnection {0} getting a new read packet {1} ", 
                          this.toString(), readpkt.toString());
                  }
               }
               assert readpkt != null;
           } catch (OutOfMemoryError err) {
               // ack .. got error loading header
               // Dump header to help with debugging (i.e. was it an
               // unusually large packet? Corrupted read?)
               Globals.handleGlobalError(err,
                    Globals.getBrokerResources().getKString(
                    BrokerResources.M_LOW_MEMORY_READALLOC) + ": " +
                    readpkt.headerToString());
           }

           if (isValid()) {
               try {
                   boolean OK = true;
            
                   OK = readInPacket(readpkt);
                   msgsIn ++;
                   if (readpkt.getPacketType() < PacketType.LAST)
                       pktsIn[readpkt.getPacketType()] ++;
    
                   if (!OK) { // we didnt finish reading
                       return Operation.PROCESS_WRITE_INCOMPLETE;

                   } else if (packetVersion == NO_VERSION) {
                       // ok this is the first packet, get the
                       // version of protocol the client is talking
                       packetVersion = readpkt.getVersion();


                       // XXX-LKS need to use protocol version
                       // not just packet version

                       if (packetVersion < CURVERSION)
                           convertPkt = new ConvertPacket(this, 
                                    packetVersion,
                                    CURVERSION);
                   }
                   // convert to new packet type if necessary
                   if (convertPkt != null)
                       convertPkt.handleReadPacket(readpkt); 
    
               } catch (IllegalArgumentException ex) {
                   logger.log(Logger.INFO,"Internal Error ", ex);
                   // queue a HELLO_REPLY w/ error
                   Packet pkt = new Packet(useDirectBuffers());
                   pkt.setIP(ipAddress);
                   pkt.setPort(getLocalPort());
    
                   pkt.setPacketType(PacketType.HELLO_REPLY);
                   Hashtable hash = new Hashtable();
                   hash.put("JMQStatus", new Integer(Status.BAD_VERSION));
                   pkt.setProperties(hash);
                   sendControlMessage(pkt);
                   flushControl(1000);
                   destroyConnection(true, GoodbyeReason.CON_FATAL_ERROR,
                           (getDestroyReason() == null ? ex.toString()
                          : getDestroyReason()));
                   throw ex;
                       
               } catch (OutOfMemoryError ex) {
                   // Dump header to help with debugging (i.e. was it an
                   // unusually large packet? Corrupted read?)
                   Globals.handleGlobalError(ex,
                    Globals.getBrokerResources().getKString(
                    BrokerResources.M_LOW_MEMORY_READALLOC) + ": " +
                    readpkt.headerToString());
    
                   // re-read the packet ... 
                   //  in 99.??? % of the time, we just lost a packet
                   // 
                   // if we fail a second time  or get an unexpected
                   // error ... its fatal for the connection
                    boolean OK = readInPacket(readpkt);
                    if (!OK) { // we didnt finish reading
                        return Operation.PROCESS_WRITE_INCOMPLETE;
                    }
    
                } catch (StreamCorruptedException ex) {
                   String connStr = getRemoteConnectionString();
                    logger.logStack(Logger.WARNING,
                        BrokerResources.W_STREAM_CORRUPTED, connStr, ex);
                    logger.log(Logger.WARNING,
                        "Last good packet received from connection " +
                        connStr + ": type = " + lastPacketType +
                        ", size = " + lastPacketSize);

                    throw ex;
                } catch (BigPacketException e) {
                    // The packet exceeded the maximum packet size. This
                    // Should only occur for JMS message packets since all
                    // control packets are relatively small.
                    // We ignore the packet, log a warning and send a
                    // reply indicating the error.
                    String params[] = {
                        String.valueOf(readpkt.getPacketSize()),
                        readpkt.toString(),
                        String.valueOf(readpkt.getMaxPacketSize())
                        };
                    logger.log(Logger.WARNING, 
                        BrokerResources.X_IND_PACKET_SIZE_EXCEEDED,
                        params, e);
                    sendReply(readpkt, Status.ENTITY_TOO_LARGE);
                    // Packet is garbage. Null it out so we don't reuse it.
                    readpkt.reset();
                    readpkt = clearReadPacket(readpkt);
                    return Operation.PROCESS_PACKETS_REMAINING;
                }
    
                if (Globals.getConnectionManager().PING_ENABLED) {
                    updateAccessTime(true);
                }
	            if (METRICS_ON) {
                    countInPacket(readpkt);
                }
            }
                      
            if (IMQBasicConnection.DEBUG || DUMP_PACKET || IN_DUMP_PACKET) {
                int flag = (DUMP_PACKET || IN_DUMP_PACKET) ? Logger.INFO
                    : Logger.DEBUGHIGH;

                logger.log(flag, "\n------------------------------"
                    + "\nReceived incoming Packet - Dumping"
                    + "\nConnection: " + this 
                    + "\n------------------------------"
                    + "\n" + readpkt.dumpPacketString(">>>>****") 
                    + "\n------------------------------");
            }

            // Save info of the last good packet read
            lastPacketType = readpkt.getPacketType();
            lastPacketSize = readpkt.getPacketSize();

            try {
                if (MemoryGlobals.MEM_EXPLICITLY_CHECK ||
                    (MemoryGlobals.MEM_QUICK_CHECK && readpkt.getPacketSize() >
                     MemoryGlobals.MEM_SIZE_TO_QUICK_CHECK)) {
                    if (Globals.getMemManager() != null)
                        Globals.getMemManager().quickMemoryCheck();
                }
                setCritical(true);
                if (!isValid())  {
                    return Operation.PROCESS_PACKETS_COMPLETE;
                }
                try {
   	                router.handleMessage(this, readpkt);
                    readpkt = clearReadPacket(readpkt);
                } catch (OutOfMemoryError ex) {
                    logger.logStack(Logger.ERROR, 
                        BrokerResources.E_LOW_MEMORY_FAILED, 
                        ex);
                    throw ex;
                }

            } finally { 
                 setCritical(false);
            }
            return Operation.PROCESS_PACKETS_REMAINING;

        } finally {
            // used for thread safety assert
            inReadProcess = false;
        }

    } 


// -------------------------------------------------------------------------
//   Sending Messages
// -------------------------------------------------------------------------


    private Packet ctrlpkt = null;
    private Packet waitingWritePkt = null;


    /**
     * indicates that we were interrupted
     * during a control pkt write
     */
    private boolean inCtrlWrite = false;

    /**
     * indicates that we were interrupted
     * during a JMS pkt write
     */
    private boolean inJMSWrite = false;



    // flag which is used w/ asserts to make
    // sure that there is no competition in
    // the writeData processing
    boolean inWriteProcess = false;

    protected boolean writeOutPacket(Packet p) throws IOException
    {
         boolean donewriting = true;
         if (STREAMS) {
             p.writePacket(os);
         } else {
             donewriting = p.writePacket(channel, BLOCKING);
         }
         return donewriting;
    }

    protected Packet clearWritePacket(Packet p)
    {
         if (p != null)
             p.destroy();
         return null;
    }

    public int writeData(boolean wait)
         throws IOException {

        assert inWriteProcess == false;

        if (hasCtrl) {
            synchronized (control) {
                hasCtrl = !control.isEmpty();
            }
        }
        try { // catch out of memory errors
            inWriteProcess = true;

            if (IMQBasicConnection.DEBUG) {
                logger.log(Logger.DEBUG,  
                    "Writing IMQIPConnection {0} ", this.toString());
            }

            while (wait && !isBusy() && isValid() ) {
                // we are using a blocking thread pool
                // wait for someone to notify us
                waitForWork(0 /* wait forever */);
            }
            if (!isValid()) {
                throw new IOException(
                        Globals.getBrokerResources().getKString(
                        BrokerResources.X_INTERNAL_EXCEPTION,
                        "destroyed Connection"));
            }
            if (!isBusy()) {
                return Operation.PROCESS_PACKETS_COMPLETE;
            }    

            
            flushCritical = true;

            if ((!inCtrlWrite) && (!inJMSWrite)) {
                // we are not completing a packet
                if (!control.isEmpty()) {
                    ctrlpkt = (Packet) control.removeNext();
                    if (ctrlpkt != null) {
                        inCtrlWrite = true;
                    }
                } else {
                    synchronized (control) {
                        hasCtrl = !control.isEmpty();
                    }
                }

            }
            if (ctrlpkt != null)  {
                // ok , we have a new packet
                if (ctrlpkt.getPacketType() > PacketType.MESSAGE) {
                    ctrlpkt.setIP(ipAddress);
                    ctrlpkt.setPort(getLocalPort());
                }
                // first convert it
                if (convertPkt != null)
                    convertPkt.handleWritePacket(ctrlpkt); 
                if (IMQBasicConnection.DEBUG || DUMP_PACKET || OUT_DUMP_PACKET) {
                    int flag = (DUMP_PACKET || OUT_DUMP_PACKET) 
                            ? Logger.INFO : Logger.DEBUGHIGH;
                    logger.log(flag, "\n------------------------------"
                            +"\nSending Control Packet -[block = "+BLOCKING 
                            + ",nio = "+!STREAMS+"]   Dumping"
                            + "\n------------------------------"
                            + "\n" + ctrlpkt.dumpPacketString("<<<<****")
                            + "\n------------------------------");
                }
                inCtrlWrite = !writeOutPacket(ctrlpkt);
                if (!inCtrlWrite) { // we are done

                   if (ctrlpkt.getPacketType() < PacketType.LAST)
                       pktsOut[ctrlpkt.getPacketType()] ++;

                    ctrlPktsToConsumer ++;
                    if (IMQBasicConnection.DEBUG || DUMP_PACKET || OUT_DUMP_PACKET) {
                        logger.log(Logger.INFO,
                                "Finished writing packet [" + ctrlpkt +"]");
                    }
                    if (Globals.getConnectionManager().PING_ENABLED) {
                        updateAccessTime(false);
                    }
                    if (METRICS_ON) {
                        countOutPacket(ctrlpkt);
                    }

                    assert ctrlpkt != null;

                    ctrlpkt = clearWritePacket(ctrlpkt);;
                }

                // the broker is no longer in a critical state
                flushCritical = false;
                synchronized(stateLock) {
                    checkState();
                }

                // we were interrupted during the write
                if (inCtrlWrite)  {
                     return Operation.PROCESS_WRITE_INCOMPLETE;
                } else if (isBusy()) {
                    return Operation.PROCESS_PACKETS_REMAINING;
                } else {
                    return Operation.PROCESS_PACKETS_COMPLETE;
                }

            }    

            // the broker is no longer in a critical state
            flushCritical = false;

            // OK .. now try normal messages
    
            if (IMQBasicConnection.DEBUG) {
                logger.log(Logger.DEBUGHIGH, 
                    "IMQIPConnection[ {0} ] - processing "
                    + " normal msg queue", this.toString());
            }

           
            // we shouldnt be here if we are paused or waiting for a resume


            if (!inJMSWrite && !hasCtrl) {
                lockCritical = true;
                boolean validJMSPkt = false;

                assert waitingWritePkt != null;
                if (hasCtrl || !runningMsgs || paused || waitingForResumeFlow ||
                    ((validJMSPkt = fillNextPacket(waitingWritePkt)) == false) )
                {
                    synchronized(stateLock) {
                        checkState();
                    }

                    if (isBusy()) {
                        return Operation.PROCESS_PACKETS_REMAINING;
                    } else {
                        return Operation.PROCESS_PACKETS_COMPLETE;
                    }
                }
                if (!validJMSPkt) return Operation.PROCESS_PACKETS_REMAINING;
                inJMSWrite = true;

                // convert to old packet type if necessary
                if (convertPkt != null) {
                    // NOTE : if the c bit is set and this
                    // is an old packet, convert needs to 
                    // restart the resume flow

                    // LKS - XXX
                    convertPkt.handleWritePacket(waitingWritePkt); 
                }


                // check for connection flow control
                sent_count ++;
                boolean aboutToWaitForRF = flowCount != 0 && 
                                   sent_count >= flowCount;    

                if (aboutToWaitForRF) {
                    sent_count = 0;
                    waitingWritePkt.setFlowPaused(aboutToWaitForRF);
                    haltFlow();
                }
                if (IMQBasicConnection.DEBUG || DUMP_PACKET || OUT_DUMP_PACKET) {
                    int flag = (DUMP_PACKET || OUT_DUMP_PACKET) ? Logger.INFO
                            : Logger.DEBUGHIGH;
    
                    logger.log(flag, 
                          "\n------------------------------"
                         +"\nSending JMS Packet -[block = "+BLOCKING 
                         + ",nio = "+!STREAMS+"] " + this + "  Dumping"
                         + "\n" + waitingWritePkt.dumpPacketString("<<<<****")
                         + "\n------------------------------");
                }

            }

    
            if (inJMSWrite) { 
                inJMSWrite = !writeOutPacket(waitingWritePkt);
            }
            lockCritical = false;

            if (inJMSWrite) { // we were interrupted
                return Operation.PROCESS_WRITE_INCOMPLETE;
            }
            msgsToConsumer ++;

            if (waitingWritePkt.getPacketType() < PacketType.LAST)
                   pktsOut[waitingWritePkt.getPacketType()] ++;
   
            if (Globals.getConnectionManager().PING_ENABLED) {
                updateAccessTime(false);
            }
            if (METRICS_ON) {
                countOutPacket(waitingWritePkt);
            }

            if (isBusy()) {
                return Operation.PROCESS_PACKETS_REMAINING;
            } else {
                return Operation.PROCESS_PACKETS_COMPLETE;
            }
        } catch (OutOfMemoryError err) {
            Globals.handleGlobalError(err,
                Globals.getBrokerResources().getKString(
                BrokerResources.M_LOW_MEMORY_WRITE));
        } catch (IOException ex) {
             // connection is gone
             logger.log(logger.DEBUGMED, "closed connection " + this, ex);
             inJMSWrite = false;
             destroyConnection(false, GoodbyeReason.CLIENT_CLOSED, 
                Globals.getBrokerResources().getKString(
                    BrokerResources.M_CONNECTION_CLOSE));
             throw ex; 
        } finally {

             inWriteProcess = false;

             synchronized (flushCtrlLock) { 
                 if (flushCtrl) {
                      if ((ctrlpkt == null && control.isEmpty()) 
                                  || !isValid()) {
                          if (IMQBasicConnection.DEBUG) {
                              logger.log(Logger.DEBUG,
                                  "Done flushing control messages on " 
                                  + this);
                          }
                          flushCtrl = false;
                          flushCtrlLock.notifyAll();
                          if (os != null)
                              os.flush();
                     }
                 }
             }
             if (flush) {
                 synchronized (flushLock) { 
                      if (!isBusy() || !isValid()) {
                          if (IMQBasicConnection.DEBUG) {
                              logger.log(Logger.DEBUG,
                                     "Done flushing control messages on " 
                                      + this);
                          }
                          flush = false;
                          flushLock.notifyAll();
                          os.flush();
                     }
                 }
             }
             if (!isValid()) {
                 synchronized(this) {
                     waitingWritePkt = clearWritePacket(waitingWritePkt);
                 }
             }
        } 

        assert false : " should never happen";

        if (isBusy()) {
            return Operation.PROCESS_PACKETS_REMAINING;
        } else  {
            return Operation.PROCESS_PACKETS_COMPLETE;
        }
    }

    protected void checkState() {
        assert Thread.holdsLock(stateLock);

        synchronized(control) {
            synchronized(busySessions) {
                boolean is_running = isValid() && (
                  (inCtrlWrite || (!paused && !control.isEmpty()))
                            ||
                  (inJMSWrite || (runningMsgs && !waitingForResumeFlow
                       && !busySessions.isEmpty())));

               if (!isValid() || busy != is_running) {
                    busy = is_running;
                    stateLock.notifyAll();
                    if (ninfo != null) {
                        ninfo.setReadyToWrite(this, busy);
                    }
                }
            } 
        } 
                 
    }


    private boolean waitForWork(long time) {
        synchronized (stateLock) {
            if (isValid() && !busy) {
                try {
                    if (time == 0) {
                        stateLock.wait();
                    } else {
                        stateLock.wait(time);
                    }
                } catch (InterruptedException ex) {
                    assert false;
                    logger.logStack(Logger.INFO,"Internal error, "
                      + "got interrupted exception", ex);
                }
            }
            return busy;
        }
    }

    /**
     * Send a reply to pkt with the given status.
     */
    private void sendReply(Packet pkt, int status) {
        if (pkt.getSendAcknowledge()) {
            Packet reply = new Packet(useDirectBuffers());
            reply.setPacketType(PacketType.SEND_REPLY);
            reply.setConsumerID(pkt.getConsumerID());
            Hashtable hash = new Hashtable();
            hash.put("JMQStatus", new Integer(status));
            reply.setProperties(hash);
            sendControlMessage(reply);
        }
    }

// ---------------------------------------
//     Abstract Connection methods
// ---------------------------------------

    public void cleanupMemory(boolean persist) {
        // does nothing right now
    }

    protected void sayGoodbye(int reason, String reasonstr) {
        sayGoodbye(false, reason, reasonstr);
    }
    protected void sayGoodbye(boolean force, int reason, String reasonStr) {
        Packet goodbye_pkt = new Packet(useDirectBuffers());
        goodbye_pkt.setPacketType(PacketType.GOODBYE);
        Hashtable hash = new Hashtable();
        hash.put("JMQExit", Boolean.valueOf(force));
        hash.put("JMQGoodbyeReason", new Integer(reason));
        hash.put("JMQGoodbyeReasonString", reasonStr);
        goodbye_pkt.setProperties(hash);
        sendControlMessage(goodbye_pkt);
    }

    protected void checkConnection() {
        boolean sendAck = false;
        if (enablePingReply && closeInterval > 0 && getClientProtocolVersion() >= PacketType.VERSION364 ) {
            sendAck = true;
            
            // see if we need to kill the connection
            // get access time
            long access = getLastResponseTime();

            // is delta greater than ping_interval * closeInterval
            long delta = System.currentTimeMillis() - access;
            long interval = ConnectionManager.pingTimeout * (closeInterval+1);

            // if is, kill the connection
            if (delta >= interval) {
                logger.log(Logger.INFO, BrokerResources.W_UNRESPONSIVE_CONNECTION,
                   String.valueOf(this.getConnectionUID().longValue()), 
                   String.valueOf(delta/1000));
                    
                destroyConnection(false,GoodbyeReason.ADMIN_KILLED_CON,
                    "Connection unresponsive");
            }

        }

        Packet flush_pkt = new Packet(useDirectBuffers());
        flush_pkt.setPacketType(PacketType.PING);
        if (sendAck) {
            flush_pkt.setSendAcknowledge(true);
        }
        sendControlMessage(flush_pkt);
    }

    protected void flushConnection(long timeout) {
        flushControl(timeout);
    }

    public void flowPaused(long size) {
        if (Globals.getMemManager() != null)
            Globals.getMemManager().notifyWhenAvailable(this, size);
    }

    public void resumeMemory(int cnt, long memory, long max) {
        sendResume(cnt, memory, max, false);
    }

    public void updateMemory(int cnt, long memory, long max) {
        sendResume(cnt, memory, max, true);
    }

    protected void sendResume(int cnt, long memory, 
          long max, boolean priority) 
    {
        if (packetVersion < Packet.VERSION1)
            return; // older protocol cant handle resume

        Packet resume_pkt = new Packet(useDirectBuffers());
        resume_pkt.setPacketType(PacketType.RESUME_FLOW);
        Hashtable hash = new Hashtable();
        if (Globals.getMemManager() != null) {
            hash.put("JMQSize", new Integer(Globals.getMemManager().getJMQSize()));
            hash.put("JMQBytes", new Long(Globals.getMemManager().getJMQBytes()));
            hash.put("JMQMaxMsgBytes", new Long(
                     Globals.getMemManager().getJMQMaxMsgBytes()));
        }
        resume_pkt.setProperties(hash);

        sendControlMessage(resume_pkt, priority);
    }


    /**
     * called when either the session or the
     * control message is busy 
     */
    public void eventOccured(EventType type,  Reason r,
            Object target, Object oldval, Object newval, 
            Object userdata) 
    {

        // LKS - at this point, we are in a write lock
        // only one person can change the values
        // at a time

        synchronized (stateLock) {
            if (type == EventType.EMPTY) {
    
                // this can only be from the control queue
                assert target == control;
                assert newval instanceof Boolean;
                assert newval != null;
    
            } else if (type == 
                    EventType.BUSY_STATE_CHANGED) {
                assert target instanceof Session;
                assert newval instanceof Boolean;
                assert newval != null;
    
                Session s = (Session) target;
    
                synchronized(busySessions) {
                    synchronized (s.getBusyLock()) {
                        if (s.isBusy()) {
                            busySessions.add(s);
                        }
                    }
                }
                
            }
            checkState();
        }
    }

    private boolean fillNextPacket(Packet p) 
    {
        Session s = null;
        
        synchronized(busySessions) {
           Iterator itr = busySessions.iterator();
           while (itr.hasNext()) {
               s = (Session)itr.next();
               itr.remove();
               if (s == null) 
                   continue;
               synchronized (s.getBusyLock()) {
                   if (s.isBusy()) {
                       busySessions.add(s);
                       break;
                   }
               }
               
           }
        }

        if (s == null) return false;

        return s.fillNextPacket(p) != null;
    }
}



