Add frame builder that supports TLS in NIO mode

References #319

Author: acogoluegnes

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

@@ -24,31 +24,50 @@
 import java.nio.channels.ReadableByteChannel;
 
 /**
- *
+ * Class to create AMQP frames from a {@link ReadableByteChannel}.
+ * Supports partial frames: a frame can be read in several attempts
+ * from the {@link NioLoop}. This can happen when the channel won't
+ * read any more bytes in the middle of a frame building. The state
+ * of the outstanding frame is saved up, and the builder will
+ * start where it left off when the {@link NioLoop} comes back to
+ * this connection.
+ * This class is not thread safe.
+ * @since 4.4.0
  */
 public class FrameBuilder {
 
     private static final int PAYLOAD_OFFSET = 1 /* type */ + 2 /* channel */ + 4 /* payload size */;
 
-    private final ReadableByteChannel channel;
+    protected final ReadableByteChannel channel;
 
-    private final ByteBuffer buffer;
+    protected final ByteBuffer applicationBuffer;
 
     private int frameType;
     private int frameChannel;
     private byte [] framePayload;
 
     private int bytesRead = 0;
 
+    // to store the bytes of the outstanding data
+    // 3 byte-long because the longest we read is an unsigned int
+    // (not need to store the latest byte)
     private final int [] frameBuffer = new int[3];
 
     public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer) {
         this.channel = channel;
-        this.buffer = buffer;
+        this.applicationBuffer = buffer;
     }
 
+    /**
+     * Read a frame from the network.
+     * This method returns null f a frame could not have been fully built from
+     * the network. The client must then retry later (typically
+     * when the channel notifies it has something to read).
+     * @return a complete frame or null if a frame couldn't have been fully built
+     * @throws IOException
+     */
     public Frame readFrame() throws IOException {
-        while(readFromNetworkIfNecessary()) {
+        while(somethingToRead()) {
             if (bytesRead == 0) {
                 // type
                 // FIXME check first byte isn't 'A' and thus a header indicating protocol version mismatch
@@ -90,18 +109,24 @@ public Frame readFrame() throws IOException {
     }
 
     private int read() throws IOException {
-        return NioHelper.read(channel, buffer);
+        return NioHelper.read(channel, applicationBuffer);
     }
 
     private int readFromBuffer() {
-        return buffer.get() & 0xff;
+        return applicationBuffer.get() & 0xff;
     }
 
-    private boolean readFromNetworkIfNecessary() throws IOException {
-        if(!buffer.hasRemaining()) {
-            buffer.clear();
+    /**
+     * Tells whether there's something to read in the application buffer or not.
+     * Tries to read from the network if necessary.
+     * @return true if there's something to read in the application buffer
+     * @throws IOException
+     */
+    protected boolean somethingToRead() throws IOException {
+        if(!applicationBuffer.hasRemaining()) {
+            applicationBuffer.clear();
             int read = read();
-            buffer.flip();
+            applicationBuffer.flip();
             if (read > 0) {
                 return true;
             } else {

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

@@ -29,29 +29,4 @@ 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;
-    }
-
-    static int readWithRetry(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
-        int bytesRead = NioHelper.read(channel, buffer);
-        if (bytesRead <= 0) {
-            bytesRead = NioHelper.retryRead(channel, buffer);
-        }
-        return bytesRead;
-    }
 }

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

@@ -20,7 +20,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -144,17 +143,10 @@ public void run() {
                                     continue;
                                 }
 
-                                DataInputStream inputStream = state.inputStream;
-
                                 state.prepareForReadSequence();
 
                                 while (state.continueReading()) {
-                                    final Frame frame;
-                                    if (state.frameBuilder == null) {
-                                        frame = Frame.readFrom(inputStream);
-                                    } else {
-                                        frame = state.frameBuilder.readFrame();
-                                    }
+                                    final Frame frame = state.frameBuilder.readFrame();
 
                                     if (frame != null) {
                                         try {

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

@@ -26,7 +26,7 @@
 
 /**
  * Parameters used to configure the NIO mode of a {@link com.rabbitmq.client.ConnectionFactory}.
- *
+ * @since 4.0.0
  */
 public class NioParams {
 

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

@@ -21,7 +21,6 @@
 import org.slf4j.LoggerFactory;
 
 import javax.net.ssl.SSLEngine;
-import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -75,8 +74,6 @@ public class SocketChannelFrameHandlerState {
 
     final DataOutputStream outputStream;
 
-    final DataInputStream inputStream;
-
     final FrameBuilder frameBuilder;
 
     public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioLoopsState, NioParams nioParams, SSLEngine sslEngine) {
@@ -96,9 +93,6 @@ public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioL
             this.outputStream = new DataOutputStream(
                 new ByteBufferOutputStream(channel, plainOut)
             );
-            this.inputStream = new DataInputStream(
-                new ByteBufferInputStream(channel, plainIn)
-            );
 
             this.frameBuilder = new FrameBuilder(channel, plainIn);
 
@@ -112,11 +106,7 @@ public SocketChannelFrameHandlerState(SocketChannel channel, NioLoopContext nioL
             this.outputStream = new DataOutputStream(
                 new SslEngineByteBufferOutputStream(sslEngine, plainOut, cipherOut, channel)
             );
-            this.inputStream = new DataInputStream(
-                new SslEngineByteBufferInputStream(sslEngine, plainIn, cipherIn, channel)
-            );
-
-            this.frameBuilder = null;
+            this.frameBuilder = new SslEngineFrameBuilder(sslEngine, plainIn, cipherIn, channel);
         }
 
     }
@@ -202,6 +192,10 @@ boolean continueReading() throws IOException {
             if (!plainIn.hasRemaining() && !cipherIn.hasRemaining()) {
                 // need to try to read something
                 cipherIn.clear();
+
+                // FIXME this logic may be simplified:
+                // flipping cipherIn and return cipherIn.hasRemaining should be enough
+
                 int bytesRead = NioHelper.read(channel, cipherIn);
                 if (bytesRead <= 0) {
                     return false;