Add retry logic in NIO read

References #11

Author: acogoluegnes

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

@@ -49,19 +49,7 @@ private static void readFromNetworkIfNecessary(ReadableByteChannel channel, Byte
             buffer.clear();
             int read = NioHelper.read(channel, buffer);
             if(read <= 0) {
-                int attempt = 0;
-                while(attempt < 3) {
-                    try {
-                        Thread.sleep(100L);
-                    } catch (InterruptedException e) {
-                        // ignore
-                    }
-                    read = NioHelper.read(channel, buffer);
-                    if(read > 0) {
-                        break;
-                    }
-                    attempt++;
-                }
+                NioHelper.retryRead(channel, buffer);
             }
             buffer.flip();
         }

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

@@ -18,7 +18,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.SocketChannel;
 
 public class NioHelper {
 
@@ -30,4 +29,22 @@ static int read(ReadableByteChannel channel, ByteBuffer buffer) throws IOExcepti
         return read;
     }
 
+    static int retryRead(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
+        int attempt = 0;
+        int read = 0;
+        while(attempt < 3) {
+            try {
+                Thread.sleep(100L);
+            } catch (InterruptedException e) {
+                // ignore
+            }
+            read = read(channel, buffer);
+            if(read > 0) {
+                break;
+            }
+            attempt++;
+        }
+        return read;
+    }
+
 }

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

@@ -69,7 +69,10 @@ public int read() throws IOException {
                 if (bytesRead > 0) {
                     cipherIn.flip();
                 } else {
-                    throw new IllegalStateException("Should be reading something from the network");
+                    bytesRead = NioHelper.retryRead(channel, cipherIn);
+                    if(bytesRead <= 0) {
+                        throw new IllegalStateException("Should be reading something from the network");
+                    }
                 }
                 plainIn.clear();
                 result = sslEngine.unwrap(cipherIn, plainIn);

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

@@ -87,7 +87,10 @@ private static SSLEngineResult.HandshakeStatus unwrap(ByteBuffer cipherIn, ByteB
                 throw new SSLException("Buffer overflow during handshake");
             case BUFFER_UNDERFLOW:
                 cipherIn.compact();
-                NioHelper.read(channel, cipherIn);
+                int read = NioHelper.read(channel, cipherIn);
+                if(read <= 0) {
+                    NioHelper.retryRead(channel, cipherIn);
+                }
                 cipherIn.flip();
                 break;
             case CLOSED: