Implement basic NIO support

Fixes #11

Author: acogoluegnes

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -108,6 +108,9 @@ public class ConnectionFactory implements Cloneable {
 
     private MetricsCollector metricsCollector;
 
+    private boolean nio = false;
+    private FrameHandlerFactory frameHandlerFactory;
+
     /** @return the default host to use for connections */
     public String getHost() {
         return host;
@@ -642,7 +645,15 @@ public MetricsCollector getMetricsCollector() {
     }
 
     protected FrameHandlerFactory createFrameHandlerFactory() throws IOException {
-        return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);
+        if(nio) {
+            if(this.frameHandlerFactory == null) {
+                this.frameHandlerFactory = new SocketChannelFrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);
+            }
+            return this.frameHandlerFactory;
+        } else {
+            return new FrameHandlerFactory(connectionTimeout, factory, socketConf, isSSL(), this.shutdownExecutor);
+        }
+
     }
 
     /**
@@ -1019,4 +1030,8 @@ public void setNetworkRecoveryInterval(int networkRecoveryInterval) {
     public void setNetworkRecoveryInterval(long networkRecoveryInterval) {
         this.networkRecoveryInterval = networkRecoveryInterval;
     }
+
+    public void setNio(boolean nio) {
+        this.nio = nio;
+    }
 }

src/main/java/com/rabbitmq/client/impl/Frame.java

@@ -21,8 +21,11 @@
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
+import java.lang.reflect.Field;
 import java.math.BigDecimal;
 import java.net.SocketTimeoutException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
 import java.sql.Timestamp;
 import java.util.Date;
 import java.util.Map;
@@ -123,6 +126,31 @@ public static Frame readFrom(DataInputStream is) throws IOException {
         return new Frame(type, channel, payload);
     }
 
+    public static Frame readFrom(SocketChannel socketChannel, ByteBuffer buffer) throws IOException {
+        // FIXME make frame read better
+        int type;
+        int channel;
+
+        type = buffer.get() & 0xff;
+
+        // FIXME check not a version mismatch
+
+        int ch1 = buffer.get() & 0xff;
+        int ch2 = buffer.get() & 0xff;
+
+        channel = (ch1 << 8) + (ch2 << 0);
+        int payloadSize = buffer.getInt();
+
+        byte[] payload = new byte[payloadSize];
+        buffer.get(payload);
+        int frameEndMarker = buffer.get() & 0xff;
+        if (frameEndMarker != AMQP.FRAME_END) {
+            throw new MalformedFrameException("Bad frame end marker: " + frameEndMarker);
+        }
+
+        return new Frame(type, channel, payload);
+    }
+
     /**
      * Private API - A protocol version mismatch is detected by checking the
      * three next bytes if a frame type of (int)'A' is read from an input
@@ -197,6 +225,24 @@ public void writeTo(DataOutputStream os) throws IOException {
         os.write(AMQP.FRAME_END);
     }
 
+    public void writeTo(SocketChannel socketChannel, ByteBuffer buffer) throws IOException {
+        buffer.put((byte) type);
+        buffer.put((byte) ((channel >>> 8) & 0xFF));
+        buffer.put((byte) ((channel >>> 0) & 0xFF));
+
+        if(accumulator != null) {
+            buffer.putInt(accumulator.size());
+            buffer.put(accumulator.toByteArray());
+        } else {
+            buffer.putInt(payload.length);
+            buffer.put(payload);
+        }
+        buffer.put((byte) AMQP.FRAME_END);
+
+        buffer.flip();
+        while(buffer.hasRemaining() && socketChannel.write(buffer) != -1);
+    }
+
     /**
      * Public API - retrieves the frame payload
      */

src/main/java/com/rabbitmq/client/impl/FrameHandlerFactory.java

@@ -28,7 +28,7 @@
 public class FrameHandlerFactory {
     private final int connectionTimeout;
     private final SocketFactory factory;
-    private final SocketConfigurator configurator;
+    protected final SocketConfigurator configurator;
     private final ExecutorService shutdownExecutor;
     private final boolean ssl;
 

src/main/java/com/rabbitmq/client/impl/SocketChannelFrameHandler.java

@@ -0,0 +1,107 @@
+// 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;
+
+import com.rabbitmq.client.ConnectionFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.SocketException;
+import java.util.concurrent.TimeUnit;
+
+/**
+ *
+ */
+public class SocketChannelFrameHandler implements FrameHandler {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(SocketChannelFrameHandler.class);
+
+    private final SocketChannelFrameHandlerState state;
+
+    private volatile int readTimeout = ConnectionFactory.DEFAULT_HEARTBEAT * 1000;
+
+    public SocketChannelFrameHandler(SocketChannelFrameHandlerState state) {
+        this.state = state;
+    }
+
+    @Override
+    public InetAddress getLocalAddress() {
+        return state.getChannel().socket().getLocalAddress();
+    }
+
+    @Override
+    public int getLocalPort() {
+        return state.getChannel().socket().getLocalPort();
+    }
+
+    @Override
+    public InetAddress getAddress() {
+        return state.getChannel().socket().getInetAddress();
+    }
+
+    @Override
+    public int getPort() {
+        return state.getChannel().socket().getPort();
+    }
+
+    @Override
+    public void setTimeout(int timeoutMs) throws SocketException {
+        state.getChannel().socket().setSoTimeout(timeoutMs);
+        this.readTimeout = timeoutMs;
+    }
+
+    @Override
+    public int getTimeout() throws SocketException {
+        return state.getChannel().socket().getSoTimeout();
+    }
+
+    @Override
+    public void sendHeader() throws IOException {
+        state.setSendHeader(true);
+    }
+
+    @Override
+    public Frame readFrame() throws IOException {
+        try {
+            return state.getReadQueue().poll(readTimeout, TimeUnit.MILLISECONDS);
+        } catch (InterruptedException e) {
+            throw new IOException("Timeout while polling read queue", e);
+        }
+    }
+
+    @Override
+    public void writeFrame(Frame frame) throws IOException {
+        state.write(frame);
+    }
+
+    @Override
+    public void flush() throws IOException {
+
+    }
+
+    @Override
+    public void close() {
+        state.getReadSelectionKey().cancel();
+        state.getWriteSelectionKey().cancel();
+        try {
+            state.getChannel().close();
+        } catch (IOException e) {
+            LOGGER.error("Error while closing SocketChannel", e);
+        }
+    }
+}