diff --git a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java b/src/main/java/com/rabbitmq/client/impl/AMQConnection.java index 61a25c20e..08143f01a 100644 --- a/src/main/java/com/rabbitmq/client/impl/AMQConnection.java +++ b/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; diff --git a/src/main/java/com/rabbitmq/client/impl/FrameHandler.java b/src/main/java/com/rabbitmq/client/impl/FrameHandler.java index ce7e8bbfb..0eba50ec3 100644 --- a/src/main/java/com/rabbitmq/client/impl/FrameHandler.java +++ b/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. diff --git a/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java b/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java index 23f298cd0..2dac67026 100644 --- a/src/main/java/com/rabbitmq/client/impl/NettyFrameHandlerFactory.java +++ b/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); diff --git a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java b/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java index 0fef5d0a6..e10b7897b 100644 --- a/src/main/java/com/rabbitmq/client/impl/SocketFrameHandler.java +++ b/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 diff --git a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java b/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java index 51d5d2f1b..c10a06883 100644 --- a/src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandler.java +++ b/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); diff --git a/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java b/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java index 7e5d54b6c..47f77731f 100644 --- a/src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java +++ b/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; } diff --git a/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java b/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java index 7b7c2b708..cf5c8533d 100644 --- a/src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java +++ b/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 }