Merge branch '4.3.x-stable'

Author: acogoluegnes

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

@@ -66,14 +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");
                     }
                 }
+                cipherIn.flip();
+
                 plainIn.clear();
                 result = sslEngine.unwrap(cipherIn, plainIn);
 

src/test/java/com/rabbitmq/client/test/server/PersistenceGuarantees.java

@@ -34,7 +34,8 @@ protected NioParams nioParams() {
         NioParams nioParams = super.nioParams();
         // may need a higher enqueuing timeout on slow environments
         return nioParams
-            .setWriteEnqueuingTimeoutInMs(nioParams.getWriteEnqueuingTimeoutInMs() * 2);
+            .setWriteEnqueuingTimeoutInMs(nioParams.getWriteEnqueuingTimeoutInMs() * 3)
+            .setWriteQueueCapacity(nioParams.getWriteQueueCapacity() * 2);
     }
 
     protected void declareQueue() throws IOException {

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,20 +95,31 @@ public void configure(SSLEngine sslEngine) throws IOException {
         }
     }
 
-    private Connection basicGetBasicConsume(Connection connection, String queue, final CountDownLatch latch)
+    @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)
         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) {
 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                 getChannel().basicAck(envelope.getDeliveryTag(), false);
                 latch.countDown();
+                getChannel().basicCancel(consumerTag);
             }
         });