Connection.java

/*
 * Copyright (c) 1999, 2020, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */

package com.sun.jndi.ldap;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.Socket;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.Arrays;

import javax.naming.CommunicationException;
import javax.naming.InterruptedNamingException;
import javax.naming.NamingException;
import javax.naming.ServiceUnavailableException;
import javax.naming.ldap.Control;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSocket;

/**
  * A thread that creates a connection to an LDAP server.
  * After the connection, the thread reads from the connection.
  * A caller can invoke methods on the instance to read LDAP responses
  * and to send LDAP requests.
  * <p>
  * There is a one-to-one correspondence between an LdapClient and
  * a Connection. Access to Connection and its methods is only via
  * LdapClient with two exceptions: SASL authentication and StartTLS.
  * SASL needs to access Connection's socket IO streams (in order to do encryption
  * of the security layer). StartTLS needs to do replace IO streams
  * and close the IO  streams on nonfatal close. The code for SASL
  * authentication can be treated as being the same as from LdapClient
  * because the SASL code is only ever called from LdapClient, from
  * inside LdapClient's synchronized authenticate() method. StartTLS is called
  * directly by the application but should only occur when the underlying
  * connection is quiet.
  * <p>
  * In terms of synchronization, worry about data structures
  * used by the Connection thread because that usage might contend
  * with calls by the main threads (i.e., those that call LdapClient).
  * Main threads need to worry about contention with each other.
  * Fields that Connection thread uses:
  *     inStream - synced access and update; initialized in constructor;
  *           referenced outside class unsync'ed (by LdapSasl) only
  *           when connection is quiet
  *     traceFile, traceTagIn, traceTagOut - no sync; debugging only
  *     parent - no sync; initialized in constructor; no updates
  *     pendingRequests - sync
  *     pauseLock - per-instance lock;
  *     paused - sync via pauseLock (pauseReader())
  * Members used by main threads (LdapClient):
  *     host, port - unsync; read-only access for StartTLS and debug messages
  *     setBound(), setV3() - no sync; called only by LdapClient.authenticate(),
  *             which is a sync method called only when connection is "quiet"
  *     getMsgId() - sync
  *     writeRequest(), removeRequest(),findRequest(), abandonOutstandingReqs() -
  *             access to shared pendingRequests is sync
  *     writeRequest(),  abandonRequest(), ldapUnbind() - access to outStream sync
  *     cleanup() - sync
  *     readReply() - access to sock sync
  *     unpauseReader() - (indirectly via writeRequest) sync on pauseLock
  * Members used by SASL auth (main thread):
  *     inStream, outStream - no sync; used to construct new stream; accessed
  *             only when conn is "quiet" and not shared
  *     replaceStreams() - sync method
  * Members used by StartTLS:
  *     inStream, outStream - no sync; used to record the existing streams;
  *             accessed only when conn is "quiet" and not shared
  *     replaceStreams() - sync method
  * <p>
  * Handles anonymous, simple, and SASL bind for v3; anonymous and simple
  * for v2.
  * %%% made public for access by LdapSasl %%%
  *
  * @author Vincent Ryan
  * @author Rosanna Lee
  * @author Jagane Sundar
  */
public final class Connection implements Runnable {

    private static final boolean debug = false;
    private static final int dump = 0; // > 0 r, > 1 rw


    final private Thread worker;    // Initialized in constructor

    private boolean v3 = true;       // Set in setV3()

    final public String host;  // used by LdapClient for generating exception messages
                         // used by StartTlsResponse when creating an SSL socket
    final public int port;     // used by LdapClient for generating exception messages
                         // used by StartTlsResponse when creating an SSL socket

    private boolean bound = false;   // Set in setBound()

    // All three are initialized in constructor and read-only afterwards
    private OutputStream traceFile = null;
    private String traceTagIn = null;
    private String traceTagOut = null;

