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());
}
}
}