Flip ByteBuffer after read retry

References #307

Author: acogoluegnes

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

@@ -73,6 +73,8 @@ public int read() throws IOException {
                     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();
                 }
                 plainIn.clear();
                 result = sslEngine.unwrap(cipherIn, plainIn);

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

@@ -40,6 +40,7 @@ public void openConnection()
         throws IOException, TimeoutException {
         try {
             connectionFactory.useSslProtocol();
+            connectionFactory.useNio();
         } catch (Exception ex) {
             throw new IOException(ex.toString());
         }
@@ -55,15 +56,15 @@ public void openConnection()
             }
         }
         if(connection == null) {
-            fail("Couldn't open TLS connection after 3 attemps");
+            fail("Couldn't open TLS connection after 3 attempts");
         }
 
     }
 
     @Test
     public void connectionGetConsume() throws Exception {
         CountDownLatch latch = new CountDownLatch(1);
-        connection = basicGetBasicConsume(connection, "tls.nio.queue", latch);
+        connection = basicGetBasicConsume(connection, "tls.nio.queue", latch, 100 * 1000);
         boolean messagesReceived = latch.await(5, TimeUnit.SECONDS);
         assertTrue("Message has not been received", messagesReceived);
     }
@@ -94,13 +95,20 @@ public void configure(SSLEngine sslEngine) throws IOException {
         }
     }
 
-    private Connection basicGetBasicConsume(Connection connection, String queue, final CountDownLatch latch)
+    @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);
+    }
+
+    private Connection basicGetBasicConsume(Connection connection, String queue, final CountDownLatch latch, int msgSize)
         throws IOException, TimeoutException {
         Channel channel = connection.createChannel();
         channel.queueDeclare(queue, false, false, false, null);
         channel.queuePurge(queue);
 
-        channel.basicPublish("", queue, null, new byte[100 * 1000]);
+        channel.basicPublish("", queue, null, new byte[msgSize]);
 
         channel.basicConsume(queue, false, new DefaultConsumer(channel) {