Handle EOF properly in NIO

On MacOS, SocketChannel.read() can return EOF when the SocketChannel
is closed. On Linux, it would throw an exception. This could lead
to spinning threads on MacOS, because the NIO loop was never closed.

Fixes #11

Author: acogoluegnes

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

@@ -47,7 +47,7 @@ private int readFromBuffer(ByteBuffer buffer) {
     private static void readFromNetworkIfNecessary(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
         if(!buffer.hasRemaining()) {
             buffer.clear();
-            int read = channel.read(buffer);
+            int read = NioHelper.read(channel, buffer);
             if(read <= 0) {
                 int attempt = 0;
                 while(attempt < 3) {
@@ -56,7 +56,7 @@ private static void readFromNetworkIfNecessary(ReadableByteChannel channel, Byte
                     } catch (InterruptedException e) {
                         // ignore
                     }
-                    read = channel.read(buffer);
+                    read = NioHelper.read(channel, buffer);
                     if(read > 0) {
                         break;
                     }

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

@@ -0,0 +1,21 @@
+package com.rabbitmq.client.impl.nio;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.SocketChannel;
+
+/**
+ * Created by acogoluegnes on 04/10/2016.
+ */
+public class NioHelper {
+
+    static int read(ReadableByteChannel channel, ByteBuffer buffer) throws IOException {
+        int read = channel.read(buffer);
+        if(read < 0) {
+            throw new IOException("Channel has reached EOF");
+        }
+        return read;
+    }
+
+}

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

@@ -22,6 +22,7 @@
 
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
@@ -255,6 +256,12 @@ public void run() {
     protected void handleIoError(SocketChannelFrameHandlerState state, Throwable ex) {
         if (needToDispatchIoError(state)) {
             dispatchIoErrorToConnection(state, ex);
+        } else {
+            try {
+                state.close();
+            } catch (IOException e) {
+
+            }
         }
     }
 

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

@@ -186,7 +186,7 @@ void prepareForReadSequence() throws IOException {
             cipherIn.flip();
             plainIn.flip();
         } else {
-            channel.read(plainIn);
+            NioHelper.read(channel, plainIn);
             plainIn.flip();
         }
     }
@@ -196,7 +196,7 @@ boolean continueReading() throws IOException {
             if (!plainIn.hasRemaining() && !cipherIn.hasRemaining()) {
                 // need to try to read something
                 cipherIn.clear();
-                int bytesRead = channel.read(cipherIn);
+                int bytesRead = NioHelper.read(channel, cipherIn);
                 if (bytesRead <= 0) {
                     return false;
                 } else {
@@ -209,7 +209,7 @@ boolean continueReading() throws IOException {
         } else {
             if (!plainIn.hasRemaining()) {
                 plainIn.clear();
-                channel.read(plainIn);
+                NioHelper.read(channel, plainIn);
                 plainIn.flip();
             }
             return plainIn.hasRemaining();

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

@@ -65,7 +65,7 @@ public int read() throws IOException {
                     cipherIn.clear();
                 }
 
-                int bytesRead = channel.read(cipherIn);
+                int bytesRead = NioHelper.read(channel, cipherIn);
                 if (bytesRead > 0) {
                     cipherIn.flip();
                 } else {

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

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