Merge pull request #331 from rabbitmq/rabbitmq-java-client-319

Get rid of read retry in NIO

Author: michaelklishin

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

@@ -0,0 +1,203 @@
+// Copyright (c) 2007-Present Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+
+package com.rabbitmq.client.impl.nio;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.MalformedFrameException;
+import com.rabbitmq.client.impl.Frame;
+
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+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 */;
+
+    protected final ReadableByteChannel channel;
+
+    protected final ByteBuffer applicationBuffer;
+    // 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];
+    private int frameType;
+    private int frameChannel;
+    private byte[] framePayload;
+    private int bytesRead = 0;
+
+    public FrameBuilder(ReadableByteChannel channel, ByteBuffer buffer) {
+        this.channel = channel;
+        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
+     * @see Frame#readFrom(DataInputStream)
+     */
+    public Frame readFrame() throws IOException {
+        while (somethingToRead()) {
+            if (bytesRead == 0) {
+                // type
+                frameType = readFromBuffer();
+                if (frameType == 'A') {
+                    handleProtocolVersionMismatch();
+                }
+            } else if (bytesRead == 1) {
+                // channel 1/2
+                frameBuffer[0] = readFromBuffer();
+            } else if (bytesRead == 2) {
+                // channel 2/2
+                frameChannel = (frameBuffer[0] << 8) + (readFromBuffer() << 0);
+            } else if (bytesRead == 3) {
+                // payload size 1/4
+                frameBuffer[0] = readFromBuffer();
+            } else if (bytesRead == 4) {
+                // payload size 2/4
+                frameBuffer[1] = readFromBuffer();
+            } else if (bytesRead == 5) {
+                // payload size 3/4
+                frameBuffer[2] = readFromBuffer();
+            } else if (bytesRead == 6) {
+                // payload size 4/4
+                int framePayloadSize = ((frameBuffer[0] << 24) + (frameBuffer[1] << 16) + (frameBuffer[2] << 8) + (readFromBuffer() << 0));
+                framePayload = new byte[framePayloadSize];
+            } else if (bytesRead >= PAYLOAD_OFFSET && bytesRead < framePayload.length + PAYLOAD_OFFSET) {
+                framePayload[bytesRead - PAYLOAD_OFFSET] = (byte) readFromBuffer();
+            } else if (bytesRead == framePayload.length + PAYLOAD_OFFSET) {
+                int frameEndMarker = readFromBuffer();
+                if (frameEndMarker != AMQP.FRAME_END) {
+                    throw new MalformedFrameException("Bad frame end marker: " + frameEndMarker);
+                }
+                bytesRead = 0;
+                return new Frame(frameType, frameChannel, framePayload);
+            } else {
+                throw new IllegalStateException("Number of read bytes incorrect: " + bytesRead);
+            }
+            bytesRead++;
+        }
+        return null;
+    }
+
+    /**
+     * 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 = NioHelper.read(channel, applicationBuffer);
+            applicationBuffer.flip();
+            if (read > 0) {
+                return true;
+            } else {
+                return false;
+            }
+        } else {
+            return true;
+        }
+    }
+
+    private int readFromBuffer() {
+        return applicationBuffer.get() & 0xff;
+    }
+
+    /**
+     * Handle a protocol version mismatch.
+     * @return
+     * @throws IOException
+     * @see Frame#protocolVersionMismatch(DataInputStream)
+     */
+    private void handleProtocolVersionMismatch() throws IOException {
+        // Probably an AMQP.... header indicating a version mismatch
+        // Otherwise meaningless, so try to read the version,
+        // and throw an exception, whether we read the version
+        // okay or not.
+        // Try to read everything from the network, this header
+        // is small and should never require several network reads.
+        byte[] expectedBytes = new byte[] { 'M', 'Q', 'P' };
+        int expectedBytesCount = 0;
+        while (somethingToRead() && expectedBytesCount < 3) {
+            // We expect the letters M, Q, P in that order: generate an informative error if they're not found
+            int nextByte = readFromBuffer();
+            if (nextByte != expectedBytes[expectedBytesCount]) {
+                throw new MalformedFrameException("Invalid AMQP protocol header from server: expected character " +
+                    expectedBytes[expectedBytesCount] + ", got " + nextByte);
+            }
+            expectedBytesCount++;
+        }
+
+        if (expectedBytesCount != 3) {
+            throw new MalformedFrameException("Invalid AMQP protocol header from server: read only "
+                + (expectedBytesCount + 1) + " byte(s) instead of 4");
+        }
+
+        int[] signature = new int[4];
+
+        for (int i = 0; i < 4; i++) {
+            if (somethingToRead()) {
+                signature[i] = readFromBuffer();
+            } else {
+                throw new MalformedFrameException("Invalid AMQP protocol header from server");
+            }
+        }
+
+        MalformedFrameException x;
+
+        if (signature[0] == 1 &&
+            signature[1] == 1 &&
+            signature[2] == 8 &&
+            signature[3] == 0) {
+            x = new MalformedFrameException("AMQP protocol version mismatch; we are version " +
+                AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR + "-" + AMQP.PROTOCOL.REVISION +
+                ", server is 0-8");
+        } else {
+            String sig = "";
+            for (int i = 0; i < 4; i++) {
+                if (i != 0)
+                    sig += ",";
+                sig += signature[i];
+            }
+
+            x = new MalformedFrameException("AMQP protocol version mismatch; we are version " +
+                AMQP.PROTOCOL.MAJOR + "-" + AMQP.PROTOCOL.MINOR + "-" + AMQP.PROTOCOL.REVISION +
+                ", server sent signature " + sig);
+        }
+        throw x;
+    }
+}

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