    // Initialized in constructor; read and used externally (LdapSasl);
    // Updated in replaceStreams() during "quiet", unshared, period
    public InputStream inStream;   // must be public; used by LdapSasl

    // Initialized in constructor; read and used externally (LdapSasl);
    // Updated in replaceOutputStream() during "quiet", unshared, period
    public OutputStream outStream; // must be public; used by LdapSasl

    // Initialized in constructor; read and used externally (TLS) to
    // get new IO streams; closed during cleanup
    public Socket sock;            // for TLS

    // For processing "disconnect" unsolicited notification
    // Initialized in constructor
    final private LdapClient parent;

    // Incremented and returned in sync getMsgId()
    private int outMsgId = 0;

    //
    // The list of ldapRequests pending on this binding
    //
    // Accessed only within sync methods
    private LdapRequest pendingRequests = null;

    volatile IOException closureReason = null;
    volatile boolean useable = true;  // is Connection still useable

    int readTimeout;
    int connectTimeout;

    // Is connection upgraded to SSL via STARTTLS extended operation
    private volatile boolean isUpgradedToStartTls;

    // Lock to maintain isUpgradedToStartTls state
    final Object startTlsLock = new Object();

    private static final boolean IS_HOSTNAME_VERIFICATION_DISABLED
            = hostnameVerificationDisabledValue();

    private static boolean hostnameVerificationDisabledValue() {
        PrivilegedAction<String> act = () -> System.getProperty(
                "com.sun.jndi.ldap.object.disableEndpointIdentification");
        String prop = AccessController.doPrivileged(act);
        if (prop == null) {
            return false;
        }
        return prop.isEmpty() ? true : Boolean.parseBoolean(prop);
    }
    // true means v3; false means v2
    // Called in LdapClient.authenticate() (which is synchronized)
    // when connection is "quiet" and not shared; no need to synchronize
    void setV3(boolean v) {
        v3 = v;
    }

    // A BIND request has been successfully made on this connection
    // When cleaning up, remember to do an UNBIND
    // Called in LdapClient.authenticate() (which is synchronized)
    // when connection is "quiet" and not shared; no need to synchronize
    void setBound() {
        bound = true;
    }

    ////////////////////////////////////////////////////////////////////////////
    //
    // Create an LDAP Binding object and bind to a particular server
    //
    ////////////////////////////////////////////////////////////////////////////

    Connection(LdapClient parent, String host, int port, String socketFactory,
        int connectTimeout, int readTimeout, OutputStream trace) throws NamingException {

        this.host = host;
        this.port = port;
        this.parent = parent;
        this.readTimeout = readTimeout;
        this.connectTimeout = connectTimeout;

        if (trace != null) {
            traceFile = trace;
            traceTagIn = "<- " + host + ":" + port + "\n\n";
            traceTagOut = "-> " + host + ":" + port + "\n\n";
        }

        //
        // Connect to server
        //
        try {
            sock = createSocket(host, port, socketFactory, connectTimeout);

            if (debug) {
                System.err.println("Connection: opening socket: " + host + "," + port);
            }

            inStream = new BufferedInputStream(sock.getInputStream());
            outStream = new BufferedOutputStream(sock.getOutputStream());

        } catch (InvocationTargetException e) {
            Throwable realException = e.getTargetException();
            // realException.printStackTrace();

            CommunicationException ce =
                new CommunicationException(host + ":" + port);
            ce.setRootCause(realException);
            throw ce;
        } catch (Exception e) {
            // Class.forName() seems to do more error checking
            // and will throw IllegalArgumentException and such.
            // That's why we need to have a catch all here and
            // ignore generic exceptions.
            // Also catches all IO errors generated by socket creation.
            CommunicationException ce =
                new CommunicationException(host + ":" + port);
            ce.setRootCause(e);
            throw ce;
        }

        worker = Obj.helper.createThread(this);
        worker.setDaemon(true);
        worker.start();
    }

