Improve ByteBuffer#flip logic when using TLS

Fixes #307

Author: acogoluegnes

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

@@ -66,16 +66,15 @@ public int read() throws IOException {
                 }
 
                 int bytesRead = NioHelper.read(channel, cipherIn);
-                if (bytesRead > 0) {
-                    cipherIn.flip();
-                } else {
+                // see https://github.com/rabbitmq/rabbitmq-java-client/issues/307
+                if (bytesRead <= 0) {
                     bytesRead = NioHelper.retryRead(channel, cipherIn);
                     if(bytesRead <= 0) {
                         throw new IllegalStateException("Should be reading something from the network");
                     }
-                    // see https://github.com/rabbitmq/rabbitmq-java-client/issues/307
-                    cipherIn.flip();
                 }
+                cipherIn.flip();
+
                 plainIn.clear();
                 result = sslEngine.unwrap(cipherIn, plainIn);
 

src/test/java/com/rabbitmq/client/test/ssl/NioTlsUnverifiedConnection.java

@@ -95,11 +95,14 @@ public void configure(SSLEngine sslEngine) throws IOException {
         }
     }
 
-    @Test public void largeMessage() throws Exception {
-        CountDownLatch latch = new CountDownLatch(1);
-        connection = basicGetBasicConsume(connection, "tls.nio.queue", latch, 1 * 1000 * 1000);
-        boolean messagesReceived = latch.await(5, TimeUnit.SECONDS);
-        assertTrue("Message has not been received", messagesReceived);
+    @Test public void messageSize() throws Exception {
+        int [] sizes = new int [] {100, 1000, 10 * 1000, 1 * 1000 * 1000, 5 * 1000 * 1000};
+        for(int size : sizes) {
+            CountDownLatch latch = new CountDownLatch(1);
+            connection = basicGetBasicConsume(connection, "tls.nio.queue", latch, size);
+            boolean messagesReceived = latch.await(5, TimeUnit.SECONDS);
+            assertTrue("Message has not been received", messagesReceived);
+        }
     }
 
     private Connection basicGetBasicConsume(Connection connection, String queue, final CountDownLatch latch, int msgSize)
@@ -116,6 +119,7 @@ private Connection basicGetBasicConsume(Connection connection, String queue, fin
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                 getChannel().basicAck(envelope.getDeliveryTag(), false);
                 latch.countDown();
+                getChannel().basicCancel(consumerTag);
             }
         });