Merge pull request #2872 from ControlSystemStudio/pva_2871

PVA: TCPHandler started receiver too early

Author: kasemir

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

@@ -124,9 +124,11 @@ public ClientTCPHandler(final PVAClient client, final InetSocketAddress address,
         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);
-        // Don't start the send thread, yet.
+
+        // 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

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

@@ -16,6 +16,7 @@
 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;
@@ -61,7 +62,7 @@ abstract public class TCPHandler
     private final boolean client_mode;
 
     /** TCP socket to PVA peer
-     * 
+     *
      *  Reading and writing is handled by receive and send threads,
      *  but 'protected' so that derived classes may peek at socket properties.
      */
@@ -105,10 +106,10 @@ public void encodeRequest(final byte version, final ByteBuffer buffer) throws Ex
     });
 
     /** Thread that runs {@link TCPHandler#receiver()} */
-    private final Future<Void> receive_thread;
+    private volatile Future<Void> receive_thread = null;
 
     /** Thread that runs {@link TCPHandler#sender()} */
-    private volatile Future<Void> send_thread;
+    private volatile Future<Void> send_thread = null;
 
     /** Start receiving messages
      *
@@ -122,16 +123,21 @@ public void encodeRequest(final byte version, final ByteBuffer buffer) throws Ex
      */
     public TCPHandler(final Socket socket, final boolean client_mode)
     {
-        this.socket = socket;
+        this.socket = Objects.requireNonNull(socket);
         this.client_mode = client_mode;
 
         // Receive buffer byte order is set based on header flag of each received message.
         // Send buffer of server and client starts out with native byte order.
         // For server, it stays that way.
         // For client, order is updated during connection validation (PVAHeader.CTRL_SET_BYTE_ORDER)
         send_buffer.order(ByteOrder.nativeOrder());
+    }
 
-        // Start receiving data
+    /** Start receiving data
+     *  To be called by Client/ServerTCPHandler when fully constructed
+     */
+    protected void startReceiver()
+    {
         receive_thread = thread_pool.submit(this::receiver);
     }
 
@@ -558,7 +564,7 @@ public void close(final boolean wait)
         {
             running = false;
             socket.close();
-            if (wait)
+            if (wait && receive_thread != null)
                 receive_thread.get(5, TimeUnit.SECONDS);
         }
         catch (Exception ex)

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

@@ -12,6 +12,7 @@
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.nio.ByteBuffer;
+import java.util.Objects;
 import java.util.logging.Level;
 
 import org.epics.pva.common.CommandHandlers;
@@ -62,10 +63,11 @@ class ServerTCPHandler extends TCPHandler
     public ServerTCPHandler(final PVAServer server, final Socket client, final TLSHandshakeInfo tls_info) throws Exception
     {
         super(client, false);
-        this.server = server;
+        this.server = Objects.requireNonNull(server);
         this.tls_info = tls_info;
 
         server.register(this);
+        startReceiver();
         startSender();
 
         // Initialize TCP connection by setting byte order..