    /*
     * Create an InetSocketAddress using the specified hostname and port number.
     */
    private Object createInetSocketAddress(String host, int port)
            throws NoSuchMethodException {

        try {
            Class<?> inetSocketAddressClass =
                Class.forName("java.net.InetSocketAddress");

            Constructor<?> inetSocketAddressCons =
                inetSocketAddressClass.getConstructor(new Class<?>[]{
                String.class, int.class});

            return inetSocketAddressCons.newInstance(new Object[]{
                host, new Integer(port)});

        } catch (ClassNotFoundException |
                 InstantiationException |
                 InvocationTargetException |
                 IllegalAccessException e) {
            throw new NoSuchMethodException();

        }
    }

    /*
     * Create a Socket object using the specified socket factory and time limit.
     *
     * If a timeout is supplied and unconnected sockets are supported then
     * an unconnected socket is created and the timeout is applied when
     * connecting the socket. If a timeout is supplied but unconnected sockets
     * are not supported then the timeout is ignored and a connected socket
     * is created.
     */
    private Socket createSocket(String host, int port, String socketFactory,
            int connectTimeout) throws Exception {

        Socket socket = null;

        if (socketFactory != null) {

            // create the factory

            Class<?> socketFactoryClass = Obj.helper.loadClass(socketFactory);
            Method getDefault =
                socketFactoryClass.getMethod("getDefault", new Class<?>[]{});
            Object factory = getDefault.invoke(null, new Object[]{});

            // create the socket

            Method createSocket = null;

            if (connectTimeout > 0) {

                try {
                    createSocket = socketFactoryClass.getMethod("createSocket",
                        new Class<?>[]{});

                    Method connect = Socket.class.getMethod("connect",
                        new Class<?>[]{Class.forName("java.net.SocketAddress"),
                        int.class});
                    Object endpoint = createInetSocketAddress(host, port);

                    // unconnected socket
                    socket =
                        (Socket)createSocket.invoke(factory, new Object[]{});

                    if (debug) {
                        System.err.println("Connection: creating socket with " +
                            "a timeout using supplied socket factory");
                    }

                    // connected socket
                    connect.invoke(socket, new Object[]{
                        endpoint, new Integer(connectTimeout)});

                } catch (NoSuchMethodException e) {
                    // continue (but ignore connectTimeout)
                }
            }

            if (socket == null) {
                createSocket = socketFactoryClass.getMethod("createSocket",
                    new Class<?>[]{String.class, int.class});

                if (debug) {
                    System.err.println("Connection: creating socket using " +
                        "supplied socket factory");
                }
                // connected socket
                socket = (Socket) createSocket.invoke(factory,
                    new Object[]{host, new Integer(port)});
            }
        } else {

            if (connectTimeout > 0) {

                try {
                    Constructor<Socket> socketCons =
                        Socket.class.getConstructor(new Class<?>[]{});

                    Method connect = Socket.class.getMethod("connect",
                        new Class<?>[]{Class.forName("java.net.SocketAddress"),
                        int.class});
                    Object endpoint = createInetSocketAddress(host, port);

                    socket = socketCons.newInstance(new Object[]{});

                    if (debug) {
                        System.err.println("Connection: creating socket with " +
                            "a timeout");
                    }
                    connect.invoke(socket, new Object[]{
                        endpoint, new Integer(connectTimeout)});

                } catch (NoSuchMethodException e) {
                    // continue (but ignore connectTimeout)
                }
            }

            if (socket == null) {
                if (debug) {
                    System.err.println("Connection: creating socket");
                }
                // connected socket
                socket = new Socket(host, port);
            }
        }

        // For LDAP connect timeouts on LDAP over SSL connections must treat
        // the SSL handshake following socket connection as part of the timeout.
        // So explicitly set a socket read timeout, trigger the SSL handshake,
        // then reset the timeout.
        if (socket instanceof SSLSocket) {
            SSLSocket sslSocket = (SSLSocket) socket;
            if (!IS_HOSTNAME_VERIFICATION_DISABLED) {
                SSLParameters param = sslSocket.getSSLParameters();
                param.setEndpointIdentificationAlgorithm("LDAPS");
                sslSocket.setSSLParameters(param);
            }
            if (connectTimeout > 0) {
                int socketTimeout = sslSocket.getSoTimeout();
                sslSocket.setSoTimeout(connectTimeout); // reuse full timeout value
                sslSocket.startHandshake();
                sslSocket.setSoTimeout(socketTimeout);
            }
        }
        return socket;
    }

