PVA: Handle client's connection to server in receive thread

Author: kasemir

core/pva/src/main/java/org/epics/pva/client/ClientTCPHandler.java

@@ -10,7 +10,6 @@
 import static org.epics.pva.PVASettings.logger;
 
 import java.net.InetSocketAddress;
-import java.net.Socket;
 import java.nio.ByteBuffer;
 import java.util.Collection;
 import java.util.concurrent.ConcurrentHashMap;
@@ -56,6 +55,12 @@ class ClientTCPHandler extends TCPHandler
                               new GetTypeHandler(),
                               new RPCHandler());
 
+    /** Address of server to which this client will connect */
+    private final InetSocketAddress server_address;
+
+    /** Is this a TLS connection or plain TCP? */
+    private final boolean tls;
+
     /** Client context */
     private final PVAClient client;
 
@@ -112,32 +117,34 @@ class ClientTCPHandler extends TCPHandler
 
     public ClientTCPHandler(final PVAClient client, final InetSocketAddress address, final Guid guid, final boolean tls) throws Exception
     {
-        super(createSocket(address, tls), true);
+        super(true);
         logger.log(Level.FINE, () -> "TCPHandler " + (tls ? "(TLS) " : "") + guid + " for " + address + " created ============================");
+        this.server_address = address;
+        this.tls = tls;
         this.client = client;
         this.guid = guid;
 
-        // For TLS, check if the socket has a name that's used to authenticate
-        x509_name = tls ? SecureSockets.getPrincipalCN(((SSLSocket) socket).getSession().getLocalPrincipal()) : null;
-
-        // For default EPICS_CA_CONN_TMO: 30 sec, send echo at ~15 sec:
-        // Check every ~3 seconds
-        last_life_sign = last_message_sent = System.currentTimeMillis();
-        final long period = Math.max(1, PVASettings.EPICS_PVA_CONN_TMO * 1000L / 30 * 3);
-        alive_check = timer.scheduleWithFixedDelay(this::checkResponsiveness, period, period, TimeUnit.MILLISECONDS);
-
         // Start receiver, but not the send thread, yet.
         // To prevent sending messages before the server is ready,
         // it's started when server confirms the connection.
         startReceiver();
     }
 
-    private static Socket createSocket(final InetSocketAddress address, final boolean tls) throws Exception
+    @Override
+    protected void initializeSocket() throws Exception
     {
-        final Socket socket = SecureSockets.createClientSocket(address, tls);
+        socket = SecureSockets.createClientSocket(server_address, tls);
         socket.setTcpNoDelay(true);
         socket.setKeepAlive(true);
-        return socket;
+
+        // For TLS, check if the socket has a name that's used to authenticate
+        x509_name = tls ? SecureSockets.getPrincipalCN(((SSLSocket) socket).getSession().getLocalPrincipal()) : null;
+
+        // For default EPICS_CA_CONN_TMO: 30 sec, send echo at ~15 sec:
+        // Check every ~3 seconds
+        last_life_sign = last_message_sent = System.currentTimeMillis();
+        final long period = Math.max(1, PVASettings.EPICS_PVA_CONN_TMO * 1000L / 30 * 3);
+        alive_check = timer.scheduleWithFixedDelay(this::checkResponsiveness, period, period, TimeUnit.MILLISECONDS);
     }
 
     /** @return Client context */

core/pva/src/main/java/org/epics/pva/common/TCPHandler.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2019-2023 Oak Ridge National Laboratory.
+ * Copyright (c) 2019-2025 Oak Ridge National Laboratory.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -16,7 +16,6 @@
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
-import java.util.Objects;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -62,11 +61,16 @@ abstract public class TCPHandler
     private final boolean client_mode;
 
     /** TCP socket to PVA peer
+     *
+     *  Server got this client socket from `accept`.
+     *  Client needs to create the socket and connect to server's address.
      *
      *  Reading and writing is handled by receive and send threads,
      *  but 'protected' so that derived classes may peek at socket properties.
+     *
+     *  @see {@link #initializeSocket()}
      */
-    protected final Socket socket;
+    protected Socket socket = null;
 
     /** Flag to indicate that 'close' was called to close the 'socket' */
     protected volatile boolean running = true;
@@ -117,13 +121,11 @@ public void encodeRequest(final byte version, final ByteBuffer buffer) throws Ex
      *  but will only start sending them when the
      *  send thread is running
      *
-     *  @param socket Socket to read/write
      *  @param client_mode Is this the client, expecting to receive messages from server?
      *  @see #startSender()
      */
-    public TCPHandler(final Socket socket, final boolean client_mode)
+    public TCPHandler(final boolean client_mode)
     {
-        this.socket = Objects.requireNonNull(socket);
         this.client_mode = client_mode;
 
         // Receive buffer byte order is set based on header flag of each received message.
@@ -133,6 +135,13 @@ public TCPHandler(final Socket socket, final boolean client_mode)
         send_buffer.order(ByteOrder.nativeOrder());
     }
 
+    /** Initialize the {@link #socket}. Called by receiver.
+     *
+     *  Server received socket from `accept` during construction and this may be a NOP.
+     *  Client will have to create socket and connect to server's address in here
+     */
+    abstract protected void initializeSocket() throws Exception;
+
     /** Start receiving data
      *  To be called by Client/ServerTCPHandler when fully constructed
      */
@@ -260,6 +269,8 @@ private Void receiver()
     {
         try
         {
+            Thread.currentThread().setName("TCP receiver");
+            initializeSocket();
             Thread.currentThread().setName("TCP receiver " + socket.getLocalSocketAddress());
             logger.log(Level.FINER, () -> Thread.currentThread().getName() + " started for " + socket.getRemoteSocketAddress());
             logger.log(Level.FINER, "Native byte order " + receive_buffer.order());

core/pva/src/main/java/org/epics/pva/server/ServerTCPHandler.java

@@ -1,5 +1,5 @@
 /*******************************************************************************
- * Copyright (c) 2019-2023 Oak Ridge National Laboratory.
+ * Copyright (c) 2019-2025 Oak Ridge National Laboratory.
  * All rights reserved. This program and the accompanying materials
  * are made available under the terms of the Eclipse Public License v1.0
  * which accompanies this distribution, and is available at
@@ -62,7 +62,9 @@ class ServerTCPHandler extends TCPHandler
 
     public ServerTCPHandler(final PVAServer server, final Socket client, final TLSHandshakeInfo tls_info) throws Exception
     {
-        super(client, false);
+        super(false);
+        // Server received the client socket from `accept`
+        this.socket = Objects.requireNonNull(client);
         this.server = Objects.requireNonNull(server);
         this.tls_info = tls_info;
 
@@ -113,6 +115,12 @@ public ServerTCPHandler(final PVAServer server, final Socket client, final TLSHa
         });
     }
 
+    @Override
+    protected void initializeSocket() throws Exception
+    {
+        // Nothing to do, received client socket on construction
+    }
+
     PVAServer getServer()
     {
         return server;