Merge pull request #3429 from ControlSystemStudio/pva_connect

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,11 +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;
 
+        // 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();
+    }
+
+    @Override
+    protected boolean initializeSocket()
+    {
+        try
+        {
+            socket = SecureSockets.createClientSocket(server_address, tls);
+            socket.setTcpNoDelay(true);
+            socket.setKeepAlive(true);
+        }
+        catch (Exception ex)
+        {   
+            logger.log(Level.WARNING, "PVA client cannot connect to " + server_address, ex);
+            return false;
+        }
+
         // For TLS, check if the socket has a name that's used to authenticate
         x509_name = tls ? SecureSockets.getPrincipalCN(((SSLSocket) socket).getSession().getLocalPrincipal()) : null;
 
@@ -126,18 +154,7 @@ public ClientTCPHandler(final PVAClient client, final InetSocketAddress address,
         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
-    {
-        final Socket socket = SecureSockets.createClientSocket(address, tls);
-        socket.setTcpNoDelay(true);
-        socket.setKeepAlive(true);
-        return socket;
+        return true;
     }
 
     /** @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,15 @@ 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.
+     *
+     *  @return Success?
+     */
+    abstract protected boolean initializeSocket();
+
     /** Start receiving data
      *  To be called by Client/ServerTCPHandler when fully constructed
      */
@@ -258,6 +269,18 @@ protected void send(final ByteBuffer buffer) throws Exception
     /** Receiver */
     private Void receiver()
     {
+        // Establish connection
+        Thread.currentThread().setName("TCP receiver");
+        while (! initializeSocket())
+            try
+            {   // Delay for (another) connection timeout, at least 1 sec
+                Thread.sleep(Math.max(1, PVASettings.EPICS_PVA_TCP_SOCKET_TMO) * 1000);
+            }
+            catch (Exception ignore)
+            {
+                // NOP
+            }
+        // Listen on the connection
         try
         {
             Thread.currentThread().setName("TCP receiver " + socket.getLocalSocketAddress());

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,13 @@ public ServerTCPHandler(final PVAServer server, final Socket client, final TLSHa
         });
     }
 
+    @Override
+    protected boolean initializeSocket()
+    {
+        // Nothing to do, received client socket on construction
+        return true;
+    }
+
     PVAServer getServer()
     {
         return server;