    ////////////////////////////////////////////////////////////////////////////
    //
    // Methods to IO to the LDAP server
    //
    ////////////////////////////////////////////////////////////////////////////

    synchronized int getMsgId() {
        return ++outMsgId;
    }

    LdapRequest writeRequest(BerEncoder ber, int msgId) throws IOException {
        return writeRequest(ber, msgId, false /* pauseAfterReceipt */, -1);
    }

    LdapRequest writeRequest(BerEncoder ber, int msgId,
        boolean pauseAfterReceipt) throws IOException {
        return writeRequest(ber, msgId, pauseAfterReceipt, -1);
    }

    LdapRequest writeRequest(BerEncoder ber, int msgId,
        boolean pauseAfterReceipt, int replyQueueCapacity) throws IOException {

        LdapRequest req =
            new LdapRequest(msgId, pauseAfterReceipt, replyQueueCapacity);
        addRequest(req);

        if (traceFile != null) {
            Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0, ber.getDataLen());
        }


        // unpause reader so that it can get response
        // NOTE: Must do this before writing request, otherwise might
        // create a race condition where the writer unblocks its own response
        unpauseReader();

        if (debug) {
            System.err.println("Writing request to: " + outStream);
        }

        try {
            synchronized (this) {
                outStream.write(ber.getBuf(), 0, ber.getDataLen());
                outStream.flush();
            }
        } catch (IOException e) {
            cleanup(null, true);
            throw (closureReason = e); // rethrow
        }

