# LowLevelSocketImpl.java

    /*
    ** Module   : LowLevelSocketImpl.java
    ** Abstract : Implementation of a client socket.
    **
    ** Copyright (c) 2013-2025, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --------------------------------Description-----------------------------------
    ** 001 CA  20130908 Created initial version.
    ** 002 HC  20131031 Changed resource ID data type to Long, see issue #2183.
    ** 003 OM  20131018 Updated to conform the new SessionListener API.
    ** 004 CA  20140805 ThinClient is explicitly imported, to isolate ChUI class usage outside of ChUI
    **                  packages.
    ** 005 OM  20151104 Fixed connectSSL() implementation. Added support for securing sockets used by
    *                   AXIS2 when accessing secured Web Services.
    ** 006 IAS 20191140 Additional logging statement added
    ** 007 EVL 20210426 Added configurable socket connect timeout value to fix unwamted delays from connect
    **                  attempts to invalid or missed socket based resources.
    ** 008 RFB 20210908 Added protocol parameter so that additional socket protocols like TLSv1.2 can be used.
    **                  Ref. #4366.
    ** 009 IAS 20211125 Added '-clientConnectTimeout' CONNECT option support
    **     CA  20220326 Fixed concurrency issues: the SocketListener thread and the main thread (where the socket
    **                  data is actually read) need to wait for eachother: the main thread needs to wait for the
    **                  SocketListener to be in the state where it waits for the read operation to start, and the
    **                  SocketListener thread needs to wait for the main thread to start and end the data read
    **                  operation.
    **     CA  20220329 Do not raise a READ-RESPONSE event if the client has terminated.  Do not attempt to
    **                  cleanup sockets if the WorkArea has not been created (as the client didn't use them).
    **     CA  20220515 memptr and library calls are supported on server-side, too.  Added TODO's for client-side
    **                  APIs which will need to call back to server-side, when memptr is used as argument.
    **     CA  20220918 Allow legacy client sockets to be ran on FWD server instead of FWD client.
    **                  Fixed a concurrency issue - look for 'isTerminated' on start/endRead and waitForNotify.
    **                  Also, ReadListener.setTerminated will notify all.
    **     CA  20220928 ReadListener was using 'waitForNotify(false);' (to wait for 'no read is active') twice
    **                  which lead to a race condition and a deadlock.
    **                  'waitingForRead' flag initial value must be true, as the ReadListener initial state is to
    **                  wait for a read operation.
    **     CA  20221206 Added support for '-servername' (server name indicator - SNI - for TLS connections).
    ** 010 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
    ** 011 CA  20230701 Unless the socket is closed, keep reading; thus, socket timeout exceptions are allowed.
    ** 012 CA  20240223 Raise proper 'unknown host' error in case of hostname resolution issues.
    ** 013 RFB 20240329 Add some logging when there is a handshake failure.
    **     CA  20240424 Improved handling of timeouts between bytes read in a chunk. Also fixed some minor comment
    **                  formats.
    ** 014 CA  20240502 'hasMoreDataworker' must not assume there is no more data, if the waiting has finished
    **                  because of a timeout.  Also, a SocketTimeoutException must be thrown only if SO_TIMEOUT is
    **                  not zero and the timeout has actually expired.
    ** 015 GBB 20240513 P2J_NOHOSTVERIFY removed, as custom params don't survive in the current version of the lib
    ** 016 CA  20240620 The context-local 'nohostverify' must be set and unset only when the SOAP request is executing,
    **                  when in SSL mode.
    ** 017 GBB 20240826 Moving to package osresource. Allowing access to methods and attributes from other packages.
    ** 018 CA  20250317 Fixed 'disconnect()' - call 'ReadListener.setTerminate(true)', to mark it terminated,
    **                  beside interrupting it.
    */
    /*
    ** This program is free software: you can redistribute it and/or modify
    ** it under the terms of the GNU Affero General Public License as
    ** published by the Free Software Foundation, either version 3 of the
    ** License, or (at your option) any later version.
    **
    ** This program 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 Affero General Public License for more details.
    **
    ** You may find a copy of the GNU Affero GPL version 3 at the following
    ** location: https://www.gnu.org/licenses/agpl-3.0.en.html
    **
    ** Additional terms under GNU Affero GPL version 3 section 7:
    **
    **   Under Section 7 of the GNU Affero GPL version 3, the following additional
    **   terms apply to the works covered under the License.  These additional terms
    **   are non-permissive additional terms allowed under Section 7 of the GNU
    **   Affero GPL version 3 and may not be removed by you.
    **
    **   0. Attribution Requirement.
    **
    **     You must preserve all legal notices or author attributions in the covered
    **     work or Appropriate Legal Notices displayed by works containing the covered
    **     work.  You may not remove from the covered work any author or developer
    **     credit already included within the covered work.
    **
    **   1. No License To Use Trademarks.
    **
    **     This license does not grant any license or rights to use the trademarks
    **     Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
    **     of Golden Code Development Corporation. You are not authorized to use the
    **     name Golden Code, FWD, or the names of any author or contributor, for
    **     publicity purposes without written authorization.
    **
    **   2. No Misrepresentation of Affiliation.
    **
    **     You may not represent yourself as Golden Code Development Corporation or FWD.
    **
    **     You may not represent yourself for publicity purposes as associated with
    **     Golden Code Development Corporation, FWD, or any author or contributor to
    **     the covered work, without written authorization.
    **
    **   3. No Misrepresentation of Source or Origin.
    **
    **     You may not represent the covered work as solely your work.  All modified
    **     versions of the covered work must be marked in a reasonable way to make it
    **     clear that the modified work is not originating from Golden Code Development
    **     Corporation or FWD.  All modified versions must contain the notices of
    **     attribution required in this license.
    */
    package com.goldencode.p2j.util.osresource;

    import java.io.*;
    import java.net.*;
    import java.net.Socket;
    import java.net.SocketImpl;
    import java.nio.channels.*;
    import java.security.*;
    import java.security.cert.*;
    import java.util.*;
    import java.util.concurrent.*;
    import java.util.logging.*;
    import javax.net.ssl.*;
    import com.goldencode.p2j.directory.*;
    import com.goldencode.p2j.main.*;
    import com.goldencode.p2j.net.*;
    import com.goldencode.p2j.net.Session;
    import com.goldencode.p2j.security.*;
    import com.goldencode.p2j.ui.*;
    import com.goldencode.p2j.ui.chui.ThinClient;
    import com.goldencode.p2j.util.*;
    import com.goldencode.p2j.util.ErrorManager;
    import com.goldencode.p2j.util.logging.*;
    import org.apache.commons.httpclient.params.*;
    import org.apache.commons.httpclient.protocol.*;

    /**
     * This class acts as a delegate for the server-side to the created java socket.
     */
    public class LowLevelSocketImpl
    implements LowLevelSocket
    {
       /** Logger */
       private static final CentralLogger LOG = CentralLogger.get(LowLevelSocketImpl.class);

       /** Token used to authenticate with the dispatcher when registering APIs. */
       private static Object modToken = null;

       /** The context-local data of this class. */
       private static ContextLocal<WorkArea> local = new ContextLocal<WorkArea>()
       {
          protected WorkArea initialValue()
          {
             return new WorkArea();
          }
       };

       /**
        * Create an instance and export its API to the network.
        *
        * @param   local
        *          <code>true</code> to register purely on a local basis (no remote/network access to these
        *          methods).
        */
       LowLevelSocketImpl(boolean local)
       {
          synchronized (LowLevelSocket.class)
          {
             modToken = RemoteObject.registerServer(LowLevelSocket.class, this, modToken, local);

             // initialize client side SSL by registering a custom SSL factory for https protocol
             ProtocolSocketFactory factory = new SSLProtocolSocketFactory()
             {
                @Override
                public Socket createSocket(String host,
                                           int port,
                                           InetAddress inetAddress,
                                           int localPort,
                                           HttpConnectionParams params)
                throws IOException,
                       IllegalArgumentException
                {
                   if (params == null)
                   {
                      IllegalArgumentException exc =
                            new IllegalArgumentException("Parameters may not be null");
                      if (LOG.isLoggable(Level.WARNING))
                      {
                         LOG.log(Level.WARNING, "Failed to create/configure SSL socket", exc);
                      }
                      throw exc;
                   }
                   boolean noHostVerify = LowLevelSocketImpl.local.get().nohostverify;
                   // TODO: The last parameter was added in the refactoring to allow for more granular connection
                   // options. This is for web-service:connect and is being addressed in #5662
                   return createSslSocket(host,
                                          port,
                                          null, // TODO: is serverName needed here?
                                          inetAddress,
                                          localPort,
                                          params.getConnectionTimeout(),
                                          noHostVerify,
                                          false,
                                          "SSL");
                }
             };
             Protocol.registerProtocol("https", new Protocol("https", factory, 443));
          }
       }

       /**
        * Sets value for the -nohostverify CONNECT arg.
        *
        * @param   noHostVerify
        *          The value for the -nohostverify CONNECT arg
        */
       public static void setNoHostVerify(boolean noHostVerify)
       {
          LowLevelSocketImpl.local.get().nohostverify = noHostVerify;
       }

       /**
        * Worker for creating a SSL socket and secure it according to specifications.
        * <p>
        * Documentation states that OpenEdge keeps the public keys (certificates) as individual files
        * in a per-installation store (directory {@code DLC/certs}). We use per-client JKS storage
        * facilities in a unified {@code trusted-cert.store} in client installation. Certificates need
        * to be imported with java {@code keytool}:
        * <pre>
        *    keytool -import -alias a_server -file a_server.cer -keystore trusted-cert.store
        * </pre>
        * The location of the store and password for access it can be passed-in at client startup as
        * bootstrap config value with the default value of {@code trusted-cert.store} in the current
        * directory.
        *
        * Apparently OpenEdge's {@code mkhashfile} utility uses the rehashing function for creating
        * (not so) unique ids for each imported certificate. A similar (if not identical) feature is
        * offered by OpenSSL:
        * <pre>
        *    openssl x509 -in a_server.pem -noout -hash
        * </pre>
        *
        * @param   host
        *          The host name/IP to connect to.
        * @param   port
        *          The port on the host.
        * @param   serverName
        *          The server name indicator (SNI) set via <code>-servername</code>, for TLS connections.
        * @param   inetAddress
        *          The local host name/IP to bind the socket to.
        * @param   localPort
        *          The port on the local machine.
        * @param   timeout
        *          Socket timeout (in milliseconds), or 0 if it never expires.
        * @param   nohostverify
        *          If {@code true}, disables host verification for connection using HTTPS.
        * @param   errManager
        *          Instruct how to handle encountered errors. If {@code true} ant exception is fed to
        *          P2J {@link ErrorManager}, otherwise let the caller handle it using normal java
        *          exceptions.
        * @param   protocol
        *          The standard name of the requested protocol as defined by the SSLContext class.
        *
        * @return  The newly build and secured socket.
        *
        * @throws  IOException
        *          If any problems are encountered in the process.
        */
       private static SSLSocket createSslSocket(String host,
                                                int port,
                                                String serverName,
                                                InetAddress inetAddress,
                                                int localPort,
                                                int timeout,
                                                boolean nohostverify,
                                                boolean errManager,
                                                String  protocol)
       throws IOException,
              SecurityException
       {
          WorkArea wa = local.get();

          // this is used only for displaying Progress errors. At this moment I don't know how to
          // extract the used certificate by java SSL when the handshake fails.
          String certId = "a4b3c2d1.0";

          SSLSocket socket = null;

          try
          {
             SSLSocketFactory factory = wa.factories.get(protocol);
             if (factory == null)
             {
                // hardcoded file-level password for keystore. Same for all stores.
                char keyStorePass[] = wa.trustStorePassword.toCharArray();

                // open key store, check the existence of the requested alias
                KeyStore ks = KeyStore.getInstance("JKS");
                ks.load(new FileInputStream(wa.trustStoreFilename), keyStorePass);

                // create a trusted certificates manager
                TrustManagerFactory tmf = TrustManagerFactory.getInstance("X509");
                tmf.init(ks);

                SSLContext sslCtx = SSLContext.getInstance(protocol); // "TLSv1"
                sslCtx.init(null, tmf.getTrustManagers(), null);

                factory = sslCtx.getSocketFactory();
                wa.factories.put(protocol, factory);
             }

             if (timeout == 0)
             {
                socket = (SSLSocket) factory.createSocket(host, port, inetAddress, localPort);
             }
             else
             {
                socket = (SSLSocket) factory.createSocket();
                socket.bind(new InetSocketAddress(inetAddress, localPort));
                socket.connect(new InetSocketAddress(host, port), timeout);
             }

             if (serverName != null)
             {
                SSLParameters sslParams = new SSLParameters();
                List<SNIServerName> serverNames = new ArrayList<>();
                serverNames.add(new SNIHostName(serverName));
                sslParams.setServerNames(serverNames);
                socket.setSSLParameters(sslParams);
             }
          }

          catch (UnknownHostException ex)
          {
             if (!errManager)
             {
                throw ex;
             }

             int errCodes[] = { 5482 };
             String errMessages[] = { "Unknown hostname " + host };
             ErrorManager.recordOrShowError(errCodes, errMessages, false, false);

             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.log(Level.WARNING, "Failed to create/configure SSL socket", ex);
             }
             return null;
          }

          catch (KeyStoreException      |
                 KeyManagementException |
                 CertificateException   |
                 IOException            |  // Keystore was tampered with, or password was incorrect
                 NoSuchAlgorithmException e)
          {
             if (!errManager)
             {
                if (e instanceof IOException)
                {
                   throw (IOException) e;
                }
                else
                {
                   throw new SecurityException(e);
                }
             }

             int errCodes[] = { 9318, 9407 };
             String errMessages[] =
             {
                "Secure Socket Layer (SSL) failure. error code 0:  Unknown SSL error",
                "Connection failure for host " + host + " port " + port + " transport TCP"
             };
             ErrorManager.recordOrShowError(errCodes, errMessages, false, false);

             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.log(Level.WARNING, "Failed to create/configure SSL socket", e);
             }
             return null;
          }

          try
          {
             socket.startHandshake(); // initial call is synchronous
          }
          catch (SSLException e)
          {
             if (!errManager)
             {
                throw e;
             }

             // cert not available failure ?
             int errCodes[] = { 9318, 9407 };
             String errMessages[] =
             {
                "Secure Socket Layer (SSL) failure. error code -54:  unable to get local issuer " +
                      "certificate: for " + certId + " in " + wa.trustStoreFilename,
                "Connection failure for host " + host + " port " + port + " transport TCP"
             };
             ErrorManager.recordOrShowError(errCodes, errMessages, false, false);

             if (LOG.isLoggable(Level.WARNING))
             {
                LOG.log(Level.WARNING, "Start handshake SSLException.", e);
             }
             return null;
          }

          if (!nohostverify)
          {
             // for client sockets, host verification is needed: unless -nohostverify is set,
             // the common name from the server's certificate and the target host name must match
             SSLSession sslSess = socket.getSession();
             boolean cnFound = false;
             String certHostName = "";
             StringTokenizer tok = new StringTokenizer(sslSess.getPeerPrincipal().getName(), ",");
             while (tok.hasMoreTokens() && !cnFound)
             {
                String token = tok.nextToken();
                if (token.startsWith("CN="))
                {
                   certHostName = token.substring(3).toLowerCase();
                   cnFound = true;
                }
             }
             boolean failed = certHostName.isEmpty();
             if (!failed)
             {
                if (certHostName.startsWith("*"))
                {
                   // Eg: certHostName = "*.google.com", host = "www.google.com"
                   failed = !host.endsWith(certHostName.substring(1)) ||
                         host.substring(0, host.length() - certHostName.length() + 1).contains(".");
                }
                else
                {
                   failed = !certHostName.equals(host);
                }
             }

             if (failed)
             {
                if (!errManager)
                {
                   throw new SecurityException("Host does not match certificate.");
                }

                // test fails if certificate does not provide the CN to compare to
                int errCodes[] = { 9318, 9407 };
                String errMessages[] =
                {
                   "Secure Socket Layer (SSL) failure. error code -55:  " +
                         "CONNECT HostName: (" + socket.getInetAddress().getHostName() +
                         ") does not match Certificate: (" + certHostName + ")",
                   "Connection failure for host " + host + " port " + port +
                         " transport TCP failed host-name verification"
                };
                ErrorManager.recordOrShowError(errCodes, errMessages, false, false);
                return null;
             }
          }

          return socket;
       }

       /**
        * When the server socket accepts a new connection, this will make sure the new socket is
        * registered and a {@link SocketImpl} resource is also created on P2J server side.
        *
        * @param    client
        *           The client socket.
        * @param    ssl
        *           Flag indicating if this is a SSL connection.
        *
        * @return   The created {@link SocketImpl} instance with the details of the new socket.
        *
        * @throws   IOException
        *           If the socket's input or output streams couldn't be obtained.
        */
       public static SocketData registerClientSocket(Socket client, boolean ssl)
       throws IOException
       {
          String sslServerName = null;
          if (ssl)
          {
             try
             {
                sslServerName = ((SSLSocket) client).getSession().getPeerPrincipal().getName();
             }
             catch (Exception e)
             {
                if (LOG.isLoggable(Level.FINER))
                {
                   LOG.log(Level.FINER, "Error determining the principal.", e);
                }
             }
          }

          int id = Utils.uniqueId();
          WorkArea wa = local.get();

          synchronized (wa)
          {
             SocketData sd = new SocketData(id, client, sslServerName);

             wa.addSocket(id, sd);

             sd.startThreads();

             return sd;
          }
       }

       /**
        * Identify the socket {@link SocketData socket data} using the given ID.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       private static SocketData locate(int id)
       {
          return local.get().locate(id);
       }

       /**
        * Identify the associated {@link Socket socket} for the given ID.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       private static Socket locateSocket(int id)
       {
          return local.get().locateSocket(id);
       }

       /**
        * Wait until all socket-related threads are initialized.
        *
        * @param    id
        *           The socket ID on P2J client side.
        * @param    ready
        *           The signal to wait on. If <code>null</code>, it returns <code>true</code>.
        *
        * @return   <code>true</code> if all threads are initialized or <code>ready</code> is null;
        *           <code>false</code> if someone interrupted the {@link CountDownLatch#await}
        */
       private static boolean isReady(int id, CountDownLatch ready)
       {
          try
          {
             ready.await();
             return true;
          }
          catch (InterruptedException e)
          {
             if (LOG.isLoggable(Level.FINER))
             {
                final String msg = "The [%d] socket is not initialized yet!";
                LOG.log(Level.FINER, String.format(msg, id), e);
             }

             return false;
          }
       }

       /**
        * Establish a connection to the given host and port.
        *
        * @param    resourceId
        *           The ID of the associated {@link SocketImpl} resource on server side.
        * @param    host
        *           The IP address or name of the target host.
        * @param    port
        *           The port number.
        * @param    timeout
        *           connect timeout
        *
        * @return   The ID of the associated socket, if the connection was established.
        *
        * @throws   IOException
        *           In case of errors while establishing the connection.
        */
       @Override
       public int connect(long resourceId, String host, int port, int timeout)
       throws IOException
       {
          int id = -1;

          InetSocketAddress address = new InetSocketAddress(host, port);
          Socket socket = new Socket();
          socket.bind(null);
          socket.connect(address, timeout < 0 ? local.get().socketConnectTimeoutMs : timeout);

          id = Utils.uniqueId();

          WorkArea wa = local.get();
          synchronized (wa)
          {
             SocketData sd = new SocketData(id, resourceId, socket, null);
             wa.addSocket(id, sd);

             sd.startThreads();
          }

          return id;
       }

       /**
        * Establish a SSL connection to the given host and port.
        * <p>
        *
        * @param   resourceId
        *          The ID of the associated {@link SocketImpl} resource on server side.
        * @param   host
        *          The IP address or name of the target host.
        * @param   port
        *          The port number.
        * @param   serverName
        *          The server name indicator (SNI) set via <code>-servername</code>, for TLS connections.
        * @param   nosessionreuse
        *          Flag indicating that the ssl session ID must not be reused.
        * @param   nohostverify
        *          Flag indicating that the target host name must not be validated against the
        *          certificate's principal name.
        * @param   protocol
        *          The standard name of the requested protocol as defined by the SSLContext class.
        * @param   timeout
        *          connect timeout
        *
        * @return  The ID of the associated socket, if the connection was established.
        *
        * @throws  IOException
        *          In case of errors while establishing the connection.
        * @throws  GeneralSecurityException
        *          In case of errors while establishing the SSL connection.
        */
       @Override
       public int connectSSL(long    resourceId,
                             String  host,
                             int     port,
                             String  serverName,
                             boolean nosessionreuse,
                             boolean nohostverify,
                             String  protocol,
                             int timeout)
       throws IOException,
              GeneralSecurityException
       {
          SSLSocket socket = createSslSocket(host.toLowerCase(), // used for ignore-case compare
                                             port,
                                             serverName,
                                             null,
                                             0,
                                             timeout < 0 ? 0 : timeout,
                                             nohostverify,
                                             true,
                                             protocol);
          if (socket == null)
          {
             // any encountered conditions were already handled by ErrorManager
             return -1;
          }

          int id = Utils.uniqueId();

          WorkArea wa = local.get();
          synchronized (wa)
          {
             SocketData sd = new SocketData(id, resourceId, socket, host);
             wa.addSocket(id, sd);

             sd.startThreads();
          }

          return id;
       }

       /**
        * Check if the java socket is still connected.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   <code>true</code> if the java socket is still connected.
        */
       @Override
       public boolean isConnected(int id)
       {
          SocketData sd = locate(id);
          return sd != null && !sd.listener.isTerminated();
       }

       /**
        * Disconnect this socket.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   <code>true</code> if no errors were encountered during disconnect.
        */
       @Override
       public boolean disconnect(int id)
       {
          WorkArea wa = local.get();

          return wa.disconnect(id);
       }

       /**
        * Get the available bytes to be read.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       @Override
       public long getBytesAvailable(int id)
       {
          SocketData sd = locate(id);

          return sd.input.getBytesAvailable();
       }

       /**
        * Java implementation for the READ() SOCKET method, this will read the specified number of
        * bytes from the available data on the socket, and put it inside the memory buffer
        * argument at the specified position.
        * <p>
        * Optionally the read mode can be set by the following values:
        * <ul>
        *    <li>1 (READ-AVAILABLE)</li>
        *    <li>2 (READ-EXACT-NUM)</li>
        * </ul>
        * <p>
        * During READ processing, the {@link ReadListener read listener} will be blocked from raising
        * new READ-RESPONSE events.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    buffer
        *           The memory buffer where read() puts the data.
        * @param    pos
        *           The position in the memory buffer where the data is added.
        * @param    len
        *           The number of bytes to read from the socket.
        * @param    mode
        *           The mode to set for the reading operation.
        *
        * @return   The number of read bytes. 0 means a timeout during read, -1 meas the stream
        *           reached EOF (socket closed).
        *
        * @throws   IOException
        *           In case IO errors were encountered during read.
        */
       @Override
       public long read(int id, memptr buffer, long pos, long len, int mode)
       throws IOException
       {
          // TODO: memptr may be on server-side, improve this

          final int CHUNK_SIZE = 4096;
          SocketData sd = locate(id);

          // do not allow the socket to listen for incoming data and raise READ-RESPONSE events,
          // as we are processing a READ
          sd.listener.startRead();

          long bytesRead = 0;

          try
          {
             if (mode == com.goldencode.p2j.util.Socket.READ_AVAILABLE)
             {
                // in this mode, we need to block until at least one byte is received and then read
                // at most len bytes without blocking
                do
                {
                   // read a number of bytes, but make sure we don't exceed the remainder length and
                   // always use chunks
                   int toRead = (int) Math.min(sd.input.getBytesAvailable(), len);
                   toRead = Math.min(CHUNK_SIZE, toRead);

                   byte[] b = new byte[toRead];

                   // save the actual number of bytes read (may be different)
                   int read = sd.input.read(b, 0, toRead);
                   if (read == 0)
                   {
                      // nothing available yet
                      continue;
                   }
                   if (read == -1)
                   {
                      // EOF
                      break;
                   }

                   byte[] bytes = new byte[read];
                   System.arraycopy(b, 0, bytes, 0, read);
                   raw r = new raw(bytes);
                   buffer.setBytes(r, pos);
                   pos = pos + read;

                   // decrement the number of bytes to be read
                   len = len - read;

                   bytesRead = bytesRead + read;

                   // terminate when no more data is being received or the max length is reached
                }
                while (sd.input.getBytesAvailable() > 0 && len > 0);
             }
             else
             {
                // in this mode, we need to read exactly len bytes
                while (len > 0)
                {
                   // while more data is needed to be read, read it in chunks
                   int toRead = (int) Math.min(CHUNK_SIZE, len);
                   byte[] b = new byte[toRead];
                   int read = sd.input.read(b, 0, toRead);
                   if (read == 0)
                   {
                      // nothing available yet
                      continue;
                   }
                   if (read == -1)
                   {
                      // EOF
                      break;
                   }

                   // copy the read bytes directly to the memptr buffer
                   byte[] bytes = new byte[read];
                   System.arraycopy(b, 0, bytes, 0, read);
                   raw r = new raw(b);
                   buffer.setBytes(r, pos);
                   pos = pos + read;

                   // decrement the number of bytes to be read
                   len = len - read;

                   bytesRead = bytesRead + read;
                }
             }
          }
          finally
          {
             sd.listener.endRead();
          }

          return bytesRead;
       }

       /**
        * Java implementation for the WRITE() SOCKET method, this will write data from the specified
        * buffer location (position and number of bytes) to the socket.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    buffer
        *           The memory buffer from which the data will be written to the socket.
        * @param    pos
        *           The position in the memory buffer from where the data to be taken starts.
        * @param    len
        *           The number of bytes to take from the memory buffer.
        *
        * @return   The number of written bytes.
        *
        * @throws   IOException
        *           In case IO errors were encoubtered during write.
        */
       @Override
       public long write(int id, memptr buffer, long pos, long len)
       throws IOException
       {
          // TODO: memptr may be on server-side, improve this

          SocketData sd = locate(id);

          long writtenLength = 0;

          // the data is in memory (native if memptr, java heap if raw). we need to access it from
          // there; write it in chunks of 32000 bytes (raw max size). this ensures we always bring
          // maximum amount of bytes into java heap (without using native memory again).
          while (len > 0)
          {
             // bring into java heap (in case memptr is used) at most raw.MAX_SIZE bytes from the
             // buffer
             BinaryData data = buffer.getBytes(pos, Math.min(len, raw.MAX_SIZE));
             byte[] b = data.getByteArray();

             len -= b.length;
             pos += b.length;
             writtenLength += b.length;

             sd.output.write(b);
          }

          return writtenLength;
       }

       /**
        * In case of a SSL connection, get the digital certificate subject name of the server for
        * the current SSL session; else, return <code>null</code>.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       @Override
       public String getSslServerName(int id)
       {
          return locate(id).sslServerName;
       }

       /**
        * Get the TCP_NODELAY option value for this socket.
        * See {@link java.net.Socket#getTcpNoDelay} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public boolean getTcpNoDelay(int id)
       throws SocketException
       {
          return locateSocket(id).getTcpNoDelay();
       }

       /**
        * Get the SO_LINGER option value for this socket.
        * See {@link java.net.Socket#getSoLinger} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public int getSoLinger(int id)
       throws SocketException
       {
          return locateSocket(id).getSoLinger();
       }

       /**
        * Get the SO_KEEPALIVE option value for this socket.
        * See {@link java.net.Socket#getKeepAlive} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public boolean getKeepAlive(int id)
       throws SocketException
       {
          return locateSocket(id).getKeepAlive();
       }

       /**
        * Get the SO_REUSEADDR option value for this socket.
        * See {@link java.net.Socket#getReuseAddress} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public boolean getReuseAddress(int id)
       throws SocketException
       {
          return locateSocket(id).getReuseAddress();
       }

       /**
        * Get the SO_RCVBUF option value for this socket.
        * See {@link java.net.Socket#getReceiveBufferSize} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public int getReceiveBufferSize(int id)
       throws SocketException
       {
          return locateSocket(id).getReceiveBufferSize();
       }

       /**
        * Get the SO_SNDBUF option value for this socket.
        * See {@link java.net.Socket#getSendBufferSize} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public int getSendBufferSize(int id)
       throws SocketException
       {
          return locateSocket(id).getSendBufferSize();
       }

       /**
        * Get the SO_RCVTIMEO option value for this socket.
        * See {@link java.net.Socket#getSoTimeout} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        *
        * @throws   SocketException
        *           If the socket option could not be interrogated.
        */
       @Override
       public int getSoTimeout(int id)
       throws SocketException
       {
          return locateSocket(id).getSoTimeout();
       }

       /**
        * Set the TCP_NODELAY option for this socket.
        * See {@link java.net.Socket#setTcpNoDelay} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    on
        *           The TCP_NODELAY value.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setTcpNoDelay(int id, boolean on)
       throws SocketException
       {
          locateSocket(id).setTcpNoDelay(on);
       }

       /**
        * Set the SO_LINGER option for this socket.
        * See {@link java.net.Socket#setSoLinger} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    on
        *           <code>true</code> to turn on the SO_LINGER option.
        * @param    linger
        *           The linger time.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setSoLinger(int id, boolean on, int linger)
       throws SocketException
       {
          locateSocket(id).setSoLinger(on, linger);
       }

       /**
        * Set the SO_KEEPALIVE option for this socket.
        * See {@link java.net.Socket#setKeepAlive} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    on
        *           The SO_KEEPALIVE value.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setKeepAlive(int id, boolean on)
       throws SocketException
       {
          locateSocket(id).setKeepAlive(on);
       }

       /**
        * Set the SO_REUSEADDR option for this socket.
        * See {@link java.net.Socket#setReuseAddress} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    on
        *           The SO_REUSEADDR value.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setReuseAddress(int id, boolean on)
       throws SocketException
       {
          locateSocket(id).setReuseAddress(on);
       }

       /**
        * Set the SO_RCVBUF value (the receive buffer size) for this socket.
        * See {@link java.net.Socket#setReceiveBufferSize} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    size
        *           The SO_RCVBUF value.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setReceiveBufferSize(int id, int size)
       throws SocketException
       {
          locateSocket(id).setReceiveBufferSize(size);
       }

       /**
        * Set the SO_SNDBUF value (the send buffer size) for this socket.
        * See {@link java.net.Socket#setSendBufferSize} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    size
        *           The SO_SNDBUF value.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setSendBufferSize(int id, int size)
       throws SocketException
       {
          locateSocket(id).setSendBufferSize(size);
       }

       /**
        * Set the SO_RCVTIMEO value (the number of seconds the socket waits to receive data before
        * timing out) for this socket.
        * See {@link java.net.Socket#setSoTimeout} for more details.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        * @param    timeout
        *           The SO_RCVTIMEO value.
        *
        * @throws   SocketException
        *           If the socket option could not be set.
        */
       @Override
       public void setSoTimeout(int id, int timeout)
       throws SocketException
       {
          locateSocket(id).setSoTimeout(timeout);
       }

       /**
        * Get the IP address of this side of this connection.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       @Override
       public String getLocalHost(int id)
       {
          return locateSocket(id).getLocalAddress().getHostAddress();
       }

       /**
        * Get the port number of this side of this connection.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       @Override
       public int getLocalPort(int id)
       {
          return locateSocket(id).getLocalPort();
       }

       /**
        * Get the host IP address of the remote side of this connection.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       @Override
       public String getRemoteHost(int id)
       {
          return locateSocket(id).getInetAddress().getHostAddress();
       }

       /**
        * Get the port number of the remote side of this connection.
        *
        * @param    id
        *           The ID of the associated socket on P2J Client side.
        *
        * @return   See above.
        */
       @Override
       public int getRemotePort(int id)
       {
          return locateSocket(id).getPort();
       }

       /**
        * A container holding the various client-socket data.
        */
       public static class SocketData
       {
          /** The ID of this socket, on P2J client-side. */
          final int id;

          /** The ID of this {@link com.goldencode.p2j.util.Socket} resource, on P2J server-side. */
          final long resourceId;

          /** Signals when all the threads have started. */
          private final CountDownLatch ready = new CountDownLatch(2);

          /** The associated socket. */
          private final Socket socket;

          /** The SSL server name, in case of a SSL connection. */
          private final String sslServerName;

          /** Listener dedicated to posting READ-RESPONSE events. */
          private final ReadListener listener;

          /**
           * A specialized instance dedicated to reading data from the socket, in a separate thread.
           */
          private final StagedInput input;

          /** This socket's output stream. */
          private final OutputStream output;

          /**
           * Save the data for this socket and create the specialized instances to
           * {@link ReadListener raise READ-RESPONSE events} and to {@link StagedInput read data} from
           * the socket.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           * @param    socket
           *           The associated java socket.
           * @param    sslServerName
           *           The SSL server name, in case of a SSL connection. <code>null</code> otherwise.
           *
           * @throws   IOException
           *           If the socket's input or output streams couldn't be obtained.
           */
          public SocketData(int id, Socket socket, String sslServerName)
          throws IOException
          {
             this.id = id;

             this.socket = socket;
             this.sslServerName = sslServerName;

             this.input = new StagedInput(id, socket, socket.getInputStream(), ready);
             this.output = socket.getOutputStream();

             // create the resource on the server if input/output were obtained
             this.resourceId = local.get().server.createSocket(id);

             this.listener = new ReadListener(this, ready);
          }

          /**
           * Save the data for this socket and create the specialized instances to
           * {@link ReadListener raise READ-RESPONSE events} and to {@link StagedInput read data} from
           * the socket.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           * @param    resourceId
           *           The ID of the associated {@link com.goldencode.p2j.util.Socket} on P2J Server
           *           side.
           * @param    socket
           *           The associated java socket.
           * @param    sslServerName
           *           The SSL server name, in case of a SSL connection. <code>null</code> otherwise.
           *
           * @throws   IOException
           *           If the socket's input or output streams couldn't be obtained.
           */
          public SocketData(int id, long resourceId, Socket socket, String sslServerName)
          throws IOException
          {
             this.id = id;
             this.resourceId = resourceId;
             this.socket = socket;
             this.sslServerName = sslServerName;

             this.input = new StagedInput(id, socket, socket.getInputStream(), ready);
             this.output = socket.getOutputStream();

             this.listener = new ReadListener(this, ready);
          }

          /**
           * Start the {@link ReadListener listener} and {@link StagedInput reader} threads.
           */
          public void startThreads()
          {
             this.input.start();
             this.listener.start();
          }

          /**
           * Getter for {@link #resourceId}.
           *
           * @return  See above.
           */
          public long getResourceId()
          {
             return resourceId;
          }
       }

       /**
        * Container for the context-local data. This will also ensure sync access to the
        * {@link #sockets socket registry}.
        */
       private static class WorkArea
       implements SessionListener
       {
          /** Map for protocol name : SSL Socket factory pair */
          private final Map<String, SSLSocketFactory> factories = new HashMap<>();

          /** -nohostverify CONNECT arg, only for SOAP calls. */
          private boolean nohostverify;

          /**
           * Filename of client-side SSL certificate store. Default is {@code trusted-cert.store} from
           * current directory.
           * <p>
           * In P4GL the store is a folder and each imported certificate being a file in PEM (ASCII plain
           * text). They should be imported to this store.
           */
          private String trustStoreFilename = null;

          /**
           * The password for access to client-side SSL certificate store. Default is blank.
           */
          private String trustStorePassword = null;

          /** A registry containing the open sockets. */
          private final Map<Integer, SocketData> sockets = new HashMap<>();

          /** The socket connect timeout value taken from directory per client session basis. */
          private int socketConnectTimeoutMs = DirectoryManager.getInstance().getInt(Directory.ID_RELATIVE_BOTH,
                                                                                     "socket-connect-timeout", 0);

          /**
           * The proxy to the {@link ServerExports} (which can be local or remote, depending on how the client
           * sockets are configured.
           */
          private final ServerExports server;

          /**
           * Initialize this instance.
           */
          public WorkArea()
          {
             boolean clientSide = SessionManager.get().isLeaf();
             ClientParameters params;
             if (clientSide)
             {
                params = ThinClient.getClientParameters();
                server = ThinClient.getInstance().getServer();
             }
             else
             {
                params = StandardServer.getClientParameters();
                server = (ServerExports) RemoteObject.obtainLocalInstance(ServerExports.class, true);
             }

             SessionManager.get().getSession().addSessionListener(this);

             trustStoreFilename = params.socketTrustStoreFilename;
             trustStorePassword = params.socketTrustStorePassword;
             if (LOG.isLoggable(Level.INFO))
             {
                LOG.info("LowLevelSocketImpl trustStoreFilename: [" + trustStoreFilename + "]");
             }
          }

          /**
           * Disconnect this socket.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           *
           * @return   <code>true</code> if no errors were encountered during disconnect.
           */
          public boolean disconnect(int id)
          {
             synchronized (this)
             {
                SocketData sd = locate(id);
                if (sd == null)
                {
                   return true; // socket might be already closed disconnected from the other side
                }
                try
                {
                   sd.socket.close();
                }
                catch (IOException e)
                {
                   if (LOG.isLoggable(Level.FINER))
                   {
                      final String msg = "Error closing socket [%d].";
                      LOG.log(Level.FINER, String.format(msg, id), e);
                   }
                }

                // notify all threads, to ensure when the client terminates the threads are aware
                sd.listener.setTerminated(true);

                removeSocket(id);

                sd.input.terminate();
             }

             return true;
          }

          /**
           * When the P2J client session is ending, disconnect all active sockets.
           *
           * @param    session
           *           The session that is ending.
           */
          @Override
          public void terminate(Session session)
          {
             synchronized (this)
             {
                Map<Integer, SocketData> copy = new HashMap<>(this.sockets);
                for (SocketData socket : copy.values())
                {
                   disconnect(socket.id);
                }
             }
          }

          /**
           * This method is called when the session is starting.
           *
           * @param   session
           *          The session that is starting.
           */
          @Override
          public void initialize(Session session)
          {
             // no-ops
          }

          /**
           * Add a new socket to the {@link #sockets registry}.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           * @param    sd
           *           The data for this socket.
           */
          public synchronized void addSocket(int id, SocketData sd)
          {
             sockets.put(id, sd);
          }

          /**
           * Locate the socket with the given ID from the {@link #sockets registry}.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           */
          private synchronized Socket locateSocket(int id)
          {
             return sockets.get(id).socket;
          }

          /**
           * Locate the {@link SocketData} for the the given ID.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           */
          private synchronized SocketData locate(int id)
          {
             return sockets.get(id);
          }

          /**
           * Remove the socket with the given ID from the {@link #sockets registry}.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           */
          private synchronized void removeSocket(int id)
          {
             sockets.remove(id);
          }
       }

       /**
        * This class is dedicated to listening for incoming socket data. While the socket is still
        * connected, it will check if the {@link StagedInput reader} can read some data without
        * blocking. If so, it will post a READ-RESPONSE event on the server (the server-side is
        * responsible of reaching back to the client and posting the event, as the server-side holds
        * the implementation).
        * <p>
        * Once a byte can be read, it will wait for a {@link LowLevelSocketImpl#read} to be executed,
        * as it must not post more than a READ-RESPONSE event per "incoming data" case.
        */
       private static class ReadListener
       implements Runnable
       {
          /** The associated {@link SocketData} instance. */
          private final SocketData sd;

          /** Flag indicating we are executing a READ. */
          private volatile boolean inRead = false;

          /** Flag indicating the listener must been terminated. */
          private volatile boolean terminated = false;

          /** The signal to use when this thread is ready. */
          private CountDownLatch ready;

          /**
           * Flag indicating that the SocketListener thread is waiting for the read operation to be either active
           * or not.
           */
          private volatile boolean waitingForRead = true;

          /**
           * Initialize this instance.
           *
           * @param    sd
           *           The {@link SocketData} instance.
           * @param    ready
        *              The latch to wait on.
           */
          public ReadListener(SocketData sd, CountDownLatch ready)
          {
             this.sd = sd;
             this.ready = ready;
          }

          /**
           * Set the {@link #terminated} flag to the specified value.
           *
           * @param    terminated
           *           The new value for the {@link #terminated} flag.
           */
          public synchronized void setTerminated(boolean terminated)
          {
             this.terminated = terminated;
             notifyAll();
          }

          /**
           * Check if the listener thread is terminated. This is true if {@link SocketData#socket} is
           * closed or the {@link #terminated} flag is true.
           *
           * @return    See above.
           */
          public synchronized boolean isTerminated()
          {
             return terminated || sd.socket.isClosed();
          }

          /**
           * Execute a loop to post READ-RESPONSE events until the {@link #isTerminated()} returns.
           * <code>true</code>.
           * <p>
           * Once the loop has been terminated, a final READ-RESPONSE event will be posted.
           */
          @Override
          public void run()
          {
             ready.countDown();

             try
             {
                ServerExports server = local.get().server;

                while (!isTerminated())
                {
                   // make sure no read is active
                   waitForNotify(false);

                   // a read can start here, but all server events during read all be discarded

                   if (sd.input.hasMoreData())
                   {
                      // post a ServerEvent. for this, must use a server-side call (as the server
                      // event is implemented there)
                      server.readResponseEvent(sd.resourceId);

                      // do not process more events until a READ is executed
                      waitForNotify(true);
                   }
                   else
                   {
                      setTerminated(true);
                   }
                }

                // socket has been closed, send a new READ-RESPONSE event
                if (!terminated)
                {
                   // do not raise a READ-RESPONSE if the client is being terminated.
                   server.readResponseEvent(sd.resourceId);
                }
             }
             finally
             {
                SessionManager.get().deregisterAsyncThread(Thread.currentThread());

                // cleanup
                local.get().removeSocket(sd.id);
             }
          }

          /**
           * Start a new thread, in which this instance will be executed.  The thread's name will be
           * <code>Socket listener [%d]</code>, where <code>%d</code> is replaced with the
           * {@link SocketData#id} value.
           * <p>
           * This thread is marked as async, as all outgoing requests must be processed async, in
           * their own {@link Dispatcher dispatcher}.
           */
          public void start()
          {
             Thread thread = new AssociatedThread(this);
             // mark this thread as async
             SessionManager.get().registerAsyncThread(thread);
             String name = String.format("Socket listener [%d]", sd.id);
             thread.setName(name);
             thread.start();
          }

          /**
           * Called when a READ is started. This will set the {@link #inRead} flag to true and notify
           * all listeners.
           */
          public synchronized void startRead()
          {
             if (!isReady(sd.id, ready))
             {
                return;
             }

             inRead = true;
             // notify all threads, to ensure when the client terminates the threads are aware
             notifyAll();

             // wait for the SocketListener thread to be in the 'waiting for the read operation active' state
             while (!waitingForRead && !isTerminated())
             {
                try
                {
                   wait();
                }
                catch (InterruptedException e)
                {
                   // ignore
                }
             }
          }

          /**
           * Called when a REAd is ended. This will set the {@link #inRead} flag to false and notify
           * all listeners.
           */
          public synchronized void endRead()
          {
             if (!isReady(sd.id, ready))
             {
                return;
             }

             inRead = false;
             // notify all threads, to ensure when the client terminates the threads are aware
             notifyAll();

             // wait for the SocketListener thread to be in the 'waiting for the read operation not active' state
             while (waitingForRead && !isTerminated())
             {
                try
                {
                   wait();
                }
                catch (InterruptedException e)
                {
                   // ignore
                }
             }
          }

          /**
           * Wait for notification that a read has either started or ended.
           *
           * @param    inRead
           *           <code>true</code> to wait for a READ start, <code>false</code> to wait for a
           *           READ end.
           */
          private synchronized void waitForNotify(boolean inRead)
          {
             while (this.inRead != inRead && !isTerminated())
             {
                try
                {
                   // wait for the other thread to do its job
                   wait();
                }
                catch (InterruptedException exc)
                {
                   // ignore
                }
             }

             // notify the read operation that we are either waiting for it to start or end
             waitingForRead = inRead;
             // notify all threads, to ensure when the client terminates the threads are aware
             notifyAll();
          }
       }

       /**
        * Due to the nature of java sockets and how the 4GL sockets behave, all reading must be done
        * in a dedicated thread. Also, reading data via a java socket input stream is not thread-safe;
        * thus, reading data must be staged, following these rules, after a byte is read:
        * <ol>
        *    <li>Save it in {@link #lastByte} field.</li>
        *    <li>Notify the {@link ReadListener} (which waits for data in {@link #hasMoreData},
        *        via the {@link #rLock}.
        *    <li>If EOF is reached, then end the loop.</li>
        *    <li>Wait for someone to read the data (on the {@link #rLock}). The {@link #rLock} will be
        *        notified only after a {@link #read} call is finished.</li>
        * </ol>
        * The {@link ReadListener} will always call {@link #hasMoreData()}, which will
        * block on the {@link #rLock} until a byte was read.
        * <p>
        * A thread wanting to read data from the socket will always call {@link #read}, which will
        * block for a byte to be read using {@link #hasMoreData()}.
        * <p>
        * A thread wanting to interrogate the available bytes on the input stream will always call
        * {@link #getBytesAvailable}, which will use the {@link InputStream#available} to determine
        * the available bytes.
        * <p>
        * This approach was chosen after chasing a few other solution:
        * <ol>
        *    <li>Using {@link InputStream#available()} on the socket's input stream to check if data
        *        is incoming is not enough because this will not block. A solution which calls
        *        <code>available()</code> in a loop (eventually after waiting for some millis) will
        *        pose unnecessary overhead on the CPU or it will cause delays from posting the
        *        READ-RESPONSE event.</li>
        *    <li>If channels are used, a {@link Selector} can be used to block until data is available.
        *        But this will force the input and output stream to be non-blocking. Also, when using
        *        channels in blocking mode, read and write can't be used in multiple threads; if a
        *        thread is blocking on a read, then a write will be possible from another thread only
        *        if the read has released the channel's {@link SocketChannel#blockingLock() lock}.
        *     </li>
        * </ol>
        */
       private static class StagedInput
       implements Runnable
       {
          /** Value indicating that no byte was read. */
          private static final int NO_BYTE = -2;

          /** Lock used for reading a byte. */
          private final Object rLock = new Object();

          /** The socket's input stream. */
          private final InputStream input;

          /** The associated java socket. */
          private final Socket socket;

          /** The ID of this socket on P2J client side. */
          private final int id;

          /** The signal to use when this thread is ready. */
          private CountDownLatch ready;

          /** The last byte read. */
          private int lastByte = NO_BYTE;

          /** Flag indicating there was a timeout in the last read operation. */
          private volatile boolean hadTimeout = false;

          /**
           * Initialize this staged input stream.
           *
           * @param    id
           *           The ID of the associated socket on P2J Client side.
           * @param    socket
           *           The associated java socket.
           * @param    input
           *           The socket's input stream.
           * @param    ready
           *           The signal to use when this thread is ready.
           */
          public StagedInput(int id, Socket socket, InputStream input, CountDownLatch ready)
          {
             this.id = id;
             this.socket = socket;
             this.input = input;
             this.ready = ready;
          }

          /**
           * Wait for data to be received by the socket, in a dedicated thread.
           */
          @Override
          public void run()
          {
             ready.countDown();

             try
             {
                while (!socket.isClosed())
                {
                   try
                   {
                      int b = input.read();
                      hadTimeout = false;

                      // I've read data, notify the listener
                      synchronized (rLock)
                      {
                         lastByte = b;
                         // notify all threads, to ensure when the client terminates the threads are aware
                         rLock.notifyAll();

                         if (b == -1)
                         {
                            // EOF reached
                            break;
                         }

                         // don't start waiting if the socket is closed
                         if (!socket.isClosed())
                         {
                            // wait for the data to be read
                            try
                            {
                               rLock.wait();
                            }
                            catch (InterruptedException e)
                            {
                               // ignore this, but log it
                               if (LOG.isLoggable(Level.FINER))
                               {
                                  final String msg =
                                     "The lock waiting for someone to read the data was interrupted.";
                                  LOG.log(Level.FINER, msg, e);
                               }
                            }
                         }
                      }
                   }
                   catch (SocketTimeoutException ex)
                   {
                      // unless the socket is closed, keep reading; thus, socket timeout exceptions are allowed
                      synchronized (rLock)
                      {
                         rLock.notifyAll();
                         hadTimeout = true;
                         if (socket.isClosed())
                         {
                            break;
                         }
                      }
                   }
                   catch (IOException e)
                   {
                      if (LOG.isLoggable(Level.FINER))
                      {
                         LOG.log(Level.FINER, "Problems reading byte!", e);
                      }

                      synchronized (rLock)
                      {
                         rLock.notifyAll();
                         break;
                      }
                   }
                }
             }
             finally
             {
                SessionManager.get().deregisterAsyncThread(Thread.currentThread());
             }
          }

          /**
           * Check if the socket has more data to be read. This will block until a byte was read or
           * EOF is reached.
           *
           * @return   <code>true</code> if data is incoming.
           */
          public boolean hasMoreData()
          {
             if (!isReady(id, ready))
             {
                return false;
             }

             try
             {
                return hasMoreDataWorker(0);
             }
             catch (SocketTimeoutException exc)
             {
                return false;
             }
          }

          /**
           * Determine the number of available bytes on the socket's input stream.  This will include
           * {@link #lastByte} too, if it was read previously.
           *
           * @return   See above.
           */
          public int getBytesAvailable()
          {
             if (!isReady(id, ready))
             {
                return 0;
             }

             synchronized (rLock)
             {
                try
                {
                   return input.available() + (lastByte >= 0 ? 1 : 0);
                }
                catch (IOException e)
                {
                   if (LOG.isLoggable(Level.FINER))
                   {
                      final String msg = "Error in getBytesAvailable for %d";
                      LOG.log(Level.FINER, String.format(msg, id), e);
                   }

                   return 0;
                }
             }
          }

          /**
           * Read <code>len</code> number of bytes and place them in the specified <code>buffer</code>,
           * starting with the specified position.
           *
           * @param    buffer
           *           The buffer where to place the read data.
           * @param    pos
           *           The position in the buffer where to place the data.
           * @param    len
           *           The number of bytes to read.
           *
           * @return   The number of read bytes.
           *
           * @throws   IOException
           *           In case of read errors.
           */
          public int read(byte[] buffer, int pos, int len)
          throws IOException
          {
             if (!isReady(id, ready))
             {
                return 0;
             }

             if (len == 0)
             {
                // nothing to read
                return 0;
             }

             int timeout = 0;
             try
             {
                timeout = socket.getSoTimeout();
             }
             catch (IOException e)
             {
                if (LOG.isLoggable(Level.FINER))
                {
                   LOG.log(Level.FINER, "Error determining SO_RCVTIMEO, assuming 0", e);
                }
             }

             // wait for data to be available, using the current SO-TIMEOUT value
             if (!hasMoreDataWorker(timeout))
             {
                return -1;
             }

             // consume the first byte
             byte b = -1;
             synchronized (rLock)
             {
                b = (byte) (lastByte & 0xFF);
                lastByte = NO_BYTE;

                try
                {
                   buffer[0] = b;
                   pos = pos + 1;
                   len = len - 1;
                   int read = 1;

                   if (len > 0)
                   {
                      read = read + input.read(buffer, pos, len);
                   }

                   return read;
                }
                finally
                {
                   // notify that LAST_BYTE was consumed, so a new byte can be read
                   rLock.notify();
                }
             }
          }

          /**
           * Start a new thread, in which this instance will be executed.  The thread's name will be
           * <code>Socket reader [%d]</code>, where <code>%d</code> is replaced with the {@link #id}
           * value.
           * <p>
           * This thread is marked as async, as all outgoing requests must be processed async, in
           * their own {@link Dispatcher dispatcher}.
           */
          public void start()
          {
             Thread thread = new AssociatedThread(this);
             // mark this thread as async
             SessionManager.get().registerAsyncThread(thread);
             String name = String.format("Socket reader [%d]", id);
             thread.setName(name);
             thread.start();
          }

          /**
           * Terminate the reader.
           */
          public void terminate()
          {
             synchronized (rLock)
             {
                // notify all threads, to ensure when the client terminates the threads are aware
                rLock.notifyAll();
             }
          }

          /**
           * Check if the socket has more data to be read. This will block until a byte was read or
           * EOF is reached.
           *
           * @param    timeout
           *           the value of millis to wait; <code>0</code> to wait indefinitely.
           *
           * @return   <code>true</code> if data is incoming.
           *
           * @throws   SocketTimeoutException
           *           If no byte was read.
           */
          private boolean hasMoreDataWorker(int timeout)
          throws SocketTimeoutException
          {
             int b = NO_BYTE;

             long t1 = System.nanoTime();
             synchronized (rLock)
             {
                while (lastByte == NO_BYTE && !hadTimeout && !socket.isClosed())
                {
                   try
                   {
                      rLock.wait(timeout);
                   }
                   catch (InterruptedException e)
                   {
                      // someone interrupted us, terminate the loop
                      break;
                   }
                }

                if (lastByte != NO_BYTE)
                {
                   b = lastByte;
                }
             }
             long t2 = System.nanoTime();
             long time = timeout - (t2 - t1) / 1000000;

             if (b == NO_BYTE && !socket.isClosed() && timeout != 0 && time <= 0)
             {
                final String msg = "Timeout during read for %d";
                throw new SocketTimeoutException(String.format(msg, id));
             }

             return b >= 0 || (hadTimeout && !socket.isClosed());
          }
       }
    }