Defer main loop start until after the header is sent

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -334,6 +334,12 @@ public void start()
             _frameHandler.close();
             throw ioe;
         }
+        // Only start reading frames once the header has actually gone out: for
+        // frame handlers with their own dedicated I/O thread (e.g. blocking IO),
+        // starting it any earlier would race with sendHeader() over who drives
+        // the TLS handshake, and the loser gets a confusing generic SocketException
+        // instead of the real SSLHandshakeException.
+        this._frameHandler.startProcessing();
 
         AMQP.Connection.Start connStart;
         AMQP.Connection.Tune connTune = null;

src/main/java/com/rabbitmq/client/impl/FrameHandler.java

@@ -55,14 +55,22 @@ default boolean internalHearbeat() {
 
     void initialize(AMQConnection connection);
 
-    default void finishConnectionNegotiation() {
+    /**
+     * Start processing incoming frames, once the connection header has been sent.
+     * Frame handlers that already read asynchronously (e.g. Netty, NIO) can ignore this,
+     * as reading may already be under way by the time {@link #initialize(AMQConnection)} runs.
+     * Frame handlers with their own dedicated reader thread should wait for this call before
+     * starting it, so that thread does not race {@link #sendHeader()} over the TLS handshake.
+     */
+    void startProcessing();
 
-    }
+    /**
+     * Callback to signal the end of the negotiation phase.
+     */
+    void finishConnectionNegotiation();
 
     /** Cap inbound frame payloads, applied once frame_max is negotiated. */
-    default void setFrameMax(int frameMax) {
-
-    }
+    void setFrameMax(int frameMax);
 
     /**
      * Read a {@link Frame} from the underlying data connection.

src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java

@@ -332,6 +332,11 @@ public void initialize(AMQConnection connection) {
       this.handler.connection = connection;
     }
 
+    @Override
+    public void startProcessing() {
+      // no op
+    }
+
     @Override
     public void finishConnectionNegotiation() {
       maybeRemoveHandler(HANDLER_PROTOCOL_VERSION_MISMATCH);

src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java

@@ -61,6 +61,8 @@ public class SocketFrameHandler implements FrameHandler {
     private volatile int framePayloadLimit;
     private final IntSupplier payloadLimitSupplier;
 
+    private volatile AMQConnection connection;
+
     /** Time to linger before closing the socket forcefully. */
     public static final int SOCKET_CLOSING_TIMEOUT = 1;
 
@@ -193,7 +195,17 @@ public void sendHeader() throws IOException {
 
     @Override
     public void initialize(AMQConnection connection) {
-        connection.startMainLoop();
+        this.connection = connection;
+    }
+
+    @Override
+    public void startProcessing() {
+        this.connection.startMainLoop();
+    }
+
+    @Override
+    public void finishConnectionNegotiation() {
+      // no op
     }
 
     @Override

src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java

@@ -83,6 +83,16 @@ public void initialize(AMQConnection connection) {
         state.setConnection(connection);
     }
 
+    @Override
+    public void startProcessing() {
+        // no op
+    }
+
+    @Override
+    public void finishConnectionNegotiation() {
+        // no op
+    }
+
     @Override
     public void setFrameMax(int frameMax) {
         this.state.setFrameMax(frameMax);

src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -231,7 +231,7 @@ public Frame readFrame() throws IOException {
             return null; // simulate a socket timeout
         }
 
-        public void sendHeader() throws IOException {
+        public void sendHeader() {
             _numHeadersSent++;
         }
 
@@ -240,6 +240,21 @@ public void initialize(AMQConnection connection) {
             connection.startMainLoop();
         }
 
+        @Override
+        public void startProcessing() {
+
+        }
+
+        @Override
+        public void finishConnectionNegotiation() {
+
+        }
+
+        @Override
+        public void setFrameMax(int frameMax) {
+
+        }
+
         public void setTimeout(int timeoutMs) throws SocketException {
             this.timeout = timeoutMs;
         }

src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java

@@ -226,6 +226,21 @@ public void initialize(AMQConnection connection) {
             connection.startMainLoop();
         }
 
+        @Override
+        public void startProcessing() {
+
+        }
+
+        @Override
+        public void finishConnectionNegotiation() {
+
+        }
+
+        @Override
+        public void setFrameMax(int frameMax) {
+
+        }
+
         public void setTimeout(int timeoutMs) {
             // no need to implement this: don't bother changing the timeout
         }