NO-JIRA: [Java Perf Test Framework] Ensure that the shutdown of the Broker during a performance test run is detected by the framework and it will exit

* Currently the clients sit awaiting their shutdown latches forever

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

3 files changed, 78 insertions, 1 deletions

diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java
index 75b1e9050a..cf9820979f 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java
@@ -60,6 +60,15 @@ public class Client
      */
     public void start()
     {
+        _clientJmsDelegate.setConnectionLostListener(new ConnectionLostListener()
+        {
+            @Override
+            public void connectionLost()
+            {
+                LOGGER.warn("Client unexpectedly lost the JMS connection. Shutting down.");
+                transitToStopped();
+            }
+        });
         _clientJmsDelegate.setInstructionListener(this);
         _clientJmsDelegate.sendRegistrationMessage();
         _state.set(ClientState.READY);
@@ -67,7 +76,13 @@ public class Client
 
     public void stop()
     {
+        _clientJmsDelegate.setConnectionLostListener(null);
         _clientJmsDelegate.sendResponseMessage(new Response(_clientJmsDelegate.getClientName(), CommandType.STOP_CLIENT, null));
+        transitToStopped();
+    }
+
+    private void transitToStopped()
+    {
         _state.set(ClientState.STOPPED);
         _latch.countDown();
     }
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConnectionLostListener.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConnectionLostListener.java
new file mode 100644
index 0000000000..9acf80fb4d
--- /dev/null
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConnectionLostListener.java
@@ -0,0 +1,26 @@
+/*
+ * 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.disttest.client;
+
+
+public interface ConnectionLostListener
+{
+    void connectionLost();
+}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
index bb9d0327a8..348f398da4 100644
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
+++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java
@@ -29,6 +29,7 @@ import javax.jms.ConnectionFactory;
 import javax.jms.ConnectionMetaData;
 import javax.jms.DeliveryMode;
 import javax.jms.Destination;
+import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageConsumer;
@@ -46,6 +47,7 @@ import org.apache.commons.lang.builder.ToStringStyle;
 import org.apache.qpid.disttest.DistributedTestConstants;
 import org.apache.qpid.disttest.DistributedTestException;
 import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.client.ConnectionLostListener;
 import org.apache.qpid.disttest.client.MessageProvider;
 import org.apache.qpid.disttest.message.Command;
 import org.apache.qpid.disttest.message.CreateConnectionCommand;
@@ -305,7 +307,7 @@ public class ClientJmsDelegate
                         jmsConsumer = session.createDurableSubscriber(topic, subscription);
 
                         _testSubscriptions.put(subscription, session);
-                        LOGGER.debug("created durable suscription " + subscription + " to topic " + topic);
+                        LOGGER.debug("created durable subscription " + subscription + " to topic " + topic);
                     }
                     else
                     {
@@ -676,4 +678,38 @@ public class ClientJmsDelegate
     {
         _testMessageProviders.put(command.getProviderName(), new MessageProvider(command.getMessageProperties()));
     }
+
+    public void setConnectionLostListener(final ConnectionLostListener connectionLostListener)
+    {
+        try
+        {
+            _controllerConnection.setExceptionListener(new ExceptionListener()
+                {
+                @Override
+                public void onException(final JMSException exception)
+                {
+                    LOGGER.warn("Caught ", exception);
+
+                    if (connectionLostListener != null)
+                    {
+                        try
+                        {
+                            _controllerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE).close();
+                        }
+                        catch (JMSException e)
+                        {
+                            LOGGER.warn("Unable to create/close a new session, assuming the connection is lost ", exception);
+
+                            connectionLostListener.connectionLost();
+                        }
+                    }
+
+                }
+            });
+        }
+        catch (JMSException e)
+        {
+            // ignore
+        }
+    }
 }