Set default continuation timeout to 10 minutes

Fixes #219

Author: acogoluegnes

src/main/java/com/rabbitmq/client/ChannelRpcTimeoutException.java renamed to src/main/java/com/rabbitmq/client/ChannelContinuationTimeoutException.java

@@ -4,13 +4,13 @@
 import java.util.concurrent.TimeoutException;
 
 /**
- * Exception thrown when a channel times out on a RPC call.
+ * Exception thrown when a channel times out on a continuation during a RPC call.
  * @since 4.1.0
  */
-public class ChannelRpcTimeoutException extends IOException {
+public class ChannelContinuationTimeoutException extends IOException {
 
     /**
-     * The channel that performed the RPC.
+     * The channel that performed the call.
      */
     private final Object channel;
 
@@ -19,7 +19,7 @@ public class ChannelRpcTimeoutException extends IOException {
      */
     private final Method method;
 
-    public ChannelRpcTimeoutException(TimeoutException cause, Object channel, Method method) {
+    public ChannelContinuationTimeoutException(TimeoutException cause, Object channel, Method method) {
         super(cause);
         this.channel = channel;
         this.method = method;

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

@@ -33,6 +33,8 @@
 import java.util.*;
 import java.util.concurrent.*;
 
+import static java.util.concurrent.TimeUnit.*;
+
 /**
  * Convenience "factory" class to facilitate opening a {@link Connection} to an AMQP broker.
  */
@@ -73,8 +75,8 @@ public class ConnectionFactory implements Cloneable {
      *  zero means wait indefinitely */
     public static final int    DEFAULT_SHUTDOWN_TIMEOUT = 10000;
 
-    /** The default timeout for RPC calls in channels: no timeout */
-    public static final int    DEFAULT_CHANNEL_RPC_TIMEOUT = -1;
+    /** The default continuation timeout for RPC calls in channels: 10 minutes */
+    public static final int    DEFAULT_CHANNEL_RPC_TIMEOUT = (int) MINUTES.toMillis(10);
 
     private static final String PREFERRED_TLS_PROTOCOL = "TLSv1.2";
 
@@ -120,7 +122,7 @@ public class ConnectionFactory implements Cloneable {
     private SSLContext sslContext;
 
     /**
-     * RPC timeout.
+     * Continuation timeout on RPC calls.
      * @since 4.1.0
      */
     private int channelRpcTimeout = DEFAULT_CHANNEL_RPC_TIMEOUT;
@@ -1091,11 +1093,14 @@ public void useBlockingIo() {
     }
 
     /**
-     * Set the timeout for RPC calls in channels.
-     * Default is no timeout.
+     * Set the continuation timeout for RPC calls in channels.
+     * Default is 10 minutes. 0 means no timeout.
      * @param channelRpcTimeout
      */
     public void setChannelRpcTimeout(int channelRpcTimeout) {
+        if(channelRpcTimeout < 0) {
+            throw new IllegalArgumentException("Timeout cannot be less than 0");
+        }
         this.channelRpcTimeout = channelRpcTimeout;
     }
 

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

@@ -39,7 +39,7 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(AMQChannel.class);
 
-    private static final int NO_RPC_TIMEOUT = ConnectionFactory.DEFAULT_CHANNEL_RPC_TIMEOUT;
+    private static final int NO_RPC_TIMEOUT = 0;
 
     /**
      * Protected; used instead of synchronizing on the channel itself,
@@ -74,6 +74,9 @@ public abstract class AMQChannel extends ShutdownNotifierComponent {
     public AMQChannel(AMQConnection connection, int channelNumber) {
         this._connection = connection;
         this._channelNumber = channelNumber;
+        if(connection.getChannelRpcTimeout() < 0) {
+            throw new IllegalArgumentException("Continuation timeout on RPC calls cannot be less than 0");
+        }
         this._rpcTimeout = connection.getChannelRpcTimeout();
     }
 
@@ -235,6 +238,7 @@ private AMQCommand privateRpc(Method m)
         // until the connection's reader-thread throws the reply over
         // the fence or the RPC times out (if enabled)
         if(_rpcTimeout == NO_RPC_TIMEOUT) {
+            System.out.println("passe");
             return k.getReply();
         } else {
             try {
@@ -247,7 +251,7 @@ private AMQCommand privateRpc(Method m)
                 } catch(Exception ex) {
                     LOGGER.warn("Error while cleaning timed out channel RPC: {}", ex.getMessage());
                 }
-                throw new ChannelRpcTimeoutException(e, this, m);
+                throw new ChannelContinuationTimeoutException(e, this, m);
             }
         }
     }

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

@@ -225,6 +225,9 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
         this.heartbeatExecutor = params.getHeartbeatExecutor();
         this.shutdownExecutor = params.getShutdownExecutor();
         this.threadFactory = params.getThreadFactory();
+        if(params.getChannelRpcTimeout() < 0) {
+            throw new IllegalArgumentException("Continuation timeout on RPC calls cannot be less than 0");
+        }
         this.channelRpcTimeout = params.getChannelRpcTimeout();
 
         this._channel0 = new AMQChannel(this, 0) {

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

@@ -15,7 +15,7 @@
 
 package com.rabbitmq.client.test;
 
-import com.rabbitmq.client.ChannelRpcTimeoutException;
+import com.rabbitmq.client.ChannelContinuationTimeoutException;
 import com.rabbitmq.client.Command;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.impl.AMQChannel;
@@ -66,7 +66,7 @@ public class AMQChannelTest {
         try {
             channel.rpc(method);
             fail("Should time out and throw an exception");
-        } catch(ChannelRpcTimeoutException e) {
+        } catch(ChannelContinuationTimeoutException e) {
             // OK
             assertThat((DummyAmqChannel) e.getChannel(), is(channel));
             assertThat(e.getMethod(), is(method));

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

@@ -99,6 +99,16 @@ public class AMQConnectionTest {
         cf.setHandshakeTimeout(7000);
     }
 
+    @Test public void negativeRpcTimeoutIsForbidden() {
+        ConnectionFactory cf = TestUtils.connectionFactory();
+        try {
+            cf.setChannelRpcTimeout(-10);
+            fail("expected an exception");
+        } catch (IllegalArgumentException _ignored) {
+            // expected
+        }
+    }
+
     /** Check the AMQConnection does send exactly 1 initial header, and deal correctly with
      * the frame handler throwing an exception when we try to read data
      */

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

@@ -71,10 +71,9 @@ public void tearDown() throws Exception {
             connection.start();
             Channel channel = connection.createChannel();
             try {
-
                 channel.queueDeclare();
                 fail("Should time out and throw an exception");
-            } catch(ChannelRpcTimeoutException e) {
+            } catch(ChannelContinuationTimeoutException e) {
                 // OK
                 assertThat((Channel) e.getChannel(), is(channel));
                 assertThat(e.getMethod(), instanceOf(AMQP.Queue.Declare.class));