        return req;
    }

    /**
     * Reads a reply; waits until one is ready.
     */
    BerDecoder readReply(LdapRequest ldr) throws IOException, NamingException {
        BerDecoder rber;

        NamingException namingException = null;
        try {
            // if no timeout is set so we wait infinitely until
            // a response is received OR until the connection is closed or cancelled
            // http://docs.oracle.com/javase/8/docs/technotes/guides/jndi/jndi-ldap.html#PROP
            rber = ldr.getReplyBer(readTimeout);
        } catch (InterruptedException ex) {
            throw new InterruptedNamingException(
                "Interrupted during LDAP operation");
        } catch (CommunicationException ce) {
            // Re-throw
            throw ce;
        } catch (NamingException ne) {
            // Connection is timed out OR closed/cancelled
            namingException = ne;
            rber = null;
        }

        if (rber == null) {
            abandonRequest(ldr, null);
        }
        // namingException can be not null in the following cases:
        //  a) The response is timed-out
        //  b) LDAP request connection has been closed or cancelled
        // The exception message is initialized in LdapRequest::getReplyBer
        if (namingException != null) {
            // Re-throw NamingException after all cleanups are done
            throw namingException;
        }
        return rber;
    }

    ////////////////////////////////////////////////////////////////////////////
    //
    // Methods to add, find, delete, and abandon requests made to server
    //
    ////////////////////////////////////////////////////////////////////////////

    private synchronized void addRequest(LdapRequest ldapRequest) {

        LdapRequest ldr = pendingRequests;
        if (ldr == null) {
            pendingRequests = ldapRequest;
            ldapRequest.next = null;
        } else {
            ldapRequest.next = pendingRequests;
            pendingRequests = ldapRequest;
        }
    }

    synchronized LdapRequest findRequest(int msgId) {

        LdapRequest ldr = pendingRequests;
        while (ldr != null) {
            if (ldr.msgId == msgId) {
                return ldr;
            }
            ldr = ldr.next;
        }
        return null;

    }

    synchronized void removeRequest(LdapRequest req) {
        LdapRequest ldr = pendingRequests;
        LdapRequest ldrprev = null;

        while (ldr != null) {
            if (ldr == req) {
                ldr.cancel();

                if (ldrprev != null) {
                    ldrprev.next = ldr.next;
                } else {
                    pendingRequests = ldr.next;
                }
                ldr.next = null;
            }
            ldrprev = ldr;
            ldr = ldr.next;
        }
    }

    void abandonRequest(LdapRequest ldr, Control[] reqCtls) {
        // Remove from queue
        removeRequest(ldr);

        BerEncoder ber = new BerEncoder(256);
        int abandonMsgId = getMsgId();

        //
        // build the abandon request.
        //
        try {
            ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
                ber.encodeInt(abandonMsgId);
                ber.encodeInt(ldr.msgId, LdapClient.LDAP_REQ_ABANDON);

                if (v3) {
                    LdapClient.encodeControls(ber, reqCtls);
                }
            ber.endSeq();

            if (traceFile != null) {
                Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(), 0,
                    ber.getDataLen());
            }

            synchronized (this) {
                outStream.write(ber.getBuf(), 0, ber.getDataLen());
                outStream.flush();
            }

        } catch (IOException ex) {
            //System.err.println("ldap.abandon: " + ex);
        }

        // Don't expect any response for the abandon request.
    }

    synchronized void abandonOutstandingReqs(Control[] reqCtls) {
        LdapRequest ldr = pendingRequests;

        while (ldr != null) {
            abandonRequest(ldr, reqCtls);
            pendingRequests = ldr = ldr.next;
        }
    }

    ////////////////////////////////////////////////////////////////////////////
    //
    // Methods to unbind from server and clear up resources when object is
    // destroyed.
    //
    ////////////////////////////////////////////////////////////////////////////

    private void ldapUnbind(Control[] reqCtls) {

        BerEncoder ber = new BerEncoder(256);
        int unbindMsgId = getMsgId();

        //
        // build the unbind request.
        //

        try {

            ber.beginSeq(Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR);
                ber.encodeInt(unbindMsgId);
                // IMPLICIT TAGS
                ber.encodeByte(LdapClient.LDAP_REQ_UNBIND);
                ber.encodeByte(0);

                if (v3) {
                    LdapClient.encodeControls(ber, reqCtls);
                }
            ber.endSeq();

            if (traceFile != null) {
                Ber.dumpBER(traceFile, traceTagOut, ber.getBuf(),
                    0, ber.getDataLen());
            }

            synchronized (this) {
                outStream.write(ber.getBuf(), 0, ber.getDataLen());
                outStream.flush();
            }

        } catch (IOException ex) {
            //System.err.println("ldap.unbind: " + ex);
        }

        // Don't expect any response for the unbind request.
    }

    /**
     * @param reqCtls Possibly null request controls that accompanies the
     *    abandon and unbind LDAP request.
     * @param notifyParent true means to call parent LdapClient back, notifying
     *    it that the connection has been closed; false means not to notify
     *    parent. If LdapClient invokes cleanup(), notifyParent should be set to
     *    false because LdapClient already knows that it is closing
     *    the connection. If Connection invokes cleanup(), notifyParent should be
     *    set to true because LdapClient needs to know about the closure.
     */
    void cleanup(Control[] reqCtls, boolean notifyParent) {
        boolean nparent = false;

        synchronized (this) {
            useable = false;

            if (sock != null) {
                if (debug) {
                    System.err.println("Connection: closing socket: " + host + "," + port);
                }
                try {
                    if (!notifyParent) {
                        abandonOutstandingReqs(reqCtls);
                    }
                    if (bound) {
                        ldapUnbind(reqCtls);
                    }
                } finally {
                    try {
                        outStream.flush();
                        sock.close();
                        unpauseReader();
                    } catch (IOException ie) {
                        if (debug)
                            System.err.println("Connection: problem closing socket: " + ie);
                    }
                    if (!notifyParent) {
                        LdapRequest ldr = pendingRequests;
                        while (ldr != null) {
                            ldr.cancel();
                            ldr = ldr.next;
                        }
                    }
                    sock = null;
                }
                nparent = notifyParent;
            }
            if (nparent) {
                LdapRequest ldr = pendingRequests;
                while (ldr != null) {
                    ldr.close();
                        ldr = ldr.next;
                    }
                }
            }
        if (nparent) {
            parent.processConnectionClosure();
        }
    }


    // Assume everything is "quiet"
    // "synchronize" might lead to deadlock so don't synchronize method
    // Use streamLock instead for synchronizing update to stream

    synchronized public void replaceStreams(InputStream newIn, OutputStream newOut) {
        if (debug) {
            System.err.println("Replacing " + inStream + " with: " + newIn);
            System.err.println("Replacing " + outStream + " with: " + newOut);
        }

        inStream = newIn;

        // Cleanup old stream
        try {
            outStream.flush();
        } catch (IOException ie) {
            if (debug)
                System.err.println("Connection: cannot flush outstream: " + ie);
        }

        // Replace stream
        outStream = newOut;
    }

    /*
     * Replace streams and set isUpdradedToStartTls flag to the provided value
     */
    synchronized public void replaceStreams(InputStream newIn, OutputStream newOut, boolean isStartTls) {
        synchronized (startTlsLock) {
            replaceStreams(newIn, newOut);
            isUpgradedToStartTls = isStartTls;
        }
    }

    /*
     * Returns true if connection was upgraded to SSL with STARTTLS extended operation
     */
    public boolean isUpgradedToStartTls() {
        return isUpgradedToStartTls;
    }

    /**
     * Used by Connection thread to read inStream into a local variable.
     * This ensures that there is no contention between the main thread
     * and the Connection thread when the main thread updates inStream.
     */
    synchronized private InputStream getInputStream() {
        return inStream;
    }


    ////////////////////////////////////////////////////////////////////////////
    //
    // Code for pausing/unpausing the reader thread ('worker')
    //
    ////////////////////////////////////////////////////////////////////////////

    /*
     * The main idea is to mark requests that need the reader thread to
     * pause after getting the response. When the reader thread gets the response,
     * it waits on a lock instead of returning to the read(). The next time a
     * request is sent, the reader is automatically unblocked if necessary.
     * Note that the reader must be unblocked BEFORE the request is sent.
     * Otherwise, there is a race condition where the request is sent and
     * the reader thread might read the response and be unblocked
     * by writeRequest().
     *
     * This pause gives the main thread (StartTLS or SASL) an opportunity to
     * update the reader's state (e.g., its streams) if necessary.
     * The assumption is that the connection will remain quiet during this pause
     * (i.e., no intervening requests being sent).
     *<p>
     * For dealing with StartTLS close,
     * when the read() exits either due to EOF or an exception,
     * the reader thread checks whether there is a new stream to read from.
     * If so, then it reattempts the read. Otherwise, the EOF or exception
     * is processed and the reader thread terminates.
     * In a StartTLS close, the client first replaces the SSL IO streams with
     * plain ones and then closes the SSL socket.
     * If the reader thread attempts to read, or was reading, from
     * the SSL socket (that is, it got to the read BEFORE replaceStreams()),
     * the SSL socket close will cause the reader thread to
     * get an EOF/exception and reexamine the input stream.
     * If the reader thread sees a new stream, it reattempts the read.
     * If the underlying socket is still alive, then the new read will succeed.
     * If the underlying socket has been closed also, then the new read will
     * fail and the reader thread exits.
     * If the reader thread attempts to read, or was reading, from the plain
     * socket (that is, it got to the read AFTER replaceStreams()), the
     * SSL socket close will have no effect on the reader thread.
     *
     * The check for new stream is made only
     * in the first attempt at reading a BER buffer; the reader should
     * never be in midst of reading a buffer when a nonfatal close occurs.
     * If this occurs, then the connection is in an inconsistent state and
     * the safest thing to do is to shut it down.
     */

    private final Object pauseLock = new Object();  // lock for reader to wait on while paused
    private boolean paused = false;           // paused state of reader

    /*
     * Unpauses reader thread if it was paused
     */
    private void unpauseReader() throws IOException {
        synchronized (pauseLock) {
            if (paused) {
                if (debug) {
                    System.err.println("Unpausing reader; read from: " +
                                        inStream);
                }
                paused = false;
                pauseLock.notify();
            }
        }
    }

     /*
     * Pauses reader so that it stops reading from the input stream.
     * Reader blocks on pauseLock instead of read().
     * MUST be called from within synchronized (pauseLock) clause.
     */
    private void pauseReader() throws IOException {
        if (debug) {
            System.err.println("Pausing reader;  was reading from: " +
                                inStream);
        }
        paused = true;
        try {
            while (paused) {
                pauseLock.wait(); // notified by unpauseReader
            }
        } catch (InterruptedException e) {
            throw new InterruptedIOException(
                    "Pause/unpause reader has problems.");
        }
    }


    ////////////////////////////////////////////////////////////////////////////
    //
    // The LDAP Binding thread. It does the mux/demux of multiple requests
    // on the same TCP connection.
    //
    ////////////////////////////////////////////////////////////////////////////


    public void run() {
        byte inbuf[];   // Buffer for reading incoming bytes
        int inMsgId;    // Message id of incoming response
        int bytesread;  // Number of bytes in inbuf
        int br;         // Temp; number of bytes read from stream
        int offset;     // Offset of where to store bytes in inbuf
        int seqlen;     // Length of ASN sequence
        int seqlenlen;  // Number of sequence length bytes
        boolean eos;    // End of stream
        BerDecoder retBer;    // Decoder for ASN.1 BER data from inbuf
        InputStream in = null;

        try {
            while (true) {
                try {
                    // type and length (at most 128 octets for long form)
                    inbuf = new byte[129];

                    offset = 0;
                    seqlen = 0;
                    seqlenlen = 0;

                    in = getInputStream();

                    // check that it is the beginning of a sequence
                    bytesread = in.read(inbuf, offset, 1);
                    if (bytesread < 0) {
                        if (in != getInputStream()) {
                            continue;   // a new stream to try
                        } else {
                            break; // EOF
                        }
                    }

                    if (inbuf[offset++] != (Ber.ASN_SEQUENCE | Ber.ASN_CONSTRUCTOR))
                        continue;

                    // get length of sequence
                    bytesread = in.read(inbuf, offset, 1);
                    if (bytesread < 0)
                        break; // EOF
                    seqlen = inbuf[offset++];

                    // if high bit is on, length is encoded in the
                    // subsequent length bytes and the number of length bytes
                    // is equal to & 0x80 (i.e. length byte with high bit off).
                    if ((seqlen & 0x80) == 0x80) {
                        seqlenlen = seqlen & 0x7f;  // number of length bytes
                        // Check the length of length field, since seqlen is int
                        // the number of bytes can't be greater than 4
                        if (seqlenlen > 4) {
                            throw new IOException("Length coded with too many bytes: " + seqlenlen);
                        }

                        bytesread = 0;
                        eos = false;

                        // Read all length bytes
                        while (bytesread < seqlenlen) {
                            br = in.read(inbuf, offset+bytesread,
                                seqlenlen-bytesread);
                            if (br < 0) {
                                eos = true;
                                break; // EOF
                            }
                            bytesread += br;
                        }

                        // end-of-stream reached before length bytes are read
                        if (eos)
                            break;  // EOF

                        // Add contents of length bytes to determine length
                        seqlen = 0;
                        for( int i = 0; i < seqlenlen; i++) {
                            seqlen = (seqlen << 8) + (inbuf[offset+i] & 0xff);
                        }
                        offset += bytesread;
                    }

                    if (seqlenlen > bytesread) {
                        throw new IOException("Unexpected EOF while reading length");
                    }

                    if (seqlen < 0) {
                        throw new IOException("Length too big: " + (((long) seqlen) & 0xFFFFFFFFL));
                    }
                    // read in seqlen bytes
                    byte[] left = readFully(in, seqlen);
                    inbuf = Arrays.copyOf(inbuf, offset + left.length);
                    System.arraycopy(left, 0, inbuf, offset, left.length);
                    offset += left.length;

                    try {
                        retBer = new BerDecoder(inbuf, 0, offset);

                        if (traceFile != null) {
                            Ber.dumpBER(traceFile, traceTagIn, inbuf, 0, offset);
                        }

                        retBer.parseSeq(null);
                        inMsgId = retBer.parseInt();
                        retBer.reset(); // reset offset

                        boolean needPause = false;

                        if (inMsgId == 0) {
                            // Unsolicited Notification
                            parent.processUnsolicited(retBer);
                        } else {
                            LdapRequest ldr = findRequest(inMsgId);

                            if (ldr != null) {

                                /**
                                 * Grab pauseLock before making reply available
                                 * to ensure that reader goes into paused state
                                 * before writer can attempt to unpause reader
                                 */
                                synchronized (pauseLock) {
                                    needPause = ldr.addReplyBer(retBer);
                                    if (needPause) {
                                        /*
                                         * Go into paused state; release
                                         * pauseLock
                                         */
                                        pauseReader();
                                    }

                                    // else release pauseLock
                                }
                            } else {
                                // System.err.println("Cannot find" +
                                //              "LdapRequest for " + inMsgId);
                            }
                        }
                    } catch (Ber.DecodeException e) {
                        //System.err.println("Cannot parse Ber");
                    }
                } catch (IOException ie) {
                    if (debug) {
                        System.err.println("Connection: Inside Caught " + ie);
                        ie.printStackTrace();
                    }

                    if (in != getInputStream()) {
                        // A new stream to try
                        // Go to top of loop and continue
                    } else {
                        if (debug) {
                            System.err.println("Connection: rethrowing " + ie);
                        }
                        throw ie;  // rethrow exception
                    }
                }
            }

            if (debug) {
                System.err.println("Connection: end-of-stream detected: "
                    + in);
            }
        } catch (IOException ex) {
            if (debug) {
                System.err.println("Connection: Caught " + ex);
            }
            closureReason = ex;
        } finally {
            cleanup(null, true); // cleanup
        }
        if (debug) {
            System.err.println("Connection: Thread Exiting");
        }
    }

    private static byte[] readFully(InputStream is, int length)
        throws IOException
    {
        byte[] buf = new byte[Math.min(length, 8192)];
        int nread = 0;
        while (nread < length) {
            int bytesToRead;
            if (nread >= buf.length) {  // need to allocate a larger buffer
                bytesToRead = Math.min(length - nread, buf.length + 8192);
                if (buf.length < nread + bytesToRead) {
                    buf = Arrays.copyOf(buf, nread + bytesToRead);
                }
            } else {
                bytesToRead = buf.length - nread;
            }
            int count = is.read(buf, nread, bytesToRead);
            if (count < 0) {
                if (buf.length != nread)
                    buf = Arrays.copyOf(buf, nread);
                break;
            }
            nread += count;
        }
        return buf;
    }
}