@@ -25,6 +25,10 @@
  */
 public class HeaderWriteRequest implements WriteRequest {
 
+    public static final WriteRequest SINGLETON = new HeaderWriteRequest();
+
+    private HeaderWriteRequest() { }
+
     @Override
     public void handle(DataOutputStream outputStream) throws IOException {
         outputStream.write("AMQP".getBytes("US-ASCII"));

src/main/java/com/rabbitmq/client/impl/nio/HeaderWriteRequest.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,27 +143,27 @@ public void run() {
                                     continue;
                                 }
 
-                                DataInputStream inputStream = state.inputStream;
-
                                 state.prepareForReadSequence();
 
                                 while (state.continueReading()) {
-                                    Frame frame = Frame.readFrom(inputStream);
-
-                                    try {
-                                        boolean noProblem = state.getConnection().handleReadFrame(frame);
-                                        if (noProblem && (!state.getConnection().isRunning() || state.getConnection().hasBrokerInitiatedShutdown())) {
-                                            // looks like the frame was Close-Ok or Close
-                                            dispatchShutdownToConnection(state);
+                                    final Frame frame = state.frameBuilder.readFrame();
+
+                                    if (frame != null) {
+                                        try {
+                                            boolean noProblem = state.getConnection().handleReadFrame(frame);
+                                            if (noProblem && (!state.getConnection().isRunning() || state.getConnection().hasBrokerInitiatedShutdown())) {
+                                                // looks like the frame was Close-Ok or Close
+                                                dispatchShutdownToConnection(state);
+                                                key.cancel();
+                                                break;
+                                            }
+                                        } catch (Throwable ex) {
+                                            // problem during frame processing, tell connection, and
+                                            // we can stop for this channel
+                                            handleIoError(state, ex);
                                             key.cancel();
                                             break;
                                         }
-                                    } catch (Throwable ex) {
-                                        // problem during frame processing, tell connection, and
-                                        // we can stop for this channel
-                                        handleIoError(state, ex);
-                                        key.cancel();
-                                        break;
                                     }
                                 }
 

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 {