Set a limit in write queue

The current limit may be arbitrary right now (depends on the hardware
and the scenario).

Write as much as possible in the ByteBuffer before clearing it.

Fixes #11

Author: acogoluegnes

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

@@ -15,25 +15,19 @@
 
 package com.rabbitmq.client.impl;
 
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.lang.reflect.Field;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.LongString;
+import com.rabbitmq.client.MalformedFrameException;
+
+import java.io.*;
 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;
 import java.util.List;
-
-import com.rabbitmq.client.AMQP;
-import com.rabbitmq.client.LongString;
-import com.rabbitmq.client.MalformedFrameException;
+import java.util.Map;
 
 /**
  * Represents an AMQP wire-protocol frame, with frame type, channel number, and payload bytes.
@@ -52,6 +46,8 @@ public class Frame {
     /** Frame payload (for outbound frames) */
     private final ByteArrayOutputStream accumulator;
 
+    private static final int NON_BODY_SIZE = 1 /* type */ + 2 /* channel */ + 1 /* end character */;
+
     /**
      * Constructs a frame for output with a type and a channel number and a
      * fresh accumulator waiting for payload.
@@ -243,6 +239,14 @@ public void writeTo(SocketChannel socketChannel, ByteBuffer buffer) throws IOExc
         while(buffer.hasRemaining() && socketChannel.write(buffer) != -1);
     }
 
+    public int size() {
+        if(accumulator != null) {
+            return accumulator.size() + NON_BODY_SIZE;
+        } else {
+            return payload.length + NON_BODY_SIZE;
+        }
+    }
+
     /**
      * Public API - retrieves the frame payload
      */

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

@@ -27,7 +27,6 @@
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
-import java.nio.channels.ClosedChannelException;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
 import java.nio.channels.SocketChannel;
@@ -121,7 +120,11 @@ public void run() {
                 while(true) {
                     int select;
                     if(state.statesToBeRegistered.isEmpty()) {
-                        // we can block, registration will Selector.wakeup()
+                        // we can block, registration will call Selector.wakeup()
+
+                        // FIXME check the number of keys and stop the read and write loops
+                        // if there's no read key anymore
+
                         select = selector.select();
                     } else {
                         // we cannot block, we need to select and clean cancelled keys before registration
@@ -251,8 +254,8 @@ public void run() {
 
                                     }
 
-                                } catch (ClosedChannelException e) {
-                                    LOGGER.warn("Error in read loop because of async close");
+                                } catch (Exception e) {
+                                    LOGGER.warn("Error during reading frames: "+e.getMessage());
                                 } finally {
                                     buffer.clear();
                                 }
@@ -287,7 +290,7 @@ public void run() {
                 while(true) {
                     int select;
                     if(state.statesToBeRegistered.isEmpty()) {
-                        // we can block, registration will Selector.wakeup()
+                        // we can block, registration will call Selector.wakeup()
                         select = selector.select();
                     } else {
                         // we cannot block, we need to select and clean cancelled keys before registration
@@ -330,9 +333,12 @@ public void run() {
 
                                 Frame frame;
                                 while((frame = state.getWriteQueue().poll()) != null) {
+                                    if(buffer.remaining() < frame.size()) {
+                                        buffer.clear();
+                                    }
                                     frame.writeTo(channel, buffer);
-                                    buffer.clear();
                                 }
+                                buffer.clear();
                             }
                             key.cancel();
                         }

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

@@ -1,21 +1,25 @@
 package com.rabbitmq.client.impl;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.util.Queue;
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  *
  */
 public class SocketChannelFrameHandlerState {
 
-    private final SocketChannel channel;
+    private static final Logger LOGGER = LoggerFactory.getLogger(SocketChannelFrameHandlerState.class);
 
-    private final Queue<Frame> writeQueue = new LinkedBlockingQueue<Frame>();
+    private final SocketChannel channel;
 
-    private final BlockingQueue<Frame> readQueue = new LinkedBlockingQueue<Frame>();
+    // FIXME find appropriate default for limit in write queue
+    private final BlockingQueue<Frame> writeQueue = new ArrayBlockingQueue<Frame>(1000);
 
     private volatile AMQConnection connection;
 
@@ -32,14 +36,6 @@ public SocketChannel getChannel() {
         return channel;
     }
 
-    public void addReadFrame(Frame frame) {
-        this.readQueue.add(frame);
-    }
-
-    public BlockingQueue<Frame> getReadQueue() {
-        return readQueue;
-    }
-
     public Queue<Frame> getWriteQueue() {
         return writeQueue;
     }
@@ -56,8 +52,12 @@ public void setSendHeader(boolean sendHeader) {
     }
 
     public void write(Frame frame) {
-        this.writeQueue.add(frame);
-        this.selectorState.registerFrameHandlerState(this, SelectionKey.OP_WRITE);
+        try {
+            this.writeQueue.put(frame);
+            this.selectorState.registerFrameHandlerState(this, SelectionKey.OP_WRITE);
+        } catch (InterruptedException e) {
+            LOGGER.warn("Thread interrupted during enqueuing frame in write queue");
+        }
     }
 
     public AMQConnection getConnection() {

src/test/java/com/rabbitmq/client/test/JavaNioTest.java

@@ -58,9 +58,10 @@ private Connection basicGetBasicConsume(ConnectionFactory connectionFactory, Str
 
         channel.basicPublish("", queue, null, "hello nio world!".getBytes("UTF-8"));
 
-        channel.basicConsume(queue, true, new DefaultConsumer(channel) {
+        channel.basicConsume(queue, false, new DefaultConsumer(channel) {
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                getChannel().basicAck(envelope.getDeliveryTag(), false);
                 latch.countDown();
             }
         });