Support underflow handling without thread sleep

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

@@ -200,4 +200,9 @@ private void handleProtocolVersionMismatch() throws IOException {
         }
         throw x;
     }
+
+    //Indicates ssl underflow state - means that cipherBuffer should aggregate next chunks of bytes
+    public boolean isUnderflowHandlingEnabled() {
+        return false;
+    }
 }

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

@@ -27,6 +27,7 @@
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 
+
 /**
  *
  */
@@ -176,11 +177,14 @@ void endWriteSequence() {
 
     void prepareForReadSequence() throws IOException {
         if(ssl) {
-            cipherIn.clear();
-            plainIn.clear();
+            if (!frameBuilder.isUnderflowHandlingEnabled()) {
+                cipherIn.clear();
+                cipherIn.flip();
+            }
 
-            cipherIn.flip();
+            plainIn.clear();
             plainIn.flip();
+
         } else {
             NioHelper.read(channel, plainIn);
             plainIn.flip();
@@ -189,6 +193,15 @@ void prepareForReadSequence() throws IOException {
 
     boolean continueReading() throws IOException {
         if(ssl) {
+            if (frameBuilder.isUnderflowHandlingEnabled()) {
+                int bytesRead = NioHelper.read(channel, cipherIn);
+                if (bytesRead == 0) {
+                    return false;
+                } else {
+                    cipherIn.flip();
+                    return true;
+                }
+            }
             if (!plainIn.hasRemaining() && !cipherIn.hasRemaining()) {
                 // need to try to read something
                 cipherIn.clear();

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

@@ -22,6 +22,7 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
 
+
 /**
  * Sub-class of {@link FrameBuilder} that unwraps crypted data from the network.
  * @since 4.4.0
@@ -32,6 +33,8 @@ public class SslEngineFrameBuilder extends FrameBuilder {
 
     private final ByteBuffer cipherBuffer;
 
+    private boolean isUnderflowHandlingEnabled = false;
+
     public SslEngineFrameBuilder(SSLEngine sslEngine, ByteBuffer plainIn, ByteBuffer cipherIn, ReadableByteChannel channel) {
         super(channel, plainIn);
         this.sslEngine = sslEngine;
@@ -40,12 +43,14 @@ public SslEngineFrameBuilder(SSLEngine sslEngine, ByteBuffer plainIn, ByteBuffer
 
     @Override
     protected boolean somethingToRead() throws IOException {
-        if (applicationBuffer.hasRemaining()) {
+        if (applicationBuffer.hasRemaining() && !isUnderflowHandlingEnabled) {
             return true;
         } else {
             applicationBuffer.clear();
 
-            while (true) {
+            boolean underflowHandling = false;
+
+            try {
                 SSLEngineResult result = sslEngine.unwrap(cipherBuffer, applicationBuffer);
                 switch (result.getStatus()) {
                     case OK:
@@ -59,19 +64,23 @@ protected boolean somethingToRead() throws IOException {
                         throw new SSLException("buffer overflow in read");
                     case BUFFER_UNDERFLOW:
                         cipherBuffer.compact();
-                        int read = NioHelper.read(channel, cipherBuffer);
-                        if (read == 0) {
-                            return false;
-                        }
-                        cipherBuffer.flip();
-                        break;
+                        underflowHandling = true;
+                        return false;
                     case CLOSED:
                         throw new SSLException("closed in read");
                     default:
                         throw new IllegalStateException("Invalid SSL status: " + result.getStatus());
-                    }
+                }
+            } finally {
+                isUnderflowHandlingEnabled = underflowHandling;
             }
+
+            return false;
         }
     }
 
+    @Override
+    public boolean isUnderflowHandlingEnabled() {
+        return isUnderflowHandlingEnabled;
+    }
 }