QPID-3449: adds ability to override the session timeout via JVM properties qpid.sync_op_timeout and the legacy amqj.default_syncwrite_timeout, across all protocol versions

Applied patch by Oleksandr Rudyy <orudyy@gmail.com> and myself.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1169773 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 101 insertions, 5 deletions

diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
index 624cf67593..208496ec41 100644
--- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
+++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java
@@ -47,6 +47,7 @@ import org.apache.qpid.client.state.AMQStateManager;
 import org.apache.qpid.client.state.StateWaiter;
 import org.apache.qpid.client.state.listener.SpecificMethodFrameListener;
 import org.apache.qpid.codec.AMQCodecFactory;
+import org.apache.qpid.configuration.ClientProperties;
 import org.apache.qpid.framing.AMQBody;
 import org.apache.qpid.framing.AMQDataBlock;
 import org.apache.qpid.framing.AMQFrame;
@@ -163,7 +164,9 @@ public class AMQProtocolHandler implements ProtocolEngine
     private FailoverException _lastFailoverException;
 
     /** Defines the default timeout to use for synchronous protocol commands. */
-    private final long DEFAULT_SYNC_TIMEOUT = Long.getLong("amqj.default_syncwrite_timeout", 1000 * 30);
+    private final long DEFAULT_SYNC_TIMEOUT = Long.getLong(ClientProperties.QPID_SYNC_OP_TIMEOUT,
+                                                           Long.getLong(ClientProperties.AMQJ_DEFAULT_SYNCWRITE_TIMEOUT,
+                                                                        ClientProperties.DEFAULT_SYNC_OPERATION_TIMEOUT));
 
     /** Object to lock on when changing the latch */
     private Object _failoverLatchChange = new Object();
diff --git a/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java b/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
index 36f9a1ae2d..62ded5b2d8 100644
--- a/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
+++ b/java/common/src/main/java/org/apache/qpid/configuration/ClientProperties.java
@@ -90,7 +90,17 @@ public class ClientProperties
 
     public static final String QPID_VERIFY_CLIENT_ID = "qpid.verify_client_id";
 
-    private static ClientProperties _instance = new ClientProperties();
+    /**
+     * System properties to change the default timeout used during
+     * synchronous operations.
+     */
+    public static final String QPID_SYNC_OP_TIMEOUT = "qpid.sync_op_timeout";
+    public static final String AMQJ_DEFAULT_SYNCWRITE_TIMEOUT = "amqj.default_syncwrite_timeout";
+
+    /**
+     * A default timeout value for synchronous operations
+     */
+    public static final int DEFAULT_SYNC_OPERATION_TIMEOUT = 60000;
 
     /*
     public static final QpidProperty<Boolean>  IGNORE_SET_CLIENTID_PROP_NAME =
diff --git a/java/common/src/main/java/org/apache/qpid/transport/Session.java b/java/common/src/main/java/org/apache/qpid/transport/Session.java
index 4c996e261c..a838257fb6 100644
--- a/java/common/src/main/java/org/apache/qpid/transport/Session.java
+++ b/java/common/src/main/java/org/apache/qpid/transport/Session.java
@@ -30,6 +30,8 @@ import static org.apache.qpid.transport.Session.State.DETACHED;
 import static org.apache.qpid.transport.Session.State.NEW;
 import static org.apache.qpid.transport.Session.State.OPEN;
 import static org.apache.qpid.transport.Session.State.RESUMING;
+
+import org.apache.qpid.configuration.ClientProperties;
 import org.apache.qpid.transport.network.Frame;
 import static org.apache.qpid.transport.util.Functions.mod;
 import org.apache.qpid.transport.util.Logger;
@@ -42,7 +44,6 @@ import static org.apache.qpid.util.Serial.max;
 import static org.apache.qpid.util.Strings.toUTF8;
 
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
@@ -58,7 +59,6 @@ import java.util.concurrent.TimeUnit;
 
 public class Session extends SessionInvoker
 {
-
     private static final Logger log = Logger.get(Session.class);
 
     public enum State { NEW, DETACHED, RESUMING, OPEN, CLOSING, CLOSED }
@@ -92,7 +92,9 @@ public class Session extends SessionInvoker
     private int channel;
     private SessionDelegate delegate;
     private SessionListener listener = new DefaultSessionListener();
-    private long timeout = 60000;
+    private final long timeout = Long.getLong(ClientProperties.QPID_SYNC_OP_TIMEOUT,
+                                        Long.getLong(ClientProperties.AMQJ_DEFAULT_SYNCWRITE_TIMEOUT,
+                                                     ClientProperties.DEFAULT_SYNC_OPERATION_TIMEOUT));
     private boolean autoSync = false;
 
     private boolean incomingInit;
diff --git a/java/common/src/test/java/org/apache/qpid/transport/SessionTimeoutTest.java b/java/common/src/test/java/org/apache/qpid/transport/SessionTimeoutTest.java
new file mode 100644
index 0000000000..5f1c1254a2
--- /dev/null
+++ b/java/common/src/test/java/org/apache/qpid/transport/SessionTimeoutTest.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.transport;
+
+import org.apache.qpid.configuration.ClientProperties;
+import org.apache.qpid.test.utils.QpidTestCase;
+
+
+public class SessionTimeoutTest extends QpidTestCase
+{
+    public void testSessionTimeout()
+    {
+        try
+        {
+            long timeout = 1;
+            setTestSystemProperty("qpid.sync_op_timeout", Long.toString(timeout));
+            assertSessionTimeout(timeout);
+        }
+        finally
+        {
+            revertTestSystemProperties();
+        }
+    }
+
+    public void testSessionTimeoutSetWith_amqj_default_syncwrite_timeout()
+    {
+        try
+        {
+            long timeout = 1;
+            setTestSystemProperty("amqj.default_syncwrite_timeout", Long.toString(timeout));
+            setTestSystemProperty("qpid.sync_op_timeout", null);
+            assertSessionTimeout(timeout);
+        }
+        finally
+        {
+            revertTestSystemProperties();
+        }
+    }
+
+    private void assertSessionTimeout(long timeout)
+    {
+        Session session = new TestSession(null, null, 0);
+        long startTime = System.currentTimeMillis();
+        try
+        {
+            session.awaitOpen();
+            fail("SessionTimeoutException is expected!");
+        }
+        catch (SessionException e)
+        {
+            long elapsedTime = System.currentTimeMillis() - startTime;
+            assertTrue("Expected timeout should happened in " + timeout + " ms but timeout occured in "
+                    + elapsedTime + " ms!", elapsedTime >= timeout && elapsedTime < ClientProperties.DEFAULT_SYNC_OPERATION_TIMEOUT);
+        }
+    }
+
+    class TestSession extends Session
+    {
+        public TestSession(Connection connection, Binary name, long expiry)
+        {
+            super(connection, name, expiry);
+        }
+    }
+
+}