QPID-3929: remove old perftests + integrationtests + junit-toolkit modules and associated files

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

26 files changed, 0 insertions, 6594 deletions

diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java b/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
deleted file mode 100644
index d94ca5592b..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- *
- * 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.client.message;
-
-import javax.jms.JMSException;
-import javax.jms.Session;
-import javax.jms.ObjectMessage;
-import javax.jms.StreamMessage;
-import javax.jms.BytesMessage;
-import javax.jms.TextMessage;
-import javax.jms.DeliveryMode;
-import javax.jms.Destination;
-
-public class TestMessageFactory
-{
-    private static final String MESSAGE_DATA_BYTES = "-message payload-message paylaod-message payload-message paylaod";
-
-    public static TextMessage newTextMessage(Session session, int size) throws JMSException
-    {
-        return session.createTextMessage(createMessagePayload(size));
-    }
-
-    public static TextMessage newJMSTextMessage(Session session, int size, String encoding) throws JMSException
-    {
-
-        TextMessage message = session.createTextMessage();
-        message.clearBody();
-        message.setText(createMessagePayload(size));
-        return message;
-    }
-
-    public static BytesMessage newBytesMessage(Session session, int size) throws JMSException
-    {
-        BytesMessage message = session.createBytesMessage();
-        message.writeUTF(createMessagePayload(size));
-        return message;
-    }
-
-    public static StreamMessage newStreamMessage(Session session, int size) throws JMSException
-    {
-        StreamMessage message = session.createStreamMessage();
-        message.writeString(createMessagePayload(size));
-        return message;
-    }
-
-    public static ObjectMessage newObjectMessage(Session session, int size) throws JMSException
-    {
-        if (size == 0)
-        {
-            return session.createObjectMessage();
-        }
-        else
-        {
-            return session.createObjectMessage(createMessagePayload(size));
-        }
-    }
-
-    /**
-     * Creates an ObjectMessage with given size and sets the JMS properties (JMSReplyTo and DeliveryMode)
-     * @param session
-     * @param replyDestination
-     * @param size
-     * @param persistent
-     * @return the new ObjectMessage
-     * @throws JMSException
-     */
-    public static ObjectMessage newObjectMessage(Session session, Destination replyDestination, int size, boolean persistent) throws JMSException
-    {
-        ObjectMessage msg = newObjectMessage(session, size);
-
-        // Set the messages persistent delivery flag.
-        msg.setJMSDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
-
-        // Ensure that the temporary reply queue is set as the reply to destination for the message.
-        if (replyDestination != null)
-        {
-            msg.setJMSReplyTo(replyDestination);
-        }
-
-        return msg;
-    }
-
-    public static String createMessagePayload(int size)
-    {
-        StringBuffer buf = new StringBuffer(size);
-        int count = 0;
-        while (count <= (size - MESSAGE_DATA_BYTES.length()))
-        {
-            buf.append(MESSAGE_DATA_BYTES);
-            count += MESSAGE_DATA_BYTES.length();
-        }
-        if (count < size)
-        {
-            buf.append(MESSAGE_DATA_BYTES, 0, size - count);
-        }
-
-        return buf.toString();
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/config/AMQConnectionFactoryInitialiser.java b/qpid/java/perftests/src/main/java/org/apache/qpid/config/AMQConnectionFactoryInitialiser.java
deleted file mode 100644
index 76fd318625..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/config/AMQConnectionFactoryInitialiser.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- *
- * 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.config;
-
-import org.apache.qpid.client.AMQConnectionFactory;
-import org.apache.qpid.client.AMQConnectionURL;
-import org.apache.qpid.config.ConnectionFactoryInitialiser;
-import org.apache.qpid.config.ConnectorConfig;
-import org.apache.qpid.jms.ConnectionURL;
-import org.apache.qpid.url.URLSyntaxException;
-
-import javax.jms.ConnectionFactory;
-
-class AMQConnectionFactoryInitialiser implements ConnectionFactoryInitialiser
-{
-    public ConnectionFactory getFactory(ConnectorConfig config)
-    {
-        try
-        {
-            final ConnectionURL connectionUrl = new AMQConnectionURL(ConnectionURL.AMQ_PROTOCOL + 
-                    "://guest:guest@/test_path?brokerlist='tcp://" + config.getHost() + ":" + config.getPort() + "'");
-            return new AMQConnectionFactory(connectionUrl);
-        }
-        catch (URLSyntaxException e)
-        {
-            throw new RuntimeException("Problem building URL", e);
-        }
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/config/AbstractConfig.java b/qpid/java/perftests/src/main/java/org/apache/qpid/config/AbstractConfig.java
deleted file mode 100644
index 14db74438f..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/config/AbstractConfig.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- *
- * 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.config;
-
-public abstract class AbstractConfig
-{
-    public boolean setOptions(String[] argv)
-    {
-        try
-        {
-            for(int i = 0; i < argv.length - 1; i += 2)
-            {
-                String key = argv[i];
-                String value = argv[i+1];
-                setOption(key, value);
-            }
-            return true;
-        }
-        catch(Exception e)
-        {
-            System.out.println(e.getMessage());
-            }
-        return false;
-    }
-    
-    protected int parseInt(String msg, String i)
-    {
-        try
-        {
-            return Integer.parseInt(i);
-        }
-        catch(NumberFormatException e)
-        {
-            throw new RuntimeException(msg + ": " + i, e);
-        }
-    }
-
-    protected long parseLong(String msg, String i)
-    {
-        try
-        {
-            return Long.parseLong(i);
-        }
-        catch(NumberFormatException e)
-        {
-            throw new RuntimeException(msg + ": " + i, e);
-        }
-    }
-
-    public abstract void setOption(String key, String value);
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/config/ConnectionFactoryInitialiser.java b/qpid/java/perftests/src/main/java/org/apache/qpid/config/ConnectionFactoryInitialiser.java
deleted file mode 100644
index a9984eb09a..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/config/ConnectionFactoryInitialiser.java
+++ /dev/null
@@ -1,29 +0,0 @@
-/*
- *
- * 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.config;
-
-import javax.jms.ConnectionFactory;
-import javax.jms.JMSException;
-
-public interface ConnectionFactoryInitialiser
-{
-    public ConnectionFactory getFactory(ConnectorConfig config) throws JMSException;
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/config/Connector.java b/qpid/java/perftests/src/main/java/org/apache/qpid/config/Connector.java
deleted file mode 100644
index 81541a3260..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/config/Connector.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- *
- * 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.config;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-
-public class Connector
-{
-    public Connection createConnection(ConnectorConfig config) throws Exception
-    {
-        return getConnectionFactory(config).createConnection();
-    }
-
-    ConnectionFactory getConnectionFactory(ConnectorConfig config) throws Exception
-    {
-        String factory = config.getFactory();
-        if(factory == null)
-        {
-            factory = AMQConnectionFactoryInitialiser.class.getName();
-        }
-        System.out.println("Using " + factory);
-        return ((ConnectionFactoryInitialiser) Class.forName(factory).newInstance()).getFactory(config);
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/config/ConnectorConfig.java b/qpid/java/perftests/src/main/java/org/apache/qpid/config/ConnectorConfig.java
deleted file mode 100644
index b120ed3f12..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/config/ConnectorConfig.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- *
- * 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.config;
-
-public interface ConnectorConfig
-{
-    public String getHost();
-    public int getPort();
-    public String getFactory();
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/config/JBossConnectionFactoryInitialiser.java b/qpid/java/perftests/src/main/java/org/apache/qpid/config/JBossConnectionFactoryInitialiser.java
deleted file mode 100644
index a0248a8f79..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/config/JBossConnectionFactoryInitialiser.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- *
- * 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.config;
-
-import org.apache.qpid.config.ConnectionFactoryInitialiser;
-import org.apache.qpid.config.ConnectorConfig;
-import org.apache.qpid.client.JMSAMQException;
-
-import javax.jms.ConnectionFactory;
-import javax.jms.JMSException;
-import javax.management.MBeanServerConnection;
-import javax.management.ObjectName;
-import javax.management.MBeanException;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-import javax.naming.NameNotFoundException;
-import java.util.Hashtable;
-
-public class JBossConnectionFactoryInitialiser implements ConnectionFactoryInitialiser
-{
-    public ConnectionFactory getFactory(ConnectorConfig config) throws JMSException
-    {
-        ConnectionFactory cf = null;
-        InitialContext ic = null;
-        Hashtable ht = new Hashtable();
-        ht.put(InitialContext.INITIAL_CONTEXT_FACTORY, "org.jnp.interfaces.NamingContextFactory");
-        String jbossHost = System.getProperty("jboss.host", "eqd-lxamq01");
-        String jbossPort = System.getProperty("jboss.port", "1099");
-        ht.put(InitialContext.PROVIDER_URL, "jnp://" + jbossHost + ":" + jbossPort);
-        ht.put(InitialContext.URL_PKG_PREFIXES, "org.jboss.naming:org.jnp.interfaces");
-
-        try
-        {
-            ic = new InitialContext(ht);
-            if (!doesDestinationExist("topictest.messages", ic))
-            {
-                deployTopic("topictest.messages", ic);
-            }
-            if (!doesDestinationExist("topictest.control", ic))
-            {
-                deployTopic("topictest.control", ic);
-            }
-
-            cf = (ConnectionFactory) ic.lookup("/ConnectionFactory");
-            return cf;
-        }
-        catch (NamingException e)
-        {
-            throw new JMSAMQException("Unable to lookup object: " + e, e);
-        }
-        catch (Exception e)
-        {
-            throw new JMSAMQException("Error creating topic: " + e, e);
-        }
-    }
-
-    private boolean doesDestinationExist(String name, InitialContext ic) throws Exception
-    {
-        try
-        {
-            ic.lookup("/" + name);
-        }
-        catch (NameNotFoundException e)
-        {
-            return false;
-        }
-        return true;
-    }
-
-    private void deployTopic(String name, InitialContext ic) throws Exception
-    {
-        MBeanServerConnection mBeanServer = lookupMBeanServerProxy(ic);
-
-        ObjectName serverObjectName = new ObjectName("jboss.messaging:service=ServerPeer");
-
-        String jndiName = "/" + name;
-        try
-        {
-            mBeanServer.invoke(serverObjectName, "createTopic",
-                               new Object[]{name, jndiName},
-                               new String[]{"java.lang.String", "java.lang.String"});
-        }
-        catch (MBeanException e)
-        {
-            System.err.println("Error: " + e);
-            System.err.println("Cause: " + e.getCause());
-        }
-    }
-
-    private MBeanServerConnection lookupMBeanServerProxy(InitialContext ic) throws NamingException
-    {
-        return (MBeanServerConnection) ic.lookup("jmx/invoker/RMIAdaptor");
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Config.java b/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Config.java
deleted file mode 100644
index 5b6169ed2d..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Config.java
+++ /dev/null
@@ -1,243 +0,0 @@
-/*
- *
- * 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.oldtopic;
-
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.config.ConnectorConfig;
-import org.apache.qpid.config.ConnectionFactoryInitialiser;
-import org.apache.qpid.config.Connector;
-import org.apache.qpid.config.AbstractConfig;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-
-class Config extends AbstractConfig implements ConnectorConfig
-{
-
-    private String host = "localhost";
-    private int port = 5672;
-    private String factory = null;
-
-    private int payload = 256;
-    private int messages = 1000;
-    private int clients = 1;
-    private int batch = 1;
-    private long delay = 1;
-    private int warmup;
-    private int ackMode= AMQSession.NO_ACKNOWLEDGE;
-    private String clientId;
-    private String subscriptionId;
-    private boolean persistent;
-
-    public Config()
-    {
-    }
-
-    int getAckMode()
-    {
-        return ackMode;
-    }
-
-    void setPayload(int payload)
-    {
-        this.payload = payload;
-    }
-
-    int getPayload()
-    {
-        return payload;
-    }
-
-    void setClients(int clients)
-    {
-        this.clients = clients;
-    }
-
-    int getClients()
-    {
-        return clients;
-    }
-
-    void setMessages(int messages)
-    {
-        this.messages = messages;
-    }
-
-    int getMessages()
-    {
-        return messages;
-    }
-
-    public String getHost()
-    {
-        return host;
-    }
-
-    public void setHost(String host)
-    {
-        this.host = host;
-    }
-
-    public int getPort()
-    {
-        return port;
-    }
-
-    public String getFactory()
-    {
-        return factory;
-    }
-
-    public void setPort(int port)
-    {
-        this.port = port;
-    }
-
-    int getBatch()
-    {
-        return batch;
-    }
-
-    void setBatch(int batch)
-    {
-        this.batch = batch;
-    }
-
-    int getWarmup()
-    {
-        return warmup;
-    }
-
-    void setWarmup(int warmup)
-    {
-        this.warmup = warmup;
-    }
-
-    public long getDelay()
-    {
-        return delay;
-    }
-
-    public void setDelay(long delay)
-    {
-        this.delay = delay;
-    }
-
-    String getClientId()
-    {
-        return clientId;
-    }
-
-    String getSubscriptionId()
-    {
-        return subscriptionId;
-    }
-
-    boolean usePersistentMessages()
-    {
-        return persistent;
-    }
-
-    public void setOption(String key, String value)
-    {
-        if("-host".equalsIgnoreCase(key))
-        {
-            setHost(value);
-        }
-        else if("-port".equalsIgnoreCase(key))
-        {
-            try
-            {
-                setPort(Integer.parseInt(value));
-            }
-            catch(NumberFormatException e)
-            {
-                throw new RuntimeException("Bad port number: " + value);
-            }
-        }
-        else if("-payload".equalsIgnoreCase(key))
-        {
-            setPayload(parseInt("Bad payload size", value));
-        }
-        else if("-messages".equalsIgnoreCase(key))
-        {
-            setMessages(parseInt("Bad message count", value));
-        }
-        else if("-clients".equalsIgnoreCase(key))
-        {
-            setClients(parseInt("Bad client count", value));
-        }
-        else if("-batch".equalsIgnoreCase(key))
-        {
-            setBatch(parseInt("Bad batch count", value));
-        }
-        else if("-delay".equalsIgnoreCase(key))
-        {
-            setDelay(parseLong("Bad batch delay", value));
-        }
-        else if("-warmup".equalsIgnoreCase(key))
-        {
-            setWarmup(parseInt("Bad warmup count", value));
-        }
-        else if("-ack".equalsIgnoreCase(key))
-        {
-            ackMode = parseInt("Bad ack mode", value);
-        }
-        else if("-factory".equalsIgnoreCase(key))
-        {
-            factory = value;
-        }
-        else if("-clientId".equalsIgnoreCase(key))
-        {
-            clientId = value;
-        }
-        else if("-subscriptionId".equalsIgnoreCase(key))
-        {
-            subscriptionId = value;
-        }
-        else if("-persistent".equalsIgnoreCase(key))
-        {
-            persistent = "true".equalsIgnoreCase(value);
-        }
-        else
-        {
-            System.out.println("Ignoring unrecognised option: " + key);
-        }
-    }
-
-    static String getAckModeDescription(int ackMode)
-    {
-        switch(ackMode)
-        {
-            case AMQSession.NO_ACKNOWLEDGE: return "NO_ACKNOWLEDGE";
-            case AMQSession.AUTO_ACKNOWLEDGE: return "AUTO_ACKNOWLEDGE";
-            case AMQSession.CLIENT_ACKNOWLEDGE: return "CLIENT_ACKNOWLEDGE";
-            case AMQSession.DUPS_OK_ACKNOWLEDGE: return "DUPS_OK_ACKNOWELDGE";
-            case AMQSession.PRE_ACKNOWLEDGE: return "PRE_ACKNOWLEDGE";
-        }
-        return "AckMode=" + ackMode;
-    }
-
-    public Connection createConnection() throws Exception
-    {
-        return new Connector().createConnection(this);
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Listener.java b/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Listener.java
deleted file mode 100644
index 4732782d4c..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Listener.java
+++ /dev/null
@@ -1,141 +0,0 @@
-/*
- *
- * 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.oldtopic;
-import org.apache.log4j.*;
-import javax.jms.Connection;
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-
-public class Listener implements MessageListener
-{
-    private final Connection _connection;
-    private final MessageProducer _controller;
-    private final javax.jms.Session _session;
-    private final MessageFactory _factory;
-    private boolean init;
-    private int count;
-    private long start;
-
-    Listener(Connection connection, int ackMode) throws Exception
-    {
-        this(connection, ackMode, null);
-    }
-
-    Listener(Connection connection, int ackMode, String name) throws Exception
-    {
-        _connection = connection;
-        _session = connection.createSession(false, ackMode);
-        _factory = new MessageFactory(_session);
-
-        //register for events
-        if(name == null)
-        {
-            _factory.createTopicConsumer().setMessageListener(this);
-        }
-        else
-        {
-            _factory.createDurableTopicConsumer(name).setMessageListener(this);
-        }
-
-        _connection.start();
-
-        _controller = _factory.createControlPublisher();
-        System.out.println("Waiting for messages " +
-                Config.getAckModeDescription(ackMode)
-                + (name == null ? "" : " (subscribed with name " + name + " and client id " + connection.getClientID() + ")")
-                + "...");
-
-    }
-
-    private void shutdown()
-    {
-        try
-        {
-            _session.close();
-            _connection.stop();
-            _connection.close();
-        }
-        catch(Exception e)
-        {
-            e.printStackTrace(System.out);
-        }
-    }
-
-    private void report()
-    {
-        try
-        {
-            String msg = getReport();
-            _controller.send(_factory.createReportResponseMessage(msg));
-            System.out.println("Sent report: " + msg);
-        }
-        catch(Exception e)
-        {
-            e.printStackTrace(System.out);
-        }
-    }
-
-    private String getReport()
-    {
-        long time = (System.currentTimeMillis() - start);
-        return "Received " + count + " in " + time + "ms";
-    }
-
-    public void onMessage(Message message)
-    {
-        if(!init)
-        {
-            start = System.currentTimeMillis();
-            count = 0;
-            init = true;
-        }
-
-        if(_factory.isShutdown(message))
-        {
-            shutdown();
-        }
-        else if(_factory.isReport(message))
-        {
-            //send a report:
-            report();
-            init = false;
-        }
-        else if (++count % 100 == 0)
-        {
-            System.out.println("Received " + count + " messages.");
-        }
-    }
-
-    public static void main(String[] argv) throws Exception
-    {
-        Config config = new Config();
-        config.setOptions(argv);
-
-        Connection con = config.createConnection();
-        if(config.getClientId() != null)
-        {
-            con.setClientID(config.getClientId());
-        }
-        new Listener(con, config.getAckMode(), config.getSubscriptionId());
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/MessageFactory.java b/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/MessageFactory.java
deleted file mode 100644
index b2fbeb7e35..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/MessageFactory.java
+++ /dev/null
@@ -1,153 +0,0 @@
-/*
- *
- * 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.oldtopic;
-
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.AMQTopic;
-
-import javax.jms.*;
-
-/**
- */
-class MessageFactory
-{
-    private static final char[] DATA = "abcdefghijklmnopqrstuvwxyz".toCharArray();
-
-    private final Session _session;
-    private final Topic _topic;
-    private final Topic _control;
-    private final byte[] _payload;
-
-
-    MessageFactory(Session session) throws JMSException
-    {
-        this(session, 256);
-    }
-
-    MessageFactory(Session session, int size) throws JMSException
-    {
-        _session = session;
-/*        if(session instanceof AMQSession)
-        {
-            _topic = new AMQTopic("topictest.messages");
-            _control = new AMQTopic("topictest.control");
-        }
-        else*/
-        {
-            _topic = session.createTopic("topictest.messages");
-            _control = session.createTopic("topictest.control");
-        }
-        _payload = new byte[size];
-
-        for(int i = 0; i < size; i++)
-        {
-            _payload[i] = (byte) DATA[i % DATA.length];
-        }
-    }
-
-    Topic getTopic()
-    {
-        return _topic;
-    }
-
-    Message createEventMessage() throws JMSException
-    {
-        BytesMessage msg = _session.createBytesMessage();
-        msg.writeBytes(_payload);
-        return msg;
-    }
-
-    Message createShutdownMessage() throws JMSException
-    {
-        return _session.createTextMessage("SHUTDOWN");
-    }
-
-    Message createReportRequestMessage() throws JMSException
-    {
-        return _session.createTextMessage("REPORT");
-    }
-
-    Message createReportResponseMessage(String msg) throws JMSException
-    {
-        return _session.createTextMessage(msg);
-    }
-
-    boolean isShutdown(Message m)
-    {
-        return checkText(m, "SHUTDOWN");
-    }
-
-    boolean isReport(Message m)
-    {
-        return checkText(m, "REPORT");
-    }
-
-    Object getReport(Message m)
-    {
-        try
-        {
-            return ((TextMessage) m).getText();
-        }
-        catch (JMSException e)
-        {
-            e.printStackTrace(System.out);
-            return e.toString();
-        }
-    }
-
-    MessageConsumer createTopicConsumer() throws Exception
-    {
-        return _session.createConsumer(_topic);
-    }
-
-    MessageConsumer createDurableTopicConsumer(String name) throws Exception
-    {
-        return _session.createDurableSubscriber(_topic, name);
-    }
-
-    MessageConsumer createControlConsumer() throws Exception
-    {
-        return _session.createConsumer(_control);
-    }
-
-    MessageProducer createTopicPublisher() throws Exception
-    {
-        return _session.createProducer(_topic);
-    }
-
-    MessageProducer createControlPublisher() throws Exception
-    {
-        return _session.createProducer(_control);
-    }
-
-    private static boolean checkText(Message m, String s)
-    {
-        try
-        {
-            return m instanceof TextMessage && ((TextMessage) m).getText().equals(s);
-        }
-        catch (JMSException e)
-        {
-            e.printStackTrace(System.out);
-            return false;
-        }
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Publisher.java b/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Publisher.java
deleted file mode 100644
index 841fcc63ad..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/oldtopic/Publisher.java
+++ /dev/null
@@ -1,178 +0,0 @@
-/*
- *
- * 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.oldtopic;
-
-import javax.jms.*;
-
-public class Publisher implements MessageListener
-{
-    private final Object _lock = new Object();
-    private final Connection _connection;
-    private final Session _session;
-    private final MessageFactory _factory;
-    private final MessageProducer _publisher;
-    private int _count;
-
-    Publisher(Connection connection, int size, int ackMode, boolean persistent) throws Exception
-    {
-        _connection = connection;
-        _session = _connection.createSession(false, ackMode);
-        _factory = new MessageFactory(_session, size);
-        _publisher = _factory.createTopicPublisher();
-        _publisher.setDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
-        System.out.println("Publishing " + (persistent ? "persistent" : "non-persistent") + " messages of " + size + " bytes, " + Config.getAckModeDescription(ackMode) + ".");
-    }
-
-    private void test(Config config) throws Exception
-    {
-        test(config.getBatch(), config.getDelay(), config.getMessages(), config.getClients(), config.getWarmup());
-    }
-
-    private void test(int batches, long delay, int msgCount, int consumerCount, int warmup) throws Exception
-    {
-        _factory.createControlConsumer().setMessageListener(this);
-        _connection.start();
-
-        if(warmup > 0)
-        {
-            System.out.println("Runing warmup (" + warmup + " msgs)");
-            long time = batch(warmup, consumerCount);
-            System.out.println("Warmup completed in " + time + "ms");
-        }
-
-        long[] times = new long[batches];
-        for(int i = 0; i < batches; i++)
-        {
-            if(i > 0)
-            {
-                Thread.sleep(delay*1000);
-            }
-            times[i] = batch(msgCount, consumerCount);
-            System.out.println("Batch " + (i+1) + " of " + batches + " completed in " + times[i] + " ms.");
-        }
-
-        long min = min(times);
-        long max = max(times);
-        System.out.println("min: " + min + ", max: " + max + " avg: " + avg(times, min, max));
-
-        //request shutdown
-        _publisher.send(_factory.createShutdownMessage());
-
-        _connection.stop();
-        _connection.close();
-    }
-
-    private long batch(int msgCount, int consumerCount) throws Exception
-    {
-        _count = consumerCount;
-        long start = System.currentTimeMillis();
-        publish(msgCount);
-        waitForCompletion(consumerCount);
-        return System.currentTimeMillis() - start;
-    }
-
-    private void publish(int count) throws Exception
-    {
-
-        //send events
-        for (int i = 0; i < count; i++)
-        {
-            _publisher.send(_factory.createEventMessage());
-            if ((i + 1) % 100 == 0)
-            {
-                System.out.println("Sent " + (i + 1) + " messages");
-            }
-        }
-
-        //request report
-        _publisher.send(_factory.createReportRequestMessage());
-    }
-
-    private void waitForCompletion(int consumers) throws Exception
-    {
-        System.out.println("Waiting for completion...");
-        synchronized (_lock)
-        {
-            while (_count > 0)
-            {
-                _lock.wait();
-            }
-        }
-    }
-
-
-    public void onMessage(Message message)
-    {
-        System.out.println("Received report " + _factory.getReport(message) + " " + --_count + " remaining");
-        if (_count == 0)
-        {
-            synchronized (_lock)
-            {
-                _lock.notify();
-            }
-        }
-    }
-
-    static long min(long[] times)
-    {
-        long min = times.length > 0 ? times[0] : 0;
-        for(int i = 0; i < times.length; i++)
-        {
-            min = Math.min(min, times[i]);
-        }
-        return min;
-    }
-
-    static long max(long[] times)
-    {
-        long max = times.length > 0 ? times[0] : 0;
-        for(int i = 0; i < times.length; i++)
-        {
-            max = Math.max(max, times[i]);
-        }
-        return max;
-    }
-
-    static long avg(long[] times, long min, long max)
-    {
-        long sum = 0;
-        for(int i = 0; i < times.length; i++)
-        {
-            sum += times[i];
-        }
-        sum -= min;
-        sum -= max;
-
-        return (sum / (times.length - 2));
-    }
-
-    public static void main(String[] argv) throws Exception
-    {
-        Config config = new Config();
-        config.setOptions(argv);
-
-        Connection con = config.createConnection();
-        int size = config.getPayload();
-        int ackMode = config.getAckMode();
-        boolean persistent = config.usePersistentMessages();
-        new Publisher(con, size, ackMode, persistent).test(config);
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingAsyncTestPerf.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingAsyncTestPerf.java
deleted file mode 100644
index b55dac45c7..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingAsyncTestPerf.java
+++ /dev/null
@@ -1,323 +0,0 @@
-/*
- *
- * 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.ping;
-
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.requestreply.PingPongProducer;
-
-import org.apache.qpid.junit.extensions.TimingController;
-import org.apache.qpid.junit.extensions.TimingControllerAware;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * PingAsyncTestPerf is a performance test that outputs multiple timings from its test method, using the timing controller
- * interface supplied by the test runner from a seperate listener thread. It differs from the {@link PingTestPerf} test
- * that it extends because it can output timings as replies are received, rather than waiting until all expected replies
- * are received. This is less 'blocky' than the tests in {@link PingTestPerf}, and provides a truer simulation of sending
- * and recieving clients working asynchronously.
- *
- * <p/><table id="crc"><caption>CRC Card</caption>
- * <tr><td> Responsibilities <th> Collaborations
- * <tr><td> Send many ping messages and output timings asynchronously on batches received.
- * </table>
- */
-public class PingAsyncTestPerf extends PingTestPerf implements TimingControllerAware
-{
-    private static Logger _logger = Logger.getLogger(PingAsyncTestPerf.class);
-
-    /** Holds the name of the property to get the test results logging batch size. */
-    public static final String TEST_RESULTS_BATCH_SIZE_PROPNAME = "batchSize";
-
-    /** Holds the default test results logging batch size. */
-    public static final int TEST_RESULTS_BATCH_SIZE_DEFAULT = 1000;
-
-    /** Used to hold the timing controller passed from the test runner. */
-    private TimingController _timingController;
-
-    /** Used to generate unique correlation ids for each test run. */
-    private AtomicLong corellationIdGenerator = new AtomicLong();
-
-    /** Holds test specifics by correlation id. This consists of the expected number of messages and the timing controler. */
-    private Map<String, PerCorrelationId> perCorrelationIds =
-        Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
-
-    /** Holds the batched results listener, that does logging on batch boundaries. */
-    private BatchedResultsListener batchedResultsListener = null;
-
-    /**
-     * Creates a new asynchronous ping performance test with the specified name.
-     *
-     * @param name The test name.
-     */
-    public PingAsyncTestPerf(String name)
-    {
-        super(name);
-
-        // Sets up the test parameters with defaults.
-        testParameters.setPropertyIfNull(TEST_RESULTS_BATCH_SIZE_PROPNAME,
-            Integer.toString(TEST_RESULTS_BATCH_SIZE_DEFAULT));
-    }
-
-    /**
-     * Compile all the tests into a test suite.
-     * @return The test suite to run. Should only contain testAsyncPingOk method.
-     */
-    public static Test suite()
-    {
-        // Build a new test suite
-        TestSuite suite = new TestSuite("Ping Performance Tests");
-
-        // Run performance tests in read committed mode.
-        suite.addTest(new PingAsyncTestPerf("testAsyncPingOk"));
-
-        return suite;
-    }
-
-    /**
-     * Accepts a timing controller from the test runner.
-     *
-     * @param timingController The timing controller to register mutliple timings with.
-     */
-    public void setTimingController(TimingController timingController)
-    {
-        _timingController = timingController;
-    }
-
-    /**
-     * Gets the timing controller passed in by the test runner.
-     *
-     * @return The timing controller passed in by the test runner.
-     */
-    public TimingController getTimingController()
-    {
-        return _timingController;
-    }
-
-    /**
-     * Sends the specified number of pings, asynchronously outputs timings on every batch boundary, and waits until
-     * all replies have been received or a time out occurs before exiting this method.
-     *
-     * @param numPings The number of pings to send.
-     * @throws Exception pass all errors out to the test harness
-     */
-    public void testAsyncPingOk(int numPings) throws Exception
-    {
-        // _logger.debug("public void testAsyncPingOk(int numPings): called");
-
-        // get prefill count to update the expected count
-        int preFill = testParameters.getPropertyAsInteger(PingPongProducer.PREFILL_PROPNAME);
-
-        // Ensure that at least one ping was requeusted.
-        if (numPings + preFill == 0)
-        {
-            _logger.error("Number of pings requested was zero.");
-            fail("Number of pings requested was zero.");
-        }
-
-        // Get the per thread test setup to run the test through.
-        PerThreadSetup perThreadSetup = threadSetup.get();
-        PingClient pingClient = perThreadSetup._pingClient;
-
-        // Advance the correlation id of messages to send, to make it unique for this run.
-        perThreadSetup._correlationId = Long.toString(corellationIdGenerator.incrementAndGet());
-        // String messageCorrelationId = perThreadSetup._correlationId;
-        // _logger.debug("messageCorrelationId = " + messageCorrelationId);
-
-
-        // Initialize the count and timing controller for the new correlation id.
-        // This perCorrelationId is only used for controlling the test.
-        // The PingClient itself uses its own perCorrelationId see in PingPongProducer
-        PerCorrelationId perCorrelationId = new PerCorrelationId();
-        TimingController tc = getTimingController().getControllerForCurrentThread();
-        perCorrelationId._tc = tc;
-        perCorrelationId._expectedCount = pingClient.getExpectedNumPings(numPings + preFill);
-        perCorrelationIds.put(perThreadSetup._correlationId, perCorrelationId);
-
-        // Must be called before pingAndWaitForReply to setup the CorrelationID.
-        // This is required because pingClient.start() will start all client threads
-        // This means that the CorrelationID must be registered before hand.
-        pingClient.setupCorrelationID(perThreadSetup._correlationId, perCorrelationId._expectedCount);
-
-        // Start the client connection if:
-        // 1) we are not in a SEND_ONLY test.
-        // 2) if we have not yet started client because messages are sitting on broker.
-        // This is either due to a preFill or a consume only test.
-        if (!testParameters.getPropertyAsBoolean(PingPongProducer.SEND_ONLY_PROPNAME) &&
-            (preFill > 0 || testParameters.getPropertyAsBoolean(PingPongProducer.CONSUME_ONLY_PROPNAME)))
-        {
-            pingClient.start();
-        }
-
-        // Send the requested number of messages, and wait until they have all been received.
-        long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
-        int numReplies = pingClient.pingAndWaitForReply(null, numPings , preFill, timeout, perThreadSetup._correlationId);
-
-        // Check that all the replies were received and log a fail if they were not.
-        if (numReplies < perCorrelationId._expectedCount)
-        {
-            System.out.println("##### " + numReplies + "replies, expected " + perCorrelationId._expectedCount + " #####");
-            perCorrelationId._tc.completeTest(false, numPings - perCorrelationId._expectedCount);
-        }
-
-        // Remove the expected count and timing controller for the message correlation id, to ensure they are cleaned up.
-        perCorrelationIds.remove(perThreadSetup._correlationId);
-    }
-
-    /**
-     * Performs test fixture creation on a per thread basis. This will only be called once for each test thread.
-     */
-    public void threadSetUp()
-    {
-        _logger.debug("public void threadSetUp(): called");
-
-        try
-        {
-            // Call the set up method in the super class. This creates a PingClient pinger.
-            super.threadSetUp();
-
-            // Create the chained message listener, only if it has not already been created.  This is set up with the
-            // batch size property, to tell it what batch size to output results on. A synchronized block is used to
-            // ensure that only one thread creates this.
-            synchronized (this)
-            {
-                if (batchedResultsListener == null)
-                {
-                    int batchSize = Integer.parseInt(testParameters.getProperty(TEST_RESULTS_BATCH_SIZE_PROPNAME));
-                    batchedResultsListener = new BatchedResultsListener(batchSize);
-                }
-            }
-
-            // Get the set up that the super class created.
-            PerThreadSetup perThreadSetup = threadSetup.get();
-
-            // Register the chained message listener on the pinger to do its asynchronous test timings from.
-            perThreadSetup._pingClient.setChainedMessageListener(batchedResultsListener);
-        }
-        catch (Exception e)
-        {
-            _logger.warn("There was an exception during per thread setup.", e);
-        }
-    }
-
-    /**
-     * BatchedResultsListener is a {@link PingPongProducer.ChainedMessageListener} that can be attached to the
-     * pinger, in order to receive notifications about every message received and the number remaining to be
-     * received. Whenever the number remaining crosses a batch size boundary this results listener outputs
-     * a test timing for the actual number of messages received in the current batch.
-     */
-    private class BatchedResultsListener implements PingPongProducer.ChainedMessageListener
-    {
-        /** The test results logging batch size. */
-        int _batchSize;
-
-        /** The latency recoreded for the batch */
-        private long _batchLatency = 0;
-
-        /**
-         * Creates a results listener on the specified batch size.
-         *
-         * @param batchSize The batch size to use.
-         */
-        public BatchedResultsListener(int batchSize)
-        {
-            _batchSize = batchSize;
-        }
-
-        /**
-         * This callback method is called from all of the pingers that this test creates. It uses the correlation id
-         * from the message to identify the timing controller for the test thread that was responsible for sending those
-         * messages.
-         *
-         * @param message        The message.
-         * @param remainingCount The count of messages remaining to be received with a particular correlation id.
-         *
-         * @throws JMSException Any underlying JMSException is allowed to fall through.
-         */
-        public void onMessage(Message message, int remainingCount, long latency) throws JMSException
-        {
-            // Record the latency for the whole batch
-            _batchLatency += latency;
-            // Check if a batch boundary has been crossed.
-            if ((remainingCount % _batchSize) == 0)
-            {
-                // Extract the correlation id from the message.
-                String correlationId = message.getJMSCorrelationID();
-
-                /*_logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount
-                    + "): called on batch boundary for message id: " + correlationId + " with thread id: "
-                    + Thread.currentThread().getId());*/
-
-                // Get the details for the correlation id and check that they are not null. They can become null
-                // if a test times out.
-                PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationId);
-                if (perCorrelationId != null)
-                {
-                    // Get the timing controller and expected count for this correlation id.
-                    TimingController tc = perCorrelationId._tc;
-                    int expected = perCorrelationId._expectedCount;
-
-                    // Calculate how many messages were actually received in the last batch. This will be the batch size
-                    // except where the number expected is not a multiple of the batch size and this is the first remaining
-                    // count to cross a batch size boundary, in which case it will be the number expected modulo the batch
-                    // size.
-                    int receivedInBatch = ((expected - remainingCount) < _batchSize) ? (expected % _batchSize) : _batchSize;
-
-                    // Register a test result for the correlation id.
-                    try
-                    {
-                        // Record the total latency for the batch.
-                        // if batchSize=1 then this will just be the message latency
-                        tc.completeTest(true, receivedInBatch, null, _batchSize == 1 ? latency : _batchLatency);
-                        // Reset latency
-                        _batchLatency = 0;
-                    }
-                    catch (InterruptedException e)
-                    {
-                        // Ignore this. It means the test runner wants to stop as soon as possible.
-                        _logger.warn("Got InterruptedException.", e);
-                    }
-                }
-                // Else ignore, test timed out. Should log a fail here?
-            }
-        }
-    }
-
-    /**
-     * Holds state specific to each correlation id, needed to output test results. This consists of the count of
-     * the total expected number of messages, and the timing controller for the thread sending those message ids.
-     */
-    private static class PerCorrelationId
-    {
-        public int _expectedCount;
-        public TimingController _tc;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java
deleted file mode 100644
index dcfc67d4fc..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingClient.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
- *
- * 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.ping;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.requestreply.PingPongProducer;
-
-import javax.jms.Destination;
-
-import java.util.List;
-import java.util.Properties;
-
-/**
- * PingClient is a {@link PingPongProducer} that does not need a {@link org.apache.qpid.requestreply.PingPongBouncer}
- * to send replies to its pings. It simply listens to its own ping destinations, rather than seperate reply queues.
- * It is an all in one ping client, that produces and consumes its own pings.
- *
- * <p/>The constructor increments a count of the number of ping clients created. It is assumed that where many
- * are created they will all be run in parallel and be active in sending and consuming pings at the same time.
- * If the unique destinations flag is not set and a pub/sub ping cycle is being run, this means that they will all hear
- * pings sent by each other. The expected number of pings received will therefore be multiplied up by the number of
- * active ping clients. The {@link #getConsumersPerDestination()} method is used to supply this multiplier under these
- * conditions.
- *
- * <p/><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Create a ping producer that listens to its own pings <td> {@link PingPongProducer}
- * <tr><td> Count the number of ping producers and produce multiplier for scaling up messages expected over topic pings.
- * </table>
- */
-public class PingClient extends PingPongProducer
-{
-    /** Used for debugging. */
-    private final Logger log = Logger.getLogger(PingClient.class);
-
-    /** Used to count the number of ping clients created. */
-    private static int _pingClientCount;
-
-    /**
-     * Creates a ping producer with the specified parameters, of which there are many. See the class level comments
-     * for {@link PingPongProducer} for details. This constructor creates a connection to the broker and creates
-     * producer and consumer sessions on it, to send and recieve its pings and replies on.
-     *
-     * @param  overrides Properties containing any desired overrides to the defaults.
-     *
-     * @throws Exception Any exceptions are allowed to fall through.
-     */
-    public PingClient(Properties overrides) throws Exception
-    {
-        super(overrides);
-
-        _pingClientCount++;
-    }
-
-    /**
-     * Returns the ping destinations themselves as the reply destinations for this pinger to listen to. This has the
-     * effect of making this pinger listen to its own pings.
-     *
-     * @return The ping destinations.
-     */
-    public List<Destination> getReplyDestinations()
-    {
-        return _pingDestinations;
-    }
-
-    /**
-     * Supplies the multiplier for the number of ping clients that will hear each ping when doing pub/sub pinging.
-     *
-     * @return The scaling up of the number of expected pub/sub pings.
-     */
-    public int getConsumersPerDestination()
-    {
-        log.debug("public int getConsumersPerDestination(): called");
-
-        if (_isUnique)
-        {
-            log.debug(_noOfConsumers + " consumer per destination.");
-
-            return _noOfConsumers;
-        }
-        else
-        {
-            log.debug((_pingClientCount * _noOfConsumers) + " consumers per destination.");
-
-            return _pingClientCount * _noOfConsumers;
-        }
-    }
-
-    public int getClientCount()
-    {
-        return _pingClientCount;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java
deleted file mode 100644
index a15897c82b..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java
+++ /dev/null
@@ -1,452 +0,0 @@
-/*
- *
- * 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.ping;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.requestreply.PingPongProducer;
-import org.apache.qpid.util.CommandLineParser;
-
-import org.apache.qpid.junit.extensions.util.MathUtils;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-
-import javax.jms.*;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * PingDurableClient is a variation of the {@link PingPongProducer} ping tool. Instead of sending its pings and
- * receiving replies to them at the same time, this tool sends pings until it is signalled by some 'event' to stop
- * sending. It then waits for another signal before it re-opens a fresh connection and attempts to receive all of the
- * pings that it has succesfully sent. It is intended to be an interactive test that lets a user experiment with
- * failure conditions when using durable messaging.
- *
- * <p/>The events that can stop it from sending are input from the user on the console, failure of its connection to
- * the broker, completion of sending a specified number of messages, or expiry of a specified duration. In all cases
- * it will do its best to clean up and close the connection before opening a fresh connection to receive the pings
- * with.
- *
- * <p/>The event to re-connect and attempt to recieve the pings is input from the user on the console.
- *
- * <p/>This ping client inherits the configuration properties of its parent class ({@link PingPongProducer}) and
- * additionally accepts the following parameters:
- *
- * <p/><table><caption>Parameters</caption>
- * <tr><th> Parameter           <th> Default  <th> Comments
- * <tr><td> numMessages         <td> 100      <td> The total number of messages to send.
- * <tr><td> numMessagesToAction <td> -1       <td> The number of messages to send before taking a custom 'action'.
- * <tr><td> duration            <td> 30S      <td> The length of time to ping for. (Format dDhHmMsS, for d days, h hours,
- *                                                 m minutes and s seconds).
- * </table>
- *
- * <p/>This ping client also overrides some of the defaults of its parent class, to provide a reasonable set up
- * when no parameters are specified.
- *
- * <p/><table><caption>Parameters</caption>
- * <tr><th> Parameter        <th> Default  <th> Comments
- * <tr><td> uniqueDests      <td> false    <td> Prevents destination names being timestamped.
- * <tr><td> transacted       <td> true     <td> Only makes sense to test with transactions.
- * <tr><td> persistent       <td> true     <td> Only makes sense to test persistent.
- * <tr><td> durableDests     <td> true     <td> Should use durable queues with persistent messages.
- * <tr><td> commitBatchSize  <td> 10
- * <tr><td> rate             <td> 20       <td> Total default test time is 5 seconds.
- * </table>
- *
- * <p/>When a number of messages or duration is specified, this ping client will ping until the first of those limits
- * is reached. Reaching the limit will be interpreted as the first signal to stop sending, and the ping client will
- * wait for the second signal before receiving its pings.
- *
- * <p/>This class provides a mechanism for extensions to add arbitrary actions, after a particular number of messages
- * have been sent. When the number of messages equal the value set in the 'numMessagesToAction' property is method,
- * the {@link #takeAction} method is called. By default this does nothing, but extensions of this class can provide
- * custom behaviour with alternative implementations of this method (for example taking a backup).
- *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Send and receive pings.
- * <tr><td> Accept user input to signal stop sending.
- * <tr><td> Accept user input to signal start receiving.
- * <tr><td> Provide feedback on pings sent versus pings received.
- * <tr><td> Provide extension point for arbitrary action on a particular message count.
- * </table>
- */
-public class PingDurableClient extends PingPongProducer implements ExceptionListener
-{
-    private static final Logger log = Logger.getLogger(PingDurableClient.class);
-
-    public static final String NUM_MESSAGES_PROPNAME = "numMessages";
-    public static final String NUM_MESSAGES_DEFAULT = "100";
-    public static final String DURATION_PROPNAME = "duration";
-    public static final String DURATION_DEFAULT = "30S";
-    public static final String NUM_MESSAGES_TO_ACTION_PROPNAME = "numMessagesToAction";
-    public static final String NUM_MESSAGES_TO_ACTION_DEFAULT = "-1";
-
-    /** The maximum length of time to wait whilst receiving pings before assuming that no more are coming. */
-    private static final long TIME_OUT = 3000;
-
-    static
-    {
-        defaults.setProperty(NUM_MESSAGES_PROPNAME, NUM_MESSAGES_DEFAULT);
-        defaults.setProperty(DURATION_PROPNAME, DURATION_DEFAULT);
-        defaults.setProperty(UNIQUE_DESTS_PROPNAME, "false");
-        defaults.setProperty(TRANSACTED_PROPNAME, "true");
-        defaults.setProperty(PERSISTENT_MODE_PROPNAME, "true");
-        defaults.setProperty(TX_BATCH_SIZE_PROPNAME, "10");
-        defaults.setProperty(RATE_PROPNAME, "20");
-        defaults.setProperty(DURABLE_DESTS_PROPNAME, "true");
-        defaults.setProperty(NUM_MESSAGES_TO_ACTION_PROPNAME, NUM_MESSAGES_TO_ACTION_DEFAULT);
-    }
-
-    /** Specifies the number of pings to send, if larger than 0. 0 means send until told to stop. */
-    private int numMessages;
-
-    /** Holds the number of messages to send before taking triggering the action. */
-    private int numMessagesToAction;
-
-    /** Sepcifies how long to ping for, if larger than 0. 0 means send until told to stop. */
-    private long duration;
-
-    /** Used to indciate that this application should terminate. Set by the shutdown hook. */
-    private boolean terminate = false;
-
-    /**
-     * @throws Exception Any exceptions are allowed to fall through.
-     */
-    public PingDurableClient(Properties overrides) throws Exception
-    {
-        super(overrides);
-        log.debug("public PingDurableClient(Properties overrides = " + overrides + "): called");
-
-        // Extract the additional configuration parameters.
-        ParsedProperties properties = new ParsedProperties(defaults);
-        properties.putAll(overrides);
-
-        numMessages = properties.getPropertyAsInteger(NUM_MESSAGES_PROPNAME);
-        String durationSpec = properties.getProperty(DURATION_PROPNAME);
-        numMessagesToAction = properties.getPropertyAsInteger(NUM_MESSAGES_TO_ACTION_PROPNAME);
-
-        if (durationSpec != null)
-        {
-            duration = MathUtils.parseDuration(durationSpec) * 1000000;
-        }
-    }
-
-    /**
-     * Starts the ping/wait/receive process.
-     *
-     * @param args The command line arguments.
-     */
-    public static void main(String[] args)
-    {
-        try
-        {
-            // Create a ping producer overriding its defaults with all options passed on the command line.
-            Properties options =
-                CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
-            PingDurableClient pingProducer = new PingDurableClient(options);
-
-            // Create a shutdown hook to terminate the ping-pong producer.
-            Runtime.getRuntime().addShutdownHook(pingProducer.getShutdownHook());
-
-            // Ensure that the ping pong producer is registered to listen for exceptions on the connection too.
-            // pingProducer.getConnection().setExceptionListener(pingProducer);
-
-            // Run the test procedure.
-            int sent = pingProducer.send();
-            pingProducer.closeConnection();
-            pingProducer.waitForUser("Press return to begin receiving the pings.");
-            pingProducer.receive(sent);
-
-            System.exit(0);
-        }
-        catch (Exception e)
-        {
-            System.err.println(e.getMessage());
-            log.error("Top level handler caught execption.", e);
-            System.exit(1);
-        }
-    }
-
-    /**
-     * Performs the main test procedure implemented by this ping client. See the class level comment for details.
-     */
-    protected int send() throws Exception
-    {
-        log.debug("public void sendWaitReceive(): called");
-
-        log.debug("duration = " + duration);
-        log.debug("numMessages = " + numMessages);
-
-        if (duration > 0)
-        {
-            System.out.println("Sending for up to " + (duration / 1000000000f) + " seconds.");
-        }
-
-        if (_rate > 0)
-        {
-            System.out.println("Sending at " + _rate + " messages per second.");
-        }
-
-        if (numMessages > 0)
-        {
-            System.out.println("Sending up to " + numMessages + " messages.");
-        }
-
-        // Establish the connection and the message producer.
-        establishConnection(true, false);
-        _connection.start();
-
-        Message message = getTestMessage(getReplyDestinations().get(0), _messageSize, _persistent);
-
-        // Send pings until a terminating condition is received.
-        boolean endCondition = false;
-        int messagesSent = 0;
-        int messagesCommitted = 0;
-        int messagesNotCommitted = 0;
-        long start = System.nanoTime();
-
-        // Clear console in.
-        clearConsole();
-
-        while (!endCondition)
-        {
-            boolean committed = false;
-
-            try
-            {
-                committed = sendMessage(messagesSent, message) && _transacted;
-
-                messagesSent++;
-                messagesNotCommitted++;
-
-                // Keep count of the number of messsages currently committed and pending commit.
-                if (committed)
-                {
-                    log.debug("Adding " + messagesNotCommitted + " messages to the committed count.");
-                    messagesCommitted += messagesNotCommitted;
-                    messagesNotCommitted = 0;
-
-                    System.out.println("Commited: " + messagesCommitted);
-                }
-            }
-            catch (JMSException e)
-            {
-                log.debug("Got JMSException whilst sending.");
-                _publish = false;
-            }
-
-            // Perform the arbitrary action if the number of messages sent has reached the right number.
-            if (messagesSent == numMessagesToAction)
-            {
-                System.out.println("At action point, Messages sent = " + messagesSent + ", Messages Committed = "
-                    + messagesCommitted + ", Messages not Committed = " + messagesNotCommitted);
-                takeAction();
-            }
-
-            // Determine if the end condition has been met, based on the number of messages, time passed, errors on
-            // the connection or user input.
-            long now = System.nanoTime();
-
-            if ((duration != 0) && ((now - start) > duration))
-            {
-                System.out.println("Send halted because duration expired.");
-                endCondition = true;
-            }
-            else if ((numMessages != 0) && (messagesSent >= numMessages))
-            {
-                System.out.println("Send halted because # messages completed.");
-                endCondition = true;
-            }
-            else if (System.in.available() > 0)
-            {
-                System.out.println("Send halted by user input.");
-                endCondition = true;
-
-                clearConsole();
-            }
-            else if (!_publish)
-            {
-                System.out.println("Send halted by error on the connection.");
-                endCondition = true;
-            }
-        }
-
-        log.debug("messagesSent = " + messagesSent);
-        log.debug("messagesCommitted = " + messagesCommitted);
-        log.debug("messagesNotCommitted = " + messagesNotCommitted);
-
-        System.out.println("Messages sent: " + messagesSent + ", Messages Committed = " + messagesCommitted
-            + ", Messages not Committed = " + messagesNotCommitted);
-
-        return messagesSent;
-    }
-
-    protected void closeConnection()
-    {
-        // Clean up the connection.
-        try
-        {
-            close();
-        }
-        catch (JMSException e)
-        {
-            log.debug("There was an error whilst closing the connection: " + e, e);
-            System.out.println("There was an error whilst closing the connection.");
-
-            // Ignore as did best could manage to clean up.
-        }
-    }
-
-    protected void receive(int messagesSent) throws Exception
-    {
-        // Re-establish the connection and the message consumer.
-        _queueJVMSequenceID = new AtomicInteger();
-        _queueSharedID = new AtomicInteger();
-
-        establishConnection(false, true);
-        _consumer[0].setMessageListener(null);
-        _consumerConnection[0].start();
-
-        // Try to receive all of the pings that were successfully sent.
-        int messagesReceived = 0;
-        boolean endCondition = false;
-
-        while (!endCondition)
-        {
-            // Message received = _consumer.receiveNoWait();
-            Message received = _consumer[0].receive(TIME_OUT);
-            log.debug("received = " + received);
-
-            if (received != null)
-            {
-                messagesReceived++;
-            }
-
-            // Determine if the end condition has been met, based on the number of messages and time passed since last
-            // receiving a message.
-            if (received == null)
-            {
-                System.out.println("Timed out.");
-                endCondition = true;
-            }
-            else if (messagesReceived >= messagesSent)
-            {
-                System.out.println("Got all messages.");
-                endCondition = true;
-            }
-        }
-
-        // Ensure messages received are committed.
-        if (_consTransacted)
-        {
-            try
-            {
-                _consumerSession[0].commit();
-                System.out.println("Committed for all messages received.");
-            }
-            catch (JMSException e)
-            {
-                log.debug("Error during commit: " + e, e);
-                System.out.println("Error during commit.");
-                try
-                {
-                    _consumerSession[0].rollback();
-                    System.out.println("Rolled back on all messages received.");
-                }
-                catch (JMSException e2)
-                {
-                    log.debug("Error during rollback: " + e, e);
-                    System.out.println("Error on roll back of all messages received.");
-                }
-
-            }
-        }
-
-        log.debug("messagesReceived = " + messagesReceived);
-
-        System.out.println("Messages received: " + messagesReceived);
-
-        // Clean up the connection.
-        close();
-    }
-
-    /**
-     * Clears any pending input from the console.
-     */
-    private void clearConsole()
-    {
-        try
-        {
-            BufferedReader bis = new BufferedReader(new InputStreamReader(System.in));
-
-            // System.in.skip(System.in.available());
-            while (bis.ready())
-            {
-                bis.readLine();
-            }
-        }
-        catch (IOException e)
-        { }
-    }
-
-    /**
-     * Returns the ping destinations themselves as the reply destinations for this pinger to listen to. This has the
-     * effect of making this pinger listen to its own pings.
-     *
-     * @return The ping destinations.
-     */
-    public List<Destination> getReplyDestinations()
-    {
-        return _pingDestinations;
-    }
-
-    /**
-     * Gets a shutdown hook that will cleanly shut this down when it is running the ping loop. This can be registered with
-     * the runtime system as a shutdown hook. This shutdown hook sets an additional terminate flag, compared with the
-     * shutdown hook in {@link PingPongProducer}, because the publish flag is used to indicate that sending or receiving
-     * message should stop, not that the application should termiante.
-     *
-     * @return A shutdown hook for the ping loop.
-     */
-    public Thread getShutdownHook()
-    {
-        return new Thread(new Runnable()
-                {
-                    public void run()
-                    {
-                        stop();
-                        terminate = true;
-                    }
-                });
-    }
-
-    /**
-     * Performs an aribtrary action once the 'numMesagesToAction' count is reached on sending messages. This default
-     * implementation does nothing.
-     */
-    public void takeAction()
-    { }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java
deleted file mode 100644
index 5ba4004c56..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java
+++ /dev/null
@@ -1,311 +0,0 @@
-/*
- *
- * 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.ping;
-
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.requestreply.PingPongProducer;
-
-import org.apache.qpid.junit.extensions.TimingController;
-import org.apache.qpid.junit.extensions.TimingControllerAware;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * PingLatencyTestPerf is a performance test that outputs multiple timings from its test method, using the timing
- * controller interface supplied by the test runner from a seperate listener thread. It outputs round trip timings for
- * individual ping messages rather than for how long a complete batch of messages took to process. It also differs from
- * the {@link PingTestPerf} test that it extends because it can output timings as replies are received, rather than
- * waiting until all expected replies are received.
- *
- * <p/>This test does not output timings for every single ping message, as when running at high volume, writing the test
- * log for a vast number of messages would slow the testing down. Instead samples ping latency occasionally. The
- * frequency of ping sampling is set using the {@link #TEST_RESULTS_BATCH_SIZE_PROPNAME} property, to override the
- * default of every {@link #DEFAULT_TEST_RESULTS_BATCH_SIZE}.
- *
- * <p/>The size parameter logged for each individual ping is set to the size of the batch of messages that the
- * individual timed ping was taken from, rather than 1 for a single message. This is so that the total throughput
- * (messages / time) can be calculated in order to examine the relationship between throughput and latency.
- *
- * <p/><table id="crc"><caption>CRC Card</caption> <tr><td> Responsibilities <th> Collaborations <tr><td> Send many ping
- * messages and output timings for sampled individual pings. </table>
- */
-public class PingLatencyTestPerf extends PingTestPerf implements TimingControllerAware
-{
-    private static Logger _logger = Logger.getLogger(PingLatencyTestPerf.class);
-
-    /** Holds the name of the property to get the test results logging batch size. */
-    public static final String TEST_RESULTS_BATCH_SIZE_PROPNAME = "batchSize";
-
-    /** Holds the default test results logging batch size. */
-    public static final int DEFAULT_TEST_RESULTS_BATCH_SIZE = 1000;
-
-    /** Used to hold the timing controller passed from the test runner. */
-    private TimingController _timingController;
-
-    /** Used to generate unique correlation ids for each test run. */
-    private AtomicLong corellationIdGenerator = new AtomicLong();
-
-    /**
-     * Holds test specifics by correlation id. This consists of the expected number of messages and the timing
-     * controler.
-     */
-    private Map<String, PerCorrelationId> perCorrelationIds =
-        Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
-
-    /** Holds the batched results listener, that does logging on batch boundaries. */
-    private BatchedResultsListener batchedResultsListener = null;
-
-    /**
-     * Creates a new asynchronous ping performance test with the specified name.
-     *
-     * @param name The test name.
-     */
-    public PingLatencyTestPerf(String name)
-    {
-        super(name);
-
-        // Sets up the test parameters with defaults.
-        ParsedProperties.setSysPropertyIfNull(TEST_RESULTS_BATCH_SIZE_PROPNAME,
-            Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE));
-    }
-
-    /** Compile all the tests into a test suite. */
-    public static Test suite()
-    {
-        // Build a new test suite
-        TestSuite suite = new TestSuite("Ping Latency Tests");
-
-        // Run performance tests in read committed mode.
-        suite.addTest(new PingLatencyTestPerf("testPingLatency"));
-
-        return suite;
-    }
-
-    /**
-     * Accepts a timing controller from the test runner.
-     *
-     * @param timingController The timing controller to register mutliple timings with.
-     */
-    public void setTimingController(TimingController timingController)
-    {
-        _timingController = timingController;
-    }
-
-    /**
-     * Gets the timing controller passed in by the test runner.
-     *
-     * @return The timing controller passed in by the test runner.
-     */
-    public TimingController getTimingController()
-    {
-        return _timingController;
-    }
-
-    /**
-     * Sends the specified number of pings, asynchronously outputs timings on every batch boundary, and waits until all
-     * replies have been received or a time out occurs before exiting this method.
-     *
-     * @param numPings The number of pings to send.
-     */
-    public void testPingLatency(int numPings) throws Exception
-    {
-        _logger.debug("public void testPingLatency(int numPings): called");
-
-        // Ensure that at least one ping was requeusted.
-        if (numPings == 0)
-        {
-            _logger.error("Number of pings requested was zero.");
-        }
-
-        // Get the per thread test setup to run the test through.
-        PerThreadSetup perThreadSetup = threadSetup.get();
-        PingClient pingClient = perThreadSetup._pingClient;
-
-        // Advance the correlation id of messages to send, to make it unique for this run.
-        String messageCorrelationId = Long.toString(corellationIdGenerator.incrementAndGet());
-        _logger.debug("messageCorrelationId = " + messageCorrelationId);
-
-        // Initialize the count and timing controller for the new correlation id.
-        PerCorrelationId perCorrelationId = new PerCorrelationId();
-        TimingController tc = getTimingController().getControllerForCurrentThread();
-        perCorrelationId._tc = tc;
-        perCorrelationId._expectedCount = numPings;
-        perCorrelationIds.put(messageCorrelationId, perCorrelationId);
-
-        // Attach the chained message listener to the ping producer to listen asynchronously for the replies to these
-        // messages.
-        pingClient.setChainedMessageListener(batchedResultsListener);
-
-        // Generate a sample message of the specified size.
-        Message msg =
-            pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
-                testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
-                testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
-
-        // Send the requested number of messages, and wait until they have all been received.
-        long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
-        int numReplies = pingClient.pingAndWaitForReply(msg, numPings, timeout, null);
-
-        // Check that all the replies were received and log a fail if they were not.
-        if (numReplies < numPings)
-        {
-            tc.completeTest(false, 0);
-        }
-
-        // Remove the chained message listener from the ping producer.
-        pingClient.removeChainedMessageListener();
-
-        // Remove the expected count and timing controller for the message correlation id, to ensure they are cleaned up.
-        perCorrelationIds.remove(messageCorrelationId);
-    }
-
-    /** Performs test fixture creation on a per thread basis. This will only be called once for each test thread. */
-    public void threadSetUp()
-    {
-        _logger.debug("public void threadSetUp(): called");
-
-        try
-        {
-            // Call the set up method in the super class. This creates a PingClient pinger.
-            super.threadSetUp();
-
-            // Create the chained message listener, only if it has not already been created.  This is set up with the
-            // batch size property, to tell it what batch size to output results on. A synchronized block is used to
-            // ensure that only one thread creates this.
-            synchronized (this)
-            {
-                if (batchedResultsListener == null)
-                {
-                    int batchSize = Integer.parseInt(testParameters.getProperty(TEST_RESULTS_BATCH_SIZE_PROPNAME));
-                    batchedResultsListener = new BatchedResultsListener(batchSize);
-                }
-            }
-
-            // Get the set up that the super class created.
-            PerThreadSetup perThreadSetup = threadSetup.get();
-
-            // Register the chained message listener on the pinger to do its asynchronous test timings from.
-            perThreadSetup._pingClient.setChainedMessageListener(batchedResultsListener);
-        }
-        catch (Exception e)
-        {
-            _logger.warn("There was an exception during per thread setup.", e);
-        }
-    }
-
-    /**
-     * BatchedResultsListener is a {@link org.apache.qpid.requestreply.PingPongProducer.ChainedMessageListener} that can
-     * be attached to the pinger, in order to receive notifications about every message received and the number
-     * remaining to be received. Whenever the number remaining crosses a batch size boundary this results listener
-     * outputs a test timing for the actual number of messages received in the current batch.
-     */
-    private class BatchedResultsListener implements PingPongProducer.ChainedMessageListener
-    {
-        /** The test results logging batch size. */
-        int _batchSize;
-        private boolean _strictAMQP;
-
-        /**
-         * Creates a results listener on the specified batch size.
-         *
-         * @param batchSize The batch size to use.
-         */
-        public BatchedResultsListener(int batchSize)
-        {
-            _batchSize = batchSize;
-            _strictAMQP =
-                Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP,
-                        AMQSession.STRICT_AMQP_DEFAULT));
-        }
-
-        /**
-         * This callback method is called from all of the pingers that this test creates. It uses the correlation id
-         * from the message to identify the timing controller for the test thread that was responsible for sending those
-         * messages.
-         *
-         * @param message        The message.
-         * @param remainingCount The count of messages remaining to be received with a particular correlation id.
-         *
-         * @throws javax.jms.JMSException Any underlying JMSException is allowed to fall through.
-         */
-        public void onMessage(Message message, int remainingCount, long latency) throws JMSException
-        {
-            _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount + "): called");
-
-            // Check if a batch boundary has been crossed.
-            if ((remainingCount % _batchSize) == 0)
-            {
-                // Extract the correlation id from the message.
-                String correlationId = message.getJMSCorrelationID();
-
-                // Get the details for the correlation id and check that they are not null. They can become null
-                // if a test times out.
-                PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationId);
-                if (perCorrelationId != null)
-                {
-                    // Get the timing controller and expected count for this correlation id.
-                    TimingController tc = perCorrelationId._tc;
-                    int expected = perCorrelationId._expectedCount;
-
-                    // Calculate how many messages were actually received in the last batch. This will be the batch size
-                    // except where the number expected is not a multiple of the batch size and this is the first remaining
-                    // count to cross a batch size boundary, in which case it will be the number expected modulo the batch
-                    // size.
-                    int receivedInBatch = ((expected - remainingCount) < _batchSize) ? (expected % _batchSize) : _batchSize;
-
-                    // Register a test result for the correlation id.
-                    try
-                    {
-                        tc.completeTest(true, receivedInBatch, latency);
-                    }
-                    catch (InterruptedException e)
-                    {
-                        // Ignore this. It means the test runner wants to stop as soon as possible.
-                        _logger.warn("Got InterruptedException.", e);
-                    }
-                }
-                // Else ignore, test timed out. Should log a fail here?
-            }
-        }
-    }
-
-    /**
-     * Holds state specific to each correlation id, needed to output test results. This consists of the count of the
-     * total expected number of messages, and the timing controller for the thread sending those message ids.
-     */
-    private static class PerCorrelationId
-    {
-        public int _expectedCount;
-        public TimingController _tc;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java
deleted file mode 100644
index 2fe852af77..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java
+++ /dev/null
@@ -1,93 +0,0 @@
-/*
- *
- * 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.ping;
-
-import java.util.Properties;
-
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.ObjectMessage;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.message.TestMessageFactory;
-import org.apache.qpid.util.CommandLineParser;
-
-/**
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * </table>
- */
-public class PingSendOnlyClient extends PingDurableClient
-{
-    private static final Logger log = Logger.getLogger(PingSendOnlyClient.class);
-
-    public PingSendOnlyClient(Properties overrides) throws Exception
-    {
-        super(overrides);
-    }
-
-    /**
-     * Starts the ping/wait/receive process.
-     *
-     * @param args The command line arguments.
-     */
-    public static void main(String[] args)
-    {
-        try
-        {
-            // Create a ping producer overriding its defaults with all options passed on the command line.
-            Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
-            PingSendOnlyClient pingProducer = new PingSendOnlyClient(options);
-
-            // Create a shutdown hook to terminate the ping-pong producer.
-            Runtime.getRuntime().addShutdownHook(pingProducer.getShutdownHook());
-
-            // Ensure that the ping pong producer is registered to listen for exceptions on the connection too.
-            // pingProducer.getConnection().setExceptionListener(pingProducer);
-
-            // Run the test procedure.
-            pingProducer.send();
-            pingProducer.waitForUser("Press return to close connection and quit.");
-            pingProducer.closeConnection();
-
-            System.exit(0);
-        }
-        catch (Exception e)
-        {
-            System.err.println(e.getMessage());
-            log.error("Top level handler caught execption.", e);
-            System.exit(1);
-        }
-    }
-
-    public Message getTestMessage(Destination replyQueue, int messageSize, boolean persistent) throws JMSException
-    {
-        Message msg = TestMessageFactory.newTextMessage(_producerSession, messageSize);
-
-        // Timestamp the message in nanoseconds.
-        msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
-
-        return msg;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java
deleted file mode 100644
index cf16abc596..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java
+++ /dev/null
@@ -1,281 +0,0 @@
-/*
- *
- * 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.ping;
-
-import junit.framework.Assert;
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.requestreply.PingPongProducer;
-
-import org.apache.qpid.junit.extensions.AsymptoticTestCase;
-import org.apache.qpid.junit.extensions.TestThreadAware;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-import org.apache.qpid.junit.extensions.util.TestContextProperties;
-
-import javax.jms.*;
-
-/**
- * PingTestPerf is a ping test, that has been written with the intention of being scaled up to run many times
- * simultaneously to simluate many clients/producers/connections.
- *
- * <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of a single
- * full round trip ping. This test may be scaled up using a suitable JUnit test runner.
- *
- * <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a
- * temporary queue for replies. This setup is only established once for all the test repeats/threads that may be run,
- * except if the connection is lost in which case an attempt to re-establish the setup is made.
- *
- * <p/>The test cycle is: Connects to a queue, creates a temporary queue, creates messages containing a property that
- * is the name of the temporary queue, fires off a message on the original queue and waits for a response on the
- * temporary queue.
- *
- * <p/>Configurable test properties: message size, transacted or not, persistent or not. Broker connection details.
- *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * </table>
- */
-public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
-{
-    private static Logger _logger = Logger.getLogger(PingTestPerf.class);
-
-    /** Thread local to hold the per-thread test setup fields. */
-    ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
-
-    /** Holds a property reader to extract the test parameters from. */
-    protected ParsedProperties testParameters =
-        TestContextProperties.getInstance(PingPongProducer.defaults /*System.getProperties()*/);
-
-    public PingTestPerf(String name)
-    {
-        super(name);
-
-        _logger.debug("testParameters = " + testParameters);
-    }
-
-    /**
-     * Compile all the tests into a test suite.
-     * @return The test method testPingOk.
-     */
-    public static Test suite()
-    {
-        // Build a new test suite
-        TestSuite suite = new TestSuite("Ping Performance Tests");
-
-        // Run performance tests in read committed mode.
-        suite.addTest(new PingTestPerf("testPingOk"));
-
-        return suite;
-    }
-
-    public void testPingOk(int numPings) throws Exception
-    {
-        if (numPings == 0)
-        {
-            Assert.fail("Number of pings requested was zero.");
-        }
-
-        // Get the per thread test setup to run the test through.
-        PerThreadSetup perThreadSetup = threadSetup.get();
-
-        if (perThreadSetup == null)
-        {
-            Assert.fail("Could not get per thread test setup, it was null.");
-        }
-
-        // Generate a sample message. This message is already time stamped and has its reply-to destination set.
-        Message msg =
-            perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
-                testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
-                testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
-
-        // start the test
-        long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
-        int numReplies = perThreadSetup._pingClient.pingAndWaitForReply(msg, numPings, timeout, null);
-
-        // Fail the test if the timeout was exceeded.
-        if (numReplies != perThreadSetup._pingClient.getExpectedNumPings(numPings))
-        {
-            Assert.fail("The ping timed out after " + timeout + " ms. Messages Sent = " + numPings + ", MessagesReceived = "
-                + numReplies);
-        }
-    }
-
-    /** Performs test fixture creation on a per thread basis. This will only be called once for each test thread. */
-    public void threadSetUp()
-    {
-        _logger.debug("public void threadSetUp(): called");
-
-        try
-        {
-            PerThreadSetup perThreadSetup = new PerThreadSetup();
-
-            // This is synchronized because there is a race condition, which causes one connection to sleep if
-            // all threads try to create connection concurrently.
-            synchronized (this)
-            {
-                // Establish a client to ping a Destination and listen the reply back from same Destination
-                perThreadSetup._pingClient = new PingClient(testParameters);
-                perThreadSetup._pingClient.establishConnection(true, true);
-            }
-
-            // Attach the per-thread set to the thread.
-            threadSetup.set(perThreadSetup);
-        }
-        catch (Exception e)
-        {
-            _logger.warn("There was an exception during per thread setup.", e);
-        }
-    }
-
-    /**
-     * Called after all threads have completed their setup.
-     */    
-    public void postThreadSetUp()
-    {
-        _logger.debug("public void postThreadSetUp(): called");
-
-        PerThreadSetup perThreadSetup = threadSetup.get();
-        // Prefill the broker unless we are in consume only mode.
-        int preFill = testParameters.getPropertyAsInteger(PingPongProducer.PREFILL_PROPNAME);
-        if (!testParameters.getPropertyAsBoolean(PingPongProducer.CONSUME_ONLY_PROPNAME) && preFill > 0)
-        {
-            try
-            {
-                // Manually set the correlation ID to 1. This is not ideal but it is the
-                // value that the main test loop will use.
-                perThreadSetup._pingClient.pingNoWaitForReply(null, preFill, String.valueOf(perThreadSetup._pingClient.getClientCount()));
-
-                // Note with a large preFill and non-tx session the messages will be
-                // rapidly pushed in to the mina buffers. OOM's are a real risk here.
-                // Should perhaps consider using a TX session for the prefill.
-
-                long delayBeforeConsume = testParameters.getPropertyAsLong(PingPongProducer.DELAY_BEFORE_CONSUME_PROPNAME);
-
-                // Only delay if we are
-                //  not doing send only
-                //  and we have consumers
-                //  and a delayBeforeConsume
-                if (!(testParameters.getPropertyAsBoolean(PingPongProducer.SEND_ONLY_PROPNAME))
-                    && (testParameters.getPropertyAsInteger(PingPongProducer.NUM_CONSUMERS_PROPNAME) > 0)
-                    && delayBeforeConsume > 0)
-                {
-
-                    boolean verbose = testParameters.getPropertyAsBoolean(PingPongProducer.VERBOSE_PROPNAME);
-                    // Only do logging if in verbose mode.
-                    if (verbose)
-                    {
-                        if (delayBeforeConsume > 60000)
-                        {
-                            long minutes = delayBeforeConsume / 60000;
-                            long seconds = (delayBeforeConsume - (minutes * 60000)) / 1000;
-                            long ms = delayBeforeConsume - (minutes * 60000) - (seconds * 1000);
-                            _logger.info("Delaying for " + minutes + "m " + seconds + "s " + ms + "ms before starting test.");
-                        }
-                        else
-                        {
-                            _logger.info("Delaying for " + delayBeforeConsume + "ms before starting test.");
-                        }
-                    }
-
-                    Thread.sleep(delayBeforeConsume);
-
-                    if (verbose)
-                    {
-                        _logger.info("Starting Test.");
-                    }
-                }
-
-                // We can't start the client's here as the test client has not yet been configured to receieve messages.
-                // only when the test method is executed will the correlationID map be set up and ready to consume
-                // the messages we have sent here.
-            }
-            catch (Exception e)
-            {
-                _logger.warn("There was an exception during per thread setup.", e);
-            }
-        }
-        else //Only start the consumer if we are not preFilling.
-        {
-            // Start the consumers, unless we have data on the broker
-            // already this is signified by being in consume_only, we will
-            //  start the clients after setting up the correlation IDs.
-            // We should also not start the clients if we are in Send only
-            if (!testParameters.getPropertyAsBoolean(PingPongProducer.CONSUME_ONLY_PROPNAME) &&
-                !(testParameters.getPropertyAsBoolean(PingPongProducer.SEND_ONLY_PROPNAME)))
-            {
-                // Start the client connection
-                try
-                {
-                    perThreadSetup._pingClient.start();
-                }
-                catch (JMSException e)
-                {
-                    e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
-                }
-            }
-        }
-    }
-
-    /**
-     * Performs test fixture clean
-     */
-    public void threadTearDown()
-    {                                                                                       
-        _logger.debug("public void threadTearDown(): called");
-
-        try
-        {
-            // Get the per thread test fixture.
-            PerThreadSetup perThreadSetup = threadSetup.get();
-
-            // Close the pingers so that it cleans up its connection cleanly.
-            synchronized (this)
-            {
-                if ((perThreadSetup != null) && (perThreadSetup._pingClient != null))
-                {
-                    perThreadSetup._pingClient.close();
-                }
-            }
-        }
-        catch (JMSException e)
-        {
-            _logger.warn("There was an exception during per thread tear down.");
-        }
-        finally
-        {
-            // Ensure the per thread fixture is reclaimed.
-            threadSetup.remove();
-        }
-    }
-
-    protected static class PerThreadSetup
-    {
-        /**
-         * Holds the test ping client.
-         */
-        protected PingClient _pingClient;
-        protected String _correlationId;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongBouncer.java b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongBouncer.java
deleted file mode 100644
index 8e010ccf07..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongBouncer.java
+++ /dev/null
@@ -1,453 +0,0 @@
-/*
- *
- * 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.requestreply;
-
-import java.io.IOException;
-import java.net.InetAddress;
-import java.text.SimpleDateFormat;
-import java.util.Date;
-
-import javax.jms.*;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.jms.ConnectionListener;
-import org.apache.qpid.jms.Session;
-import org.apache.qpid.topic.Config;
-import org.apache.qpid.exchange.ExchangeDefaults;
-
-/**
- * PingPongBouncer is a message listener the bounces back messages to their reply to destination. This is used to return
- * ping messages generated by {@link org.apache.qpid.requestreply.PingPongProducer} but could be used for other purposes
- * too.
- *
- * <p/>The correlation id from the received message is extracted, and placed into the reply as the correlation id. Messages
- * are bounced back to their reply-to destination. The original sender of the message has the option to use either a unique
- * temporary queue or the correlation id to correlate the original message to the reply.
- *
- * <p/>There is a verbose mode flag which causes information about each ping to be output to the console
- * (info level logging, so usually console). This can be helpfull to check the bounce backs are happening but should
- * be disabled for real timing tests as writing to the console will slow things down.
- *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Bounce back messages to their reply to destination.
- * <tr><td> Provide command line invocation to start the bounce back on a configurable broker url.
- * </table>
- *
- * @todo Replace the command line parsing with a neater tool.
- *
- * @todo Make verbose accept a number of messages, only prints to console every X messages.
- */
-public class PingPongBouncer implements MessageListener
-{
-    private static final Logger _logger = Logger.getLogger(PingPongBouncer.class);
-
-    /** The default prefetch size for the message consumer. */
-    private static final int PREFETCH = 1;
-
-    /** The default no local flag for the message consumer. */
-    private static final boolean NO_LOCAL = true;
-
-    private static final String DEFAULT_DESTINATION_NAME = "ping";
-
-    /** The default exclusive flag for the message consumer. */
-    private static final boolean EXCLUSIVE = false;
-
-    /** A convenient formatter to use when time stamping output. */
-    protected static final SimpleDateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS");
-
-    /** Used to indicate that the reply generator should log timing info to the console (logger info level). */
-    private boolean _verbose = false;
-
-    /** Determines whether this bounce back client bounces back messages persistently. */
-    private boolean _persistent = false;
-
-    private Destination _consumerDestination;
-
-    /** Keeps track of the response destination of the previous message for the last reply to producer cache. */
-    private Destination _lastResponseDest;
-
-    /** The producer for sending replies with. */
-    private MessageProducer _replyProducer;
-
-    /** The consumer controlSession. */
-    private Session _consumerSession;
-
-    /** The producer controlSession. */
-    private Session _producerSession;
-
-    /** Holds the connection to the broker. */
-    private AMQConnection _connection;
-
-    /** Flag used to indicate if this is a point to point or pub/sub ping client. */
-    private boolean _isPubSub = false;
-
-    /**
-     * This flag is used to indicate that the user should be prompted to kill a broker, in order to test
-     * failover, immediately before committing a transaction.
-     */
-    protected boolean _failBeforeCommit = false;
-
-    /**
-     * This flag is used to indicate that the user should be prompted to a kill a broker, in order to test
-     * failover, immediate after committing a transaction.
-     */
-    protected boolean _failAfterCommit = false;
-
-    /**
-     * Creates a PingPongBouncer on the specified producer and consumer sessions.
-     *
-     * @param brokerDetails The addresses of the brokers to connect to.
-     * @param username        The broker username.
-     * @param password        The broker password.
-     * @param virtualpath     The virtual host name within the broker.
-     * @param destinationName The name of the queue to receive pings on
-     *                        (or root of the queue name where many queues are generated).
-     * @param persistent      A flag to indicate that persistent message should be used.
-     * @param transacted      A flag to indicate that pings should be sent within transactions.
-     * @param selector        A message selector to filter received pings with.
-     * @param verbose         A flag to indicate that message timings should be sent to the console.
-     *
-     * @throws Exception All underlying exceptions allowed to fall through. This is only test code...
-     */
-    public PingPongBouncer(String brokerDetails, String username, String password, String virtualpath,
-                           String destinationName, boolean persistent, boolean transacted, String selector, boolean verbose,
-                           boolean pubsub) throws Exception
-    {
-        // Create a client id to uniquely identify this client.
-        InetAddress address = InetAddress.getLocalHost();
-        String clientId = address.getHostName() + System.currentTimeMillis();
-        _verbose = verbose;
-        _persistent = persistent;
-        setPubSub(pubsub);
-        // Connect to the broker.
-        setConnection(new AMQConnection(brokerDetails, username, password, clientId, virtualpath));
-        _logger.info("Connected with URL:" + getConnection().toURL());
-
-        // Set up the failover notifier.
-        getConnection().setConnectionListener(new FailoverNotifier());
-
-        // Create a controlSession to listen for messages on and one to send replies on, transactional depending on the
-        // command line option.
-        _consumerSession = (Session) getConnection().createSession(transacted, Session.AUTO_ACKNOWLEDGE);
-        _producerSession = (Session) getConnection().createSession(transacted, Session.AUTO_ACKNOWLEDGE);
-
-        // Create the queue to listen for message on.
-        createConsumerDestination(destinationName);
-        MessageConsumer consumer =
-            _consumerSession.createConsumer(_consumerDestination, PREFETCH, NO_LOCAL, EXCLUSIVE, selector);
-
-        // Create a producer for the replies, without a default destination.
-        _replyProducer = _producerSession.createProducer(null);
-        _replyProducer.setDisableMessageTimestamp(true);
-        _replyProducer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
-
-        // Set this up to listen for messages on the queue.
-        consumer.setMessageListener(this);
-    }
-
-    /**
-     * Starts a stand alone ping-pong client running in verbose mode.
-     *
-     * @param args
-     */
-    public static void main(String[] args) throws Exception
-    {
-        System.out.println("Starting...");
-
-        // Display help on the command line.
-        if (args.length == 0)
-        {
-            _logger.info("Running test with default values...");
-            //usage();
-            //System.exit(0);
-        }
-
-        // Extract all command line parameters.
-        Config config = new Config();
-        config.setOptions(args);
-        String brokerDetails = config.getHost() + ":" + config.getPort();
-        String virtualpath = "test";
-        String destinationName = config.getDestination();
-        if (destinationName == null)
-        {
-            destinationName = DEFAULT_DESTINATION_NAME;
-        }
-
-        String selector = config.getSelector();
-        boolean transacted = config.isTransacted();
-        boolean persistent = config.usePersistentMessages();
-        boolean pubsub = config.isPubSub();
-        boolean verbose = true;
-
-        //String selector = null;
-
-        // Instantiate the ping pong client with the command line options and start it running.
-        PingPongBouncer pingBouncer =
-            new PingPongBouncer(brokerDetails, "guest", "guest", virtualpath, destinationName, persistent, transacted,
-                                selector, verbose, pubsub);
-        pingBouncer.getConnection().start();
-
-        System.out.println("Waiting...");
-    }
-
-    private static void usage()
-    {
-        System.err.println("Usage: PingPongBouncer \n" + "-host : broker host\n" + "-port : broker port\n"
-                           + "-destinationname : queue/topic name\n" + "-transacted : (true/false). Default is false\n"
-                           + "-persistent : (true/false). Default is false\n"
-                           + "-pubsub     : (true/false). Default is false\n" + "-selector   : selector string\n");
-    }
-
-    /**
-     * This is a callback method that is notified of all messages for which this has been registered as a message
-     * listener on a message consumer. It sends a reply (pong) to all messages it receieves on the reply to
-     * destination of the message.
-     *
-     * @param message The message that triggered this callback.
-     */
-    public void onMessage(Message message)
-    {
-        try
-        {
-            String messageCorrelationId = message.getJMSCorrelationID();
-            if (_verbose)
-            {
-                _logger.info(timestampFormatter.format(new Date()) + ": Got ping with correlation id, "
-                             + messageCorrelationId);
-            }
-
-            // Get the reply to destination from the message and check it is set.
-            Destination responseDest = message.getJMSReplyTo();
-
-            if (responseDest == null)
-            {
-                _logger.debug("Cannot send reply because reply-to destination is null.");
-
-                return;
-            }
-
-            // Spew out some timing information if verbose mode is on.
-            if (_verbose)
-            {
-                Long timestamp = message.getLongProperty("timestamp");
-
-                if (timestamp != null)
-                {
-                    long diff = System.currentTimeMillis() - timestamp;
-                    _logger.info("Time to bounce point: " + diff);
-                }
-            }
-
-            // Correlate the reply to the original.
-            message.setJMSCorrelationID(messageCorrelationId);
-
-            // Send the receieved message as the pong reply.
-            _replyProducer.send(responseDest, message);
-
-            if (_verbose)
-            {
-                _logger.info(timestampFormatter.format(new Date()) + ": Sent reply with correlation id, "
-                             + messageCorrelationId);
-            }
-
-            // Commit the transaction if running in transactional mode.
-            commitTx(_producerSession);
-        }
-        catch (JMSException e)
-        {
-            _logger.debug("There was a JMSException: " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Gets the underlying connection that this ping client is running on.
-     *
-     * @return The underlying connection that this ping client is running on.
-     */
-    public AMQConnection getConnection()
-    {
-        return _connection;
-    }
-
-    /**
-     * Sets the connection that this ping client is using.
-     *
-     * @param connection The ping connection.
-     */
-    public void setConnection(AMQConnection connection)
-    {
-        this._connection = connection;
-    }
-
-    /**
-     * Sets or clears the pub/sub flag to indiciate whether this client is pinging a queue or a topic.
-     *
-     * @param pubsub <tt>true</tt> if this client is pinging a topic, <tt>false</tt> if it is pinging a queue.
-     */
-    public void setPubSub(boolean pubsub)
-    {
-        _isPubSub = pubsub;
-    }
-
-    /**
-     * Checks whether this client is a p2p or pub/sub ping client.
-     *
-     * @return <tt>true</tt> if this client is pinging a topic, <tt>false</tt> if it is pinging a queue.
-     */
-    public boolean isPubSub()
-    {
-        return _isPubSub;
-    }
-
-    /**
-     * Convenience method to commit the transaction on the specified controlSession. If the controlSession to commit on is not
-     * a transactional controlSession, this method does nothing.
-     *
-     * <p/>If the {@link #_failBeforeCommit} flag is set, this will prompt the user to kill the broker before the
-     * commit is applied. If the {@link #_failAfterCommit} flag is set, this will prompt the user to kill the broker
-     * after the commit is applied.
-     *
-     * @throws javax.jms.JMSException If the commit fails and then the rollback fails.
-     */
-    protected void commitTx(Session session) throws JMSException
-    {
-        if (session.getTransacted())
-        {
-            try
-            {
-                if (_failBeforeCommit)
-                {
-                    _logger.debug("Failing Before Commit");
-                    doFailover();
-                }
-
-                session.commit();
-
-                if (_failAfterCommit)
-                {
-                    _logger.debug("Failing After Commit");
-                    doFailover();
-                }
-
-                _logger.debug("Session Commited.");
-            }
-            catch (JMSException e)
-            {
-                _logger.trace("JMSException on commit:" + e.getMessage(), e);
-
-                try
-                {
-                    session.rollback();
-                    _logger.debug("Message rolled back.");
-                }
-                catch (JMSException jmse)
-                {
-                    _logger.trace("JMSE on rollback:" + jmse.getMessage(), jmse);
-
-                    // Both commit and rollback failed. Throw the rollback exception.
-                    throw jmse;
-                }
-            }
-        }
-    }
-
-    /**
-     * Prompts the user to terminate the named broker, in order to test failover functionality. This method will block
-     * until the user supplied some input on the terminal.
-     *
-     * @param broker The name of the broker to terminate.
-     */
-    protected void doFailover(String broker)
-    {
-        System.out.println("Kill Broker " + broker + " now.");
-        try
-        {
-            System.in.read();
-        }
-        catch (IOException e)
-        { }
-
-        System.out.println("Continuing.");
-    }
-
-    /**
-     * Prompts the user to terminate the broker, in order to test failover functionality. This method will block
-     * until the user supplied some input on the terminal.
-     */
-    protected void doFailover()
-    {
-        System.out.println("Kill Broker now.");
-        try
-        {
-            System.in.read();
-        }
-        catch (IOException e)
-        { }
-
-        System.out.println("Continuing.");
-
-    }
-
-    private void createConsumerDestination(String name)
-    {
-        if (isPubSub())
-        {
-            _consumerDestination = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, name);
-        }
-        else
-        {
-            _consumerDestination = new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, name);
-        }
-    }
-
-    /**
-     * A connection listener that logs out any failover complete events. Could do more interesting things with this
-     * at some point...
-     */
-    public static class FailoverNotifier implements ConnectionListener
-    {
-        public void bytesSent(long count)
-        { }
-
-        public void bytesReceived(long count)
-        { }
-
-        public boolean preFailover(boolean redirect)
-        {
-            return true;
-        }
-
-        public boolean preResubscribe()
-        {
-            return true;
-        }
-
-        public void failoverComplete()
-        {
-            _logger.info("App got failover complete callback.");
-        }
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
deleted file mode 100644
index 639b0c72ba..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
+++ /dev/null
@@ -1,1818 +0,0 @@
-/*
- *
- * 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.requestreply;
-
-import org.apache.log4j.Logger;
-import org.apache.log4j.NDC;
-
-import org.apache.qpid.test.framework.TestUtils;
-
-import org.apache.qpid.junit.extensions.BatchedThrottle;
-import org.apache.qpid.junit.extensions.Throttle;
-import org.apache.qpid.junit.extensions.util.CommandLineParser;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-
-import javax.jms.*;
-import javax.naming.Context;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-
-import java.io.*;
-import java.net.InetAddress;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.*;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * PingPongProducer is a client that sends test messages, and waits for replies to these messages. The replies may
- * either be generated by another client (see {@link PingPongBouncer}, or an extension of it may be used that listens
- * to its own messages and does not send replies (see {@link org.apache.qpid.ping.PingClient}). The intention of ping
- * pong producer is that it is a swiss-army knife test client that makes almost every aspect of its behaviour
- * configurable.
- *
- * <p/>The pings are sent with a reply-to field set to a single temporary queue, which is the same for all pings. This
- * means that this class has to do some work to correlate pings with pongs; it expectes the original message correlation
- * id in the ping to be bounced back in the reply correlation id.
- *
- * <p/>This ping tool accepts a vast number of configuration options, all of which are passed in to the constructor. It
- * can ping topics or queues; ping multiple destinations; do persistent pings; send messages of any size; do pings within
- * transactions; control the number of pings to send in each transaction; limit its sending rate; and perform failover
- * testing. A complete list of accepted parameters, default values and comments on their usage is provided here:
- *
- * <p/><table><caption>Parameters</caption>
- * <tr><th> Parameter        <th> Default  <th> Comments
- * <tr><td> messageSize      <td> 0        <td> Message size in bytes. Not including any headers.
- * <tr><td> destinationName  <td> ping     <td> The root name to use to generate destination names to ping.
- * <tr><td> persistent       <td> false    <td> Determines whether peristent delivery is used.
- * <tr><td> transacted       <td> false    <td> Determines whether messages are sent/received in transactions.
- * <tr><td> broker           <td> tcp://localhost:5672 <td> Determines the broker to connect to.
- * <tr><td> virtualHost      <td> test     <td> Determines the virtual host to send all ping over.
- * <tr><td> rate             <td> 0        <td> The maximum rate (in hertz) to send messages at. 0 means no limit.
- * <tr><td> verbose          <td> false    <td> The verbose flag for debugging. Prints to console on every message.
- * <tr><td> pubsub           <td> false    <td> Whether to ping topics or queues. Uses p2p by default.
- * <tr><td> failAfterCommit  <td> false    <td> Whether to prompt user to kill broker after a commit batch.
- * <tr><td> failBeforeCommit <td> false    <td> Whether to prompt user to kill broker before a commit batch.
- * <tr><td> failAfterSend    <td> false    <td> Whether to prompt user to kill broker after a send.
- * <tr><td> failBeforeSend   <td> false    <td> Whether to prompt user to kill broker before a send.
- * <tr><td> failOnce         <td> true     <td> Whether to prompt for failover only once.
- * <tr><td> username         <td> guest    <td> The username to access the broker with.
- * <tr><td> password         <td> guest    <td> The password to access the broker with.
- * <tr><td> selector         <td> null     <td> Not used. Defines a message selector to filter pings with.
- * <tr><td> destinationCount <td> 1        <td> The number of destinations to send pings to.
- * <tr><td> numConsumers     <td> 1        <td> The number of consumers on each destination.
- * <tr><td> timeout          <td> 30000    <td> In milliseconds. The timeout to stop waiting for replies.
- * <tr><td> commitBatchSize  <td> 1        <td> The number of messages per transaction in transactional mode.
- * <tr><td> uniqueDests      <td> true     <td> Whether each receivers only listens to one ping destination or all.
- * <tr><td> durableDests     <td> false    <td> Whether or not durable destinations are used.
- * <tr><td> ackMode          <td> AUTO_ACK <td> The message acknowledgement mode. Possible values are:
- *                                               0 - SESSION_TRANSACTED
- *                                               1 - AUTO_ACKNOWLEDGE
- *                                               2 - CLIENT_ACKNOWLEDGE
- *                                               3 - DUPS_OK_ACKNOWLEDGE
- *                                               257 - NO_ACKNOWLEDGE
- *                                               258 - PRE_ACKNOWLEDGE
- * <tr><td> consTransacted   <td> false    <td> Whether or not consumers use transactions. Defaults to the same value
- *                                              as the 'transacted' option if not seperately defined.
- * <tr><td> consAckMode      <td> AUTO_ACK <td> The message acknowledgement mode for consumers. Defaults to the same
- *                                              value as 'ackMode' if not seperately defined.
- * <tr><td> maxPending       <td> 0        <td> The maximum size in bytes, of messages sent but not yet received.
- *                                              Limits the volume of messages currently buffered on the client
- *                                              or broker. Can help scale test clients by limiting amount of buffered
- *                                              data to avoid out of memory errors.
- * </table>
- *
- * <p/>This implements the Runnable interface with a run method that implements an infinite ping loop. The ping loop
- * does all its work through helper methods, so that code wishing to run a ping-pong cycle is not forced to do so by
- * starting a new thread. The command line invocation does take advantage of this ping loop. A shutdown hook is also
- * registered to terminate the ping-pong loop cleanly.
- *
- * <p/><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Provide a ping and wait for all responses cycle.
- * <tr><td> Provide command line invocation to loop the ping cycle on a configurable broker url.
- * </table>
- *
- * @todo Use read/write lock in the onmessage, not for reading writing but to make use of a shared and exlcusive lock pair.
- *       Obtain read lock on all messages, before decrementing the message count. At the end of the on message method add a
- *       block that obtains the write lock for the very last message, releases any waiting producer. Means that the last
- *       message waits until all other messages have been handled before releasing producers but allows messages to be
- *       processed concurrently, unlike the current synchronized block.
- */
-public class PingPongProducer implements Runnable, ExceptionListener
-{
-    /** Used for debugging. */
-    private static final Logger log = Logger.getLogger(PingPongProducer.class);
-
-    /** Holds the name of the property to determine whether of not client id is overridden at connection time.  */
-    public static final String OVERRIDE_CLIENT_ID_PROPNAME = "overrideClientId";
-
-    /** Holds the default value of the override client id flag. */
-    public static final String OVERRIDE_CLIENT_ID_DEAFULT = "false";
-
-    /** Holds the name of the property to define the JNDI factory name with. */
-    public static final String FACTORY_NAME_PROPNAME = "factoryName";
-
-    /** Holds the default JNDI name of the connection factory. */
-    public static final String FACTORY_NAME_DEAFULT = "local";
-
-    /** Holds the name of the property to set the JNDI initial context properties with. */
-    public static final String FILE_PROPERTIES_PROPNAME = "properties";
-
-    /** Holds the default file name of the JNDI initial context properties. */
-    public static final String FILE_PROPERTIES_DEAFULT = "perftests.properties";
-
-    /** Holds the name of the property to get the test message size from. */
-    public static final String MESSAGE_SIZE_PROPNAME = "messageSize";
-
-    /** Used to set up a default message size. */
-    public static final int MESSAGE_SIZE_DEAFULT = 0;
-
-    /** Holds the name of the property to get the ping queue name from. */
-    public static final String PING_QUEUE_NAME_PROPNAME = "destinationName";
-
-    /** Holds the name of the default destination to send pings on. */
-    public static final String PING_QUEUE_NAME_DEFAULT = "ping";
-
-    /** Holds the name of the property to get the queue name postfix from. */
-    public static final String QUEUE_NAME_POSTFIX_PROPNAME = "queueNamePostfix";
-
-    /** Holds the default queue name postfix value. */
-    public static final String QUEUE_NAME_POSTFIX_DEFAULT = "";
-
-    /** Holds the name of the property to get the test delivery mode from. */
-    public static final String PERSISTENT_MODE_PROPNAME = "persistent";
-
-    /** Holds the message delivery mode to use for the test. */
-    public static final boolean PERSISTENT_MODE_DEFAULT = false;
-
-    /** Holds the name of the property to get the test transactional mode from. */
-    public static final String TRANSACTED_PROPNAME = "transacted";
-
-    /** Holds the transactional mode to use for the test. */
-    public static final boolean TRANSACTED_DEFAULT = false;
-
-    /** Holds the name of the property to get the test consumer transacted mode from. */
-    public static final String CONSUMER_TRANSACTED_PROPNAME = "consTransacted";
-
-    /** Holds the consumer transactional mode default setting. */
-    public static final boolean CONSUMER_TRANSACTED_DEFAULT = false;
-
-    /** Holds the name of the property to get the test broker url from. */
-    public static final String BROKER_PROPNAME = "broker";
-
-    /** Holds the default broker url for the test. */
-    public static final String BROKER_DEFAULT = "tcp://localhost:5672";
-
-    /** Holds the name of the property to get the test broker virtual path. */
-    public static final String VIRTUAL_HOST_PROPNAME = "virtualHost";
-
-    /** Holds the default virtual path for the test. */
-    public static final String VIRTUAL_HOST_DEFAULT = "";
-
-    /** Holds the name of the property to get the message rate from. */
-    public static final String RATE_PROPNAME = "rate";
-
-    /** Defines the default rate (in pings per second) to send pings at. 0 means as fast as possible, no restriction. */
-    public static final int RATE_DEFAULT = 0;
-
-    /** Holds the name of the property to get the verbose mode proeprty from. */
-    public static final String VERBOSE_PROPNAME = "verbose";
-
-    /** Holds the default verbose mode. */
-    public static final boolean VERBOSE_DEFAULT = false;
-
-    /** Holds the name of the property to get the p2p or pub/sub messaging mode from. */
-    public static final String PUBSUB_PROPNAME = "pubsub";
-
-    /** Holds the pub/sub mode default, true means ping a topic, false means ping a queue. */
-    public static final boolean PUBSUB_DEFAULT = false;
-
-    /** Holds the name of the property to get the fail after commit flag from. */
-    public static final String FAIL_AFTER_COMMIT_PROPNAME = "failAfterCommit";
-
-    /** Holds the default failover after commit test flag. */
-    public static final boolean FAIL_AFTER_COMMIT_DEFAULT = false;
-
-    /** Holds the name of the proeprty to get the fail before commit flag from. */
-    public static final String FAIL_BEFORE_COMMIT_PROPNAME = "failBeforeCommit";
-
-    /** Holds the default failover before commit test flag. */
-    public static final boolean FAIL_BEFORE_COMMIT_DEFAULT = false;
-
-    /** Holds the name of the proeprty to get the fail after send flag from. */
-    public static final String FAIL_AFTER_SEND_PROPNAME = "failAfterSend";
-
-    /** Holds the default failover after send test flag. */
-    public static final boolean FAIL_AFTER_SEND_DEFAULT = false;
-
-    /** Holds the name of the property to get the fail before send flag from. */
-    public static final String FAIL_BEFORE_SEND_PROPNAME = "failBeforeSend";
-
-    /** Holds the default failover before send test flag. */
-    public static final boolean FAIL_BEFORE_SEND_DEFAULT = false;
-
-    /** Holds the name of the property to get the fail once flag from. */
-    public static final String FAIL_ONCE_PROPNAME = "failOnce";
-
-    /** The default failover once flag, true means only do one failover, false means failover on every commit cycle. */
-    public static final boolean FAIL_ONCE_DEFAULT = true;
-
-    /** Holds the name of the property to get the broker access username from. */
-    public static final String USERNAME_PROPNAME = "username";
-
-    /** Holds the default broker log on username. */
-    public static final String USERNAME_DEFAULT = "guest";
-
-    /** Holds the name of the property to get the broker access password from. */
-    public static final String PASSWORD_PROPNAME = "password";
-
-    /** Holds the default broker log on password. */
-    public static final String PASSWORD_DEFAULT = "guest";
-
-    /** Holds the name of the proeprty to get the. */
-    public static final String SELECTOR_PROPNAME = "selector";
-
-    /** Holds the default message selector. */
-    public static final String SELECTOR_DEFAULT = "";
-
-    /** Holds the name of the property to get the destination count from. */
-    public static final String DESTINATION_COUNT_PROPNAME = "destinationCount";
-
-    /** Defines the default number of destinations to ping. */
-    public static final int DESTINATION_COUNT_DEFAULT = 1;
-
-    /** Holds the name of the property to get the number of consumers per destination from. */
-    public static final String NUM_CONSUMERS_PROPNAME = "numConsumers";
-
-    /** Defines the default number consumers per destination. */
-    public static final int NUM_CONSUMERS_DEFAULT = 1;
-
-    /** Holds the name of the property to get the waiting timeout for response messages. */
-    public static final String TIMEOUT_PROPNAME = "timeout";
-
-    /** Default time to wait before assuming that a ping has timed out. */
-    public static final long TIMEOUT_DEFAULT = 30000;
-
-    /** Holds the name of the property to get the commit batch size from. */
-    public static final String TX_BATCH_SIZE_PROPNAME = "commitBatchSize";
-
-    /** Defines the default number of pings to send in each transaction when running transactionally. */
-    public static final int TX_BATCH_SIZE_DEFAULT = 1;
-
-    /** Holds the name of the property to get the unique destinations flag from. */
-    public static final String UNIQUE_DESTS_PROPNAME = "uniqueDests";
-
-    /** Defines the default value for the unique destinations property. */
-    public static final boolean UNIQUE_DESTS_DEFAULT = true;
-
-    /** Holds the name of the property to get the durable destinations flag from. */
-    public static final String DURABLE_DESTS_PROPNAME = "durableDests";
-
-    /** Defines the default value of the durable destinations flag. */
-    public static final boolean DURABLE_DESTS_DEFAULT = false;
-
-    /** Holds the name of the proeprty to get the message acknowledgement mode from. */
-    public static final String ACK_MODE_PROPNAME = "ackMode";
-
-    /** Defines the default message acknowledgement mode. */
-    public static final int ACK_MODE_DEFAULT = Session.AUTO_ACKNOWLEDGE;
-
-    /** Holds the name of the property to get the consumers message acknowledgement mode from. */
-    public static final String CONSUMER_ACK_MODE_PROPNAME = "consAckMode";
-
-    /** Defines the default consumers message acknowledgement mode. */
-    public static final int CONSUMER_ACK_MODE_DEFAULT = Session.AUTO_ACKNOWLEDGE;
-
-    /** Holds the name of the property to get the maximum pending message size setting from. */
-    public static final String MAX_PENDING_PROPNAME = "maxPending";
-
-    /** Defines the default value for the maximum pending message size setting. 0 means no limit. */
-    public static final int MAX_PENDING_DEFAULT = 0;
-
-    /** Defines the default prefetch size to use when consuming messages. */
-    public static final int PREFETCH_DEFAULT = 100;
-
-    /** Defines the default value of the no local flag to use when consuming messages. */
-    public static final boolean NO_LOCAL_DEFAULT = false;
-
-    /** Defines the default value of the exclusive flag to use when consuming messages. */
-    public static final boolean EXCLUSIVE_DEFAULT = false;
-
-    /** Holds the name of the property to store nanosecond timestamps in ping messages with. */
-    public static final String MESSAGE_TIMESTAMP_PROPNAME = "timestamp";
-
-    /** Holds the name of the property to get the number of message to prefill the broker with before starting the main test. */
-    public static final String PREFILL_PROPNAME = "preFill";
-
-    /** Defines the default value for the number of messages to prefill. 0,default, no messages. */
-    public static final int PREFILL_DEFAULT = 0;
-
-    /** Holds the name of the property to get the delay to wait in ms before starting the main test after having prefilled. */
-    public static final String DELAY_BEFORE_CONSUME_PROPNAME = "delayBeforeConsume";
-
-    /** Defines the default value for delay in ms to wait before starting thet test run. 0,default, no delay. */
-    public static final long  DELAY_BEFORE_CONSUME = 0;
-
-    /** Holds the name of the property to get when no messasges should be sent. */
-    public static final String CONSUME_ONLY_PROPNAME = "consumeOnly";
-
-    /** Defines the default value of the consumeOnly flag to use when publishing messages is not desired. */
-    public static final boolean CONSUME_ONLY_DEFAULT = false;
-
-    /** Holds the name of the property to get when no messasges should be sent. */
-    public static final String SEND_ONLY_PROPNAME = "sendOnly";
-
-    /** Defines the default value of the consumeOnly flag to use when publishing messages is not desired. */
-    public static final boolean SEND_ONLY_DEFAULT = false;
-
-    /** Holds the default configuration properties. */
-    public static ParsedProperties defaults = new ParsedProperties();
-
-    static
-    {
-        defaults.setPropertyIfNull(OVERRIDE_CLIENT_ID_PROPNAME, OVERRIDE_CLIENT_ID_DEAFULT);
-        defaults.setPropertyIfNull(FILE_PROPERTIES_PROPNAME, FILE_PROPERTIES_DEAFULT);
-        defaults.setPropertyIfNull(FACTORY_NAME_PROPNAME, FACTORY_NAME_DEAFULT);
-        defaults.setPropertyIfNull(BROKER_PROPNAME, BROKER_DEFAULT);
-        defaults.setPropertyIfNull(USERNAME_PROPNAME, USERNAME_DEFAULT);
-        defaults.setPropertyIfNull(PASSWORD_PROPNAME, PASSWORD_DEFAULT);
-        defaults.setPropertyIfNull(VIRTUAL_HOST_PROPNAME, VIRTUAL_HOST_DEFAULT);
-        defaults.setPropertyIfNull(PING_QUEUE_NAME_PROPNAME, PING_QUEUE_NAME_DEFAULT);
-        defaults.setPropertyIfNull(QUEUE_NAME_POSTFIX_PROPNAME, QUEUE_NAME_POSTFIX_DEFAULT);
-        defaults.setPropertyIfNull(SELECTOR_PROPNAME, SELECTOR_DEFAULT);
-        defaults.setPropertyIfNull(TRANSACTED_PROPNAME, TRANSACTED_DEFAULT);
-        defaults.setPropertyIfNull(CONSUMER_TRANSACTED_PROPNAME, CONSUMER_TRANSACTED_DEFAULT);
-        defaults.setPropertyIfNull(PERSISTENT_MODE_PROPNAME, PERSISTENT_MODE_DEFAULT);
-        defaults.setPropertyIfNull(ACK_MODE_PROPNAME, ACK_MODE_DEFAULT);
-        defaults.setPropertyIfNull(CONSUMER_ACK_MODE_PROPNAME, CONSUMER_ACK_MODE_DEFAULT);
-        defaults.setPropertyIfNull(MESSAGE_SIZE_PROPNAME, MESSAGE_SIZE_DEAFULT);
-        defaults.setPropertyIfNull(VERBOSE_PROPNAME, VERBOSE_DEFAULT);
-        defaults.setPropertyIfNull(PUBSUB_PROPNAME, PUBSUB_DEFAULT);
-        defaults.setPropertyIfNull(UNIQUE_DESTS_PROPNAME, UNIQUE_DESTS_DEFAULT);
-        defaults.setPropertyIfNull(DURABLE_DESTS_PROPNAME, DURABLE_DESTS_DEFAULT);
-        defaults.setPropertyIfNull(FAIL_BEFORE_COMMIT_PROPNAME, FAIL_BEFORE_COMMIT_DEFAULT);
-        defaults.setPropertyIfNull(FAIL_AFTER_COMMIT_PROPNAME, FAIL_AFTER_COMMIT_DEFAULT);
-        defaults.setPropertyIfNull(FAIL_BEFORE_SEND_PROPNAME, FAIL_BEFORE_SEND_DEFAULT);
-        defaults.setPropertyIfNull(FAIL_AFTER_SEND_PROPNAME, FAIL_AFTER_SEND_DEFAULT);
-        defaults.setPropertyIfNull(FAIL_ONCE_PROPNAME, FAIL_ONCE_DEFAULT);
-        defaults.setPropertyIfNull(TX_BATCH_SIZE_PROPNAME, TX_BATCH_SIZE_DEFAULT);
-        defaults.setPropertyIfNull(DESTINATION_COUNT_PROPNAME, DESTINATION_COUNT_DEFAULT);
-        defaults.setPropertyIfNull(NUM_CONSUMERS_PROPNAME, NUM_CONSUMERS_DEFAULT);
-        defaults.setPropertyIfNull(RATE_PROPNAME, RATE_DEFAULT);
-        defaults.setPropertyIfNull(TIMEOUT_PROPNAME, TIMEOUT_DEFAULT);
-        defaults.setPropertyIfNull(MAX_PENDING_PROPNAME, MAX_PENDING_DEFAULT);
-        defaults.setPropertyIfNull(PREFILL_PROPNAME, PREFILL_DEFAULT);
-        defaults.setPropertyIfNull(DELAY_BEFORE_CONSUME_PROPNAME, DELAY_BEFORE_CONSUME);
-        defaults.setPropertyIfNull(CONSUME_ONLY_PROPNAME, CONSUME_ONLY_DEFAULT);
-        defaults.setPropertyIfNull(SEND_ONLY_PROPNAME, SEND_ONLY_DEFAULT);
-    }
-
-    /** Allows setting of client ID on the connection, rather than through the connection URL. */
-    protected boolean _overrideClientId;
-
-    /** Holds the JNDI name of the JMS connection factory. */
-    protected String _factoryName;
-
-    /** Holds the name of the properties file to configure JNDI with. */
-    protected String _fileProperties;
-
-    /** Holds the broker url. */
-    protected String _brokerDetails;
-
-    /** Holds the username to access the broker with. */
-    protected String _username;
-
-    /** Holds the password to access the broker with. */
-    protected String _password;
-
-    /** Holds the virtual host on the broker to run the tests through. */
-    protected String _virtualpath;
-
-    /** Holds the root name from which to generate test destination names. */
-    protected String _destinationName;
-
-    /** Holds the default queue name postfix value. */
-    protected String _queueNamePostfix;
-
-    /** Holds the message selector to filter the pings with. */
-    protected String _selector;
-
-    /** Holds the producers transactional mode flag. */
-    protected boolean _transacted;
-
-    /** Holds the consumers transactional mode flag. */
-    protected boolean _consTransacted;
-
-    /** Determines whether this producer sends persistent messages. */
-    protected boolean _persistent;
-
-    /** Holds the acknowledgement mode used for the producers. */
-    protected int _ackMode;
-
-    /** Holds the acknowledgement mode setting for the consumers. */
-    protected int _consAckMode;
-
-    /** Determines what size of messages this producer sends. */
-    protected int _messageSize;
-
-    /** Used to indicate that the ping loop should print out whenever it pings. */
-    protected boolean _verbose;
-
-    /** Flag used to indicate if this is a point to point or pub/sub ping client. */
-    protected boolean _isPubSub;
-
-    /** Flag used to indicate if the destinations should be unique client. */
-    protected boolean _isUnique;
-
-    /** Flag used to indicate that durable destination should be used. */
-    protected boolean _isDurable;
-
-    /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover before a commit. */
-    protected boolean _failBeforeCommit;
-
-    /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover after a commit. */
-    protected boolean _failAfterCommit;
-
-    /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover before a send. */
-    protected boolean _failBeforeSend;
-
-    /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover after a send. */
-    protected boolean _failAfterSend;
-
-    /** Flag used to indicate that failover prompting should only be done on the first commit, not on every commit. */
-    protected boolean _failOnce;
-
-    /** Holds the number of sends that should be performed in every transaction when using transactions. */
-    protected int _txBatchSize;
-
-    /** Holds the number of destinations to ping. */
-    protected int _noOfDestinations;
-
-    /** Holds the number of consumers per destination. */
-    protected int _noOfConsumers;
-
-    private int[] _consumerBatchCounts;
-
-    /** Holds the maximum send rate in herz. */
-    protected int _rate;
-
-    /**
-     * Holds the size of the maximum amount of pending data that the client should buffer, sending is suspended
-     * if this limit is breached.
-     */
-    protected int _maxPendingSize;
-
-    /**
-     * Holds the number of messages to send during the setup phase, before the clients start consuming.
-     */
-    private Integer _preFill;
-
-    /**
-     * Holds the time in ms to wait after preFilling before starting thet test.
-     */
-    private Long _delayBeforeConsume;
-
-    /**
-     * Holds a boolean value of wither this test should just consume, i.e. skips
-     * sending messages, but still expects to receive the specified number.
-     */
-    private boolean _consumeOnly;
-
-    /**
-     * Holds a boolean value of wither this test should just send, i.e. skips
-     * consuming messages, but still creates clients just doesn't start them.
-     */
-    private boolean _sendOnly;
-
-
-    /** A source for providing sequential unique correlation ids. These will be unique within the same JVM. */
-    private static AtomicLong _correlationIdGenerator = new AtomicLong(0L);
-
-    /** A source for providing sequential unqiue ids for instances of this class to be identifed with. */
-    private static AtomicInteger _instanceIdGenerator = new AtomicInteger(0);
-
-    /** Holds this instances unique id. */
-    private int instanceId;
-
-    /**
-     * Holds a map from message ids to latches on which threads wait for replies. This map is shared accross multiple
-     * ping producers on the same JVM.
-     */
-    private static Map<String, PerCorrelationId> perCorrelationIds =
-        Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
-
-    /** A convenient formatter to use when time stamping output. */
-    protected static final DateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS");
-
-    /** Holds the connection for the message producer. */
-    protected Connection _connection;
-
-    /** Holds the consumer connections. */
-    protected Connection[] _consumerConnection;
-
-    /** Holds the controlSession on which ping replies are received. */
-    protected Session[] _consumerSession;
-
-    /** Holds the producer controlSession, needed to create ping messages. */
-    protected Session _producerSession;
-
-    /** Holds the destination where the response messages will arrive. */
-    protected Destination _replyDestination;
-
-    /** Holds the set of destinations that this ping producer pings. */
-    protected List<Destination> _pingDestinations;
-
-    /** Used to restrict the sending rate to a specified limit. */
-    protected Throttle _rateLimiter;
-
-    /** Holds a message listener that this message listener chains all its messages to. */
-    protected ChainedMessageListener _chainedMessageListener = null;
-
-    /**
-     * This id generator is used to generate ids to append to the queue name to ensure that queues can be unique when
-     * creating multiple ping producers in the same JVM.
-     */
-    protected static AtomicInteger _queueJVMSequenceID = new AtomicInteger();
-
-    /**
-     * This id generator is used to generates ids that are only unique within this pinger. Creating multiple pingers
-     * on the same JVM using this id generator will allow them to ping on the same queues.
-     */
-    protected AtomicInteger _queueSharedID = new AtomicInteger();
-
-    /** Used to tell the ping loop when to terminate, it only runs while this is true. */
-    protected boolean _publish = true;
-
-    /** Holds the message producer to send the pings through. */
-    protected MessageProducer _producer;
-
-    /** Holds the message consumer to receive the ping replies through. */
-    protected MessageConsumer[] _consumer;
-
-    /** The prompt to display when asking the user to kill the broker for failover testing. */
-    private static final String KILL_BROKER_PROMPT = "Kill broker now, then press Return.";
-
-    /** Holds the name for this test client to be identified to the broker with. */
-    private String _clientID;
-
-    /** Keeps count of the total messages sent purely for debugging purposes. */
-    private static AtomicInteger numSent = new AtomicInteger();
-
-    /**
-     * Holds a monitor which is used to synchronize sender and receivers threads, where the sender has elected
-     * to wait until the number of unreceived message is reduced before continuing to send. This monitor is a
-     * fair SynchronousQueue becuase that provides fair scheduling, to ensure that all producer threads get an
-     * equal chance to produce messages.
-     */
-    static final SynchronousQueue _sendPauseMonitor = new SynchronousQueue(true);
-
-    /** Keeps a count of the number of message currently sent but not received. */
-    static AtomicInteger _unreceived = new AtomicInteger(0);
-
-    /**
-     * Creates a ping producer with the specified parameters, of which there are many. See the class level comments
-     * for details. This constructor creates a connection to the broker and creates producer and consumer sessions on
-     * it, to send and recieve its pings and replies on.
-     *
-     * @param overrides Properties containing any desired overrides to the defaults.
-     *
-     * @throws Exception Any exceptions are allowed to fall through.
-     */
-    public PingPongProducer(Properties overrides) throws Exception
-    {
-        // log.debug("public PingPongProducer(Properties overrides = " + overrides + "): called");
-        instanceId = _instanceIdGenerator.getAndIncrement();
-
-        // Create a set of parsed properties from the defaults overriden by the passed in values.
-        ParsedProperties properties = new ParsedProperties(defaults);
-        properties.putAll(overrides);
-
-        // Extract the configuration properties to set the pinger up with.
-        _overrideClientId = properties.getPropertyAsBoolean(OVERRIDE_CLIENT_ID_PROPNAME);
-        _factoryName = properties.getProperty(FACTORY_NAME_PROPNAME);
-        _fileProperties = properties.getProperty(FILE_PROPERTIES_PROPNAME);
-        _brokerDetails = properties.getProperty(BROKER_PROPNAME);
-        _username = properties.getProperty(USERNAME_PROPNAME);
-        _password = properties.getProperty(PASSWORD_PROPNAME);
-        _virtualpath = properties.getProperty(VIRTUAL_HOST_PROPNAME);
-        _destinationName = properties.getProperty(PING_QUEUE_NAME_PROPNAME);
-        _queueNamePostfix = properties.getProperty(QUEUE_NAME_POSTFIX_PROPNAME);
-        _selector = properties.getProperty(SELECTOR_PROPNAME);
-        _transacted = properties.getPropertyAsBoolean(TRANSACTED_PROPNAME);
-        _consTransacted = properties.getPropertyAsBoolean(CONSUMER_TRANSACTED_PROPNAME);
-        _persistent = properties.getPropertyAsBoolean(PERSISTENT_MODE_PROPNAME);
-        _messageSize = properties.getPropertyAsInteger(MESSAGE_SIZE_PROPNAME);
-        _verbose = properties.getPropertyAsBoolean(VERBOSE_PROPNAME);
-        _failAfterCommit = properties.getPropertyAsBoolean(FAIL_AFTER_COMMIT_PROPNAME);
-        _failBeforeCommit = properties.getPropertyAsBoolean(FAIL_BEFORE_COMMIT_PROPNAME);
-        _failAfterSend = properties.getPropertyAsBoolean(FAIL_AFTER_SEND_PROPNAME);
-        _failBeforeSend = properties.getPropertyAsBoolean(FAIL_BEFORE_SEND_PROPNAME);
-        _failOnce = properties.getPropertyAsBoolean(FAIL_ONCE_PROPNAME);
-        _txBatchSize = properties.getPropertyAsInteger(TX_BATCH_SIZE_PROPNAME);
-        _noOfDestinations = properties.getPropertyAsInteger(DESTINATION_COUNT_PROPNAME);
-        _noOfConsumers = properties.getPropertyAsInteger(NUM_CONSUMERS_PROPNAME);
-        _consumerBatchCounts = new int[_noOfConsumers];
-        _rate = properties.getPropertyAsInteger(RATE_PROPNAME);
-        _isPubSub = properties.getPropertyAsBoolean(PUBSUB_PROPNAME);
-        _isUnique = properties.getPropertyAsBoolean(UNIQUE_DESTS_PROPNAME);
-        _isDurable = properties.getPropertyAsBoolean(DURABLE_DESTS_PROPNAME);
-        _ackMode = _transacted ? 0 : properties.getPropertyAsInteger(ACK_MODE_PROPNAME);
-        _consAckMode = _consTransacted ? 0 : properties.getPropertyAsInteger(CONSUMER_ACK_MODE_PROPNAME);
-        _maxPendingSize = properties.getPropertyAsInteger(MAX_PENDING_PROPNAME);
-        _preFill = properties.getPropertyAsInteger(PREFILL_PROPNAME);
-        _delayBeforeConsume = properties.getPropertyAsLong(DELAY_BEFORE_CONSUME_PROPNAME);
-        _consumeOnly = properties.getPropertyAsBoolean(CONSUME_ONLY_PROPNAME);
-        _sendOnly = properties.getPropertyAsBoolean(SEND_ONLY_PROPNAME);
-
-        // Check that one or more destinations were specified.
-        if (_noOfDestinations < 1)
-        {
-            throw new IllegalArgumentException("There must be at least one destination.");
-        }
-
-        // Set up a throttle to control the send rate, if a rate > 0 is specified.
-        if (_rate > 0)
-        {
-            _rateLimiter = new BatchedThrottle();
-            _rateLimiter.setRate(_rate);
-        }
-
-        // Create the connection and message producers/consumers.
-        // establishConnection(true, true);
-    }
-
-    /**
-     * Establishes a connection to the broker and creates message consumers and producers based on the parameters
-     * that this ping client was created with.
-     *
-     * @param producer Flag to indicate whether or not the producer should be set up.
-     * @param consumer Flag to indicate whether or not the consumers should be set up.
-     *
-     * @throws Exception Any exceptions are allowed to fall through.
-     */
-    public void establishConnection(boolean producer, boolean consumer) throws Exception
-    {
-        // log.debug("public void establishConnection(): called");
-
-        // Generate a unique identifying name for this client, based on it ip address and the current time.
-        InetAddress address = InetAddress.getLocalHost();
-        // _clientID = address.getHostName() + System.currentTimeMillis();
-        _clientID = "perftest_" + instanceId;
-
-        // Create a connection to the broker.
-        createConnection(_clientID);
-
-        // Create transactional or non-transactional sessions, based on the command line arguments.
-        _producerSession = _connection.createSession(_transacted, _ackMode);
-
-        _consumerSession = new Session[_noOfConsumers];
-
-        for (int i = 0; i < _noOfConsumers; i++)
-        {
-            _consumerSession[i] = _consumerConnection[i].createSession(_consTransacted, _consAckMode);
-        }
-
-        // Create the destinations to send pings to and receive replies from.
-        if (_noOfConsumers > 0)
-        {
-            _replyDestination = _consumerSession[0].createTemporaryQueue();
-        }
-        createPingDestinations(_noOfDestinations, _selector, _destinationName, _isUnique, _isDurable);
-
-        // Create the message producer only if instructed to.
-        if (producer)
-        {
-            createProducer();
-        }
-
-        // Create the message consumer only if instructed to.
-        if (consumer)
-        {
-            createReplyConsumers(getReplyDestinations(), _selector);
-        }
-    }
-
-    /**
-     * Establishes a connection to the broker, based on the configuration parameters that this ping client was
-     * created with.
-     *
-     * @param clientID The clients identifier.
-     *
-     * @throws JMSException Underlying exceptions allowed to fall through.
-     * @throws NamingException Underlying exceptions allowed to fall through.
-     * @throws IOException Underlying exceptions allowed to fall through.
-     */
-    protected void createConnection(String clientID) throws JMSException, NamingException, IOException
-    {
-        // _log.debug("protected void createConnection(String clientID = " + clientID + "): called");
-
-        // _log.debug("Creating a connection for the message producer.");
-        File propsFile = new File(_fileProperties);
-        InputStream is = new FileInputStream(propsFile);
-        Properties properties = new Properties();
-        properties.load(is);
-
-        Context context = new InitialContext(properties);
-        ConnectionFactory factory = (ConnectionFactory) context.lookup(_factoryName);
-        _connection = factory.createConnection(_username, _password);
-
-        if (_overrideClientId)
-        {
-            _connection.setClientID(clientID);
-        }
-
-        // _log.debug("Creating " + _noOfConsumers + " connections for the consumers.");
-
-        _consumerConnection = new Connection[_noOfConsumers];
-
-        for (int i = 0; i < _noOfConsumers; i++)
-        {
-            _consumerConnection[i] = factory.createConnection(_username, _password);
-            // _consumerConnection[i].setClientID(clientID);
-        }
-    }
-
-    /**
-     * Starts a ping-pong loop running from the command line. The bounce back client {@link PingPongBouncer} also needs
-     * to be started to bounce the pings back again.
-     *
-     * @param args The command line arguments.
-     */
-    public static void main(String[] args)
-    {
-        try
-        {
-            Properties options =
-                CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
-
-            // Create a ping producer overriding its defaults with all options passed on the command line.
-            PingPongProducer pingProducer = new PingPongProducer(options);
-            pingProducer.establishConnection(true, true);
-
-            // Start the ping producers dispatch thread running.
-            pingProducer._connection.start();
-
-            // Create a shutdown hook to terminate the ping-pong producer.
-            Runtime.getRuntime().addShutdownHook(pingProducer.getShutdownHook());
-
-            // Ensure that the ping pong producer is registered to listen for exceptions on the connection too.
-            pingProducer._connection.setExceptionListener(pingProducer);
-
-            // Create the ping loop thread and run it until it is terminated by the shutdown hook or exception.
-            Thread pingThread = new Thread(pingProducer);
-            pingThread.run();
-            pingThread.join();
-        }
-        catch (Exception e)
-        {
-            System.err.println(e.getMessage());
-            log.error("Top level handler caught execption.", e);
-            System.exit(1);
-        }
-    }
-
-    /**
-     * Convenience method for a short pause.
-     *
-     * @param sleepTime The time in milliseconds to pause for.
-     */
-    public static void pause(long sleepTime)
-    {
-        if (sleepTime > 0)
-        {
-            try
-            {
-                Thread.sleep(sleepTime);
-            }
-            catch (InterruptedException ie)
-            { }
-        }
-    }
-
-    /**
-     * Gets all the reply destinations (to listen for replies on). In this case this will just be the single reply to
-     * destination of this pinger.
-     *
-     * @return The single reply to destination of this pinger, wrapped in a list.
-     */
-    public List<Destination> getReplyDestinations()
-    {
-        // log.debug("public List<Destination> getReplyDestinations(): called");
-
-        List<Destination> replyDestinations = new ArrayList<Destination>();
-        replyDestinations.add(_replyDestination);
-
-        // log.debug("replyDestinations = " + replyDestinations);
-
-        return replyDestinations;
-    }
-
-    /**
-     * Creates the producer to send the pings on. This is created without a default destination. Its persistent delivery
-     * flag is set accoring the ping producer creation options.
-     *
-     * @throws JMSException Any JMSExceptions are allowed to fall through.
-     */
-    public void createProducer() throws JMSException
-    {
-        // log.debug("public void createProducer(): called");
-
-        _producer = (MessageProducer) _producerSession.createProducer(null);
-        _producer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
-
-        // log.debug("Created producer for " + (_persistent ? "persistent" : "non-persistent") + " messages.");
-    }
-
-    /**
-     * Creates consumers for the specified number of destinations. The destinations themselves are also created by this
-     * method.
-     *
-     * @param noOfDestinations The number of destinations to create consumers for.
-     * @param selector         The message selector to filter the consumers with.
-     * @param rootName         The root of the name, or actual name if only one is being created.
-     * @param unique           <tt>true</tt> to make the destinations unique to this pinger, <tt>false</tt> to share the
-     *                         numbering with all pingers on the same JVM.
-     * @param durable          If the destinations are durable topics.
-     *
-     * @throws JMSException Any JMSExceptions are allowed to fall through.
-     */
-    public void createPingDestinations(int noOfDestinations, String selector, String rootName, boolean unique,
-        boolean durable) throws JMSException
-    {
-        /*log.debug("public void createPingDestinations(int noOfDestinations = " + noOfDestinations + ", String selector = "
-            + selector + ", String rootName = " + rootName + ", boolean unique = " + unique + ", boolean durable = "
-            + durable + "): called");*/
-
-        _pingDestinations = new ArrayList<Destination>();
-
-        // Create the desired number of ping destinations and consumers for them.
-        // log.debug("Creating " + noOfDestinations + " destinations to ping.");
-
-        for (int i = 0; i < noOfDestinations; i++)
-        {
-            Destination destination;
-            String id;
-
-            // Generate an id, unique within this pinger or to the whole JVM depending on the unique flag.
-            if (unique)
-            {
-                // log.debug("Creating unique destinations.");
-                id = "_" + _queueJVMSequenceID.incrementAndGet() + "_" + _connection.getClientID();
-            }
-            else
-            {
-                // log.debug("Creating shared destinations.");
-                id = "_" + _queueSharedID.incrementAndGet();
-            }
-
-            // Check if this is a pub/sub pinger, in which case create topics.
-            if (_isPubSub)
-            {
-                destination = _producerSession.createTopic(rootName + id);
-                // log.debug("Created non-durable topic " + destination);
-
-                if (durable)
-                {
-                    _producerSession.createDurableSubscriber((Topic) destination, _connection.getClientID());
-                }
-            }
-            // Otherwise this is a p2p pinger, in which case create queues.
-            else
-            {
-                destination = _producerSession.createQueue(rootName + id + _queueNamePostfix);
-                // log.debug("Created queue " + destination);
-            }
-
-            // Keep the destination.
-            _pingDestinations.add(destination);
-        }
-    }
-
-    /**
-     * Creates consumers for the specified destinations and registers this pinger to listen to their messages.
-     *
-     * @param destinations The destinations to listen to.
-     * @param selector     A selector to filter the messages with.
-     *
-     * @throws javax.jms.JMSException Any JMSExceptions are allowed to fall through.
-     */
-    public void createReplyConsumers(Collection<Destination> destinations, String selector) throws JMSException
-    {
-        /*log.debug("public void createReplyConsumers(Collection<Destination> destinations = " + destinations
-            + ", String selector = " + selector + "): called");*/
-
-        log.debug("There are " + destinations.size() + " destinations.");
-        log.debug("Creating " + _noOfConsumers + " consumers on each destination.");
-        log.debug("Total number of consumers is: " + (destinations.size() * _noOfConsumers));
-
-        for (Destination destination : destinations)
-        {
-            _consumer = new MessageConsumer[_noOfConsumers];
-
-            // If we don't have consumers then ensure we have created the
-            // destination.   
-            if (_noOfConsumers == 0)
-            {
-                _producerSession.createConsumer(destination, selector,
-                                                NO_LOCAL_DEFAULT).close();
-            }
-
-            for (int i = 0; i < _noOfConsumers; i++)
-            {
-                // Create a consumer for the destination and set this pinger to listen to its messages.
-                _consumer[i] = _consumerSession[i].createConsumer(destination, selector, NO_LOCAL_DEFAULT);
-
-                final int consumerNo = i;
-
-                _consumer[i].setMessageListener(new MessageListener()
-                    {
-                        public void onMessage(Message message)
-                        {
-                            onMessageWithConsumerNo(message, consumerNo);
-                        }
-                    });
-
-                log.debug("Set consumer " + i + " to listen to replies sent to destination: " + destination);
-            }
-        }
-    }
-
-    /**
-     * Stores the received message in the replies map, then resets the boolean latch that a thread waiting for a
-     * correlating reply may be waiting on. This is only done if the reply has a correlation id that is expected in the
-     * replies map.
-     *
-     * @param message    The received message.
-     * @param consumerNo The consumer number within this test pinger instance.
-     */
-    public void onMessageWithConsumerNo(Message message, int consumerNo)
-    {
-        // log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo = " + consumerNo + "): called");
-        try
-        {
-            long now = System.nanoTime();
-            long timestamp = getTimestamp(message);
-            long pingTime = now - timestamp;
-
-            // NDC.push("id" + instanceId + "/cons" + consumerNo);
-
-            // Extract the messages correlation id.
-            String correlationID = message.getJMSCorrelationID();
-            // log.debug("correlationID = " + correlationID);
-
-            // int num = message.getIntProperty("MSG_NUM");
-            // log.info("Message " + num + " received.");
-
-            boolean isRedelivered = message.getJMSRedelivered();
-            // log.debug("isRedelivered = " + isRedelivered);
-
-            if (!isRedelivered)
-            {
-                // Countdown on the traffic light if there is one for the matching correlation id.
-                PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationID);
-
-                if (perCorrelationId != null)
-                {
-                    CountDownLatch trafficLight = perCorrelationId.trafficLight;
-
-                    // Restart the timeout timer on every message.
-                    perCorrelationId.timeOutStart = System.nanoTime();
-
-                    // log.debug("Reply was expected, decrementing the latch for the id, " + correlationID);
-
-                    // Release waiting senders if there are some and using maxPending limit.
-                    if ((_maxPendingSize > 0))
-                    {
-                        // Decrement the count of sent but not yet received messages.
-                        int unreceived = _unreceived.decrementAndGet();
-                        int unreceivedSize =
-                            (unreceived * ((_messageSize == 0) ? 1 : _messageSize))
-                            / (_isPubSub ? getConsumersPerDestination() : 1);
-
-                        // log.debug("unreceived = " + unreceived);
-                        // log.debug("unreceivedSize = " + unreceivedSize);
-
-                        // synchronized (_sendPauseMonitor)
-                        // {
-                        if (unreceivedSize < _maxPendingSize)
-                        {
-                            _sendPauseMonitor.poll();
-                        }
-                        // }
-                    }
-
-                    // Decrement the countdown latch. Before this point, it is possible that two threads might enter this
-                    // method simultanesouly with the same correlation id. Decrementing the latch in a synchronized block
-                    // ensures that each thread will get a unique value for the remaining messages.
-                    long trueCount;
-                    long remainingCount;
-
-                    synchronized (trafficLight)
-                    {
-                        trafficLight.countDown();
-
-                        trueCount = trafficLight.getCount();
-                        remainingCount = trueCount - 1;
-
-                        if (++_consumerBatchCounts[consumerNo] == _txBatchSize)
-                        {
-                            _consumerBatchCounts[consumerNo] = 0;
-                            if (_consAckMode == 2)
-                            {
-                                // log.debug("Doing client ack for consumer " + consumerNo + ".");
-                                message.acknowledge();
-                            }
-                            else
-                            {
-                                // log.debug("Trying commit for consumer " + consumerNo + ".");
-                                commitTx(_consumerSession[consumerNo]);
-                                // log.info("Tx committed on consumer " + consumerNo);
-                            }
-                        }
-
-                        // Forward the message and remaining count to any interested chained message listener.
-                        if (_chainedMessageListener != null)
-                        {
-                            _chainedMessageListener.onMessage(message, (int) remainingCount, pingTime);
-                        }
-
-                        // Check if this is the last message, in which case release any waiting producers. This is done
-                        // after the transaction has been committed and any listeners notified.
-                        if (trueCount == 1)
-                        {
-                            trafficLight.countDown();
-                        }
-                    }
-                }
-                else
-                {
-                    log.warn(consumerNo + " Got unexpected message with correlationId: " + correlationID);
-                    log.warn(consumerNo + " Map contains:" + perCorrelationIds.entrySet());
-                }
-            }
-            else
-            {
-                log.warn("Got redelivered message, ignoring.");
-            }
-        }
-        catch (Exception e)
-        {
-            log.warn("There was a Exception: " + e.getMessage(), e);
-        }
-        finally
-        {
-            // log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo): ending");
-            // NDC.clear();
-        }
-    }
-
-    public void setupCorrelationID(String correlationId, int expectedCount)
-    {
-        PerCorrelationId perCorrelationId = new PerCorrelationId();
-
-        // Create a count down latch to count the number of replies with. This is created before the messages are
-        // sent so that the replies cannot be received before the count down is created.
-        // One is added to this, so that the last reply becomes a special case. The special case is that the
-        // chained message listener must be called before this sender can be unblocked, but that decrementing the
-        // countdown needs to be done before the chained listener can be called.
-        perCorrelationId.trafficLight = new CountDownLatch(expectedCount + 1);
-
-        perCorrelationIds.put(correlationId, perCorrelationId);
-    }
-
-
-    /**
-     * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out
-     * before a reply arrives, then a null reply is returned from this method. This method allows the caller to specify
-     * the correlation id.
-     *
-     * Can be augmented through a pre-fill property (PingPongProducer.PREFILL_PROPNAME) that will populate the destination
-     * with a set number of messages so the total pings sent and therefore expected will be PREFILL + numPings.
-     *
-     * If pre-fill is specified then the consumers will start paused to allow the prefilling to occur.
-     *
-     * @param message              The message to send. If this is null, one is generated.
-     * @param numPings             The number of ping messages to send.
-     * @param timeout              The timeout in milliseconds.
-     * @param messageCorrelationId The message correlation id. If this is null, one is generated.
-     *
-     * @return The number of replies received. This may be less than the number sent if the timeout terminated the wait
-     *         for all prematurely. If we are running in noConsumer=0 so send only mode then it will return the no msgs sent.
-     *
-     * @throws JMSException         All underlying JMSExceptions are allowed to fall through.
-     * @throws InterruptedException When interrupted by a timeout
-     */
-    public int pingAndWaitForReply(Message message, int numPings, long timeout, String messageCorrelationId)
-        throws JMSException, InterruptedException
-    {
-        return pingAndWaitForReply(message, numPings, 0, timeout, messageCorrelationId);
-    }
-
-    public int pingAndWaitForReply(Message message, int numPings, int preFill, long timeout, String messageCorrelationId)
-        throws JMSException, InterruptedException
-    {
-        /*log.debug("public int pingAndWaitForReply(Message message, int numPings = " + numPings + ", long timeout = "
-            + timeout + ", String messageCorrelationId = " + messageCorrelationId + "): called");*/
-
-        int totalPingsRequested = numPings + preFill;
-
-        // Generate a unique correlation id to put on the messages before sending them, if one was not specified.
-        if (messageCorrelationId == null)
-        {
-            messageCorrelationId = Long.toString(_correlationIdGenerator.incrementAndGet());
-
-            setupCorrelationID(messageCorrelationId, getExpectedNumPings(totalPingsRequested));
-        }
-
-        try
-        {
-            // NDC.push("prod");
-
-            PerCorrelationId perCorrelationId = perCorrelationIds.get(messageCorrelationId);
-
-            // Set up the current time as the start time for pinging on the correlation id. This is used to determine
-            // timeouts.
-            perCorrelationId.timeOutStart = System.nanoTime();
-
-            // Send the specifed number of messages for this test            
-            pingNoWaitForReply(message, numPings, messageCorrelationId);
-
-            boolean timedOut;
-            boolean allMessagesReceived;
-            int numReplies;
-
-            // We don't have a consumer so don't try and wait for the messages.
-            // this does mean that if the producerSession is !TXed then we may
-            // get to exit before all msgs have been received.
-            //
-            // Return the number of requested messages, this will let the test
-            // report a pass.
-            if (_noOfConsumers == 0 || _sendOnly)
-            {
-                return getExpectedNumPings(totalPingsRequested);
-            }
-
-            do
-            {
-                // Block the current thread until replies to all the messages are received, or it times out.
-                perCorrelationId.trafficLight.await(timeout, TimeUnit.MILLISECONDS);
-
-                // Work out how many replies were receieved.
-                numReplies = getExpectedNumPings(totalPingsRequested) - (int) perCorrelationId.trafficLight.getCount();
-
-                allMessagesReceived = numReplies == getExpectedNumPings(totalPingsRequested);
-
-                // log.debug("numReplies = " + numReplies);
-                // log.debug("allMessagesReceived = " + allMessagesReceived);
-
-                // Recheck the timeout condition.
-                long now = System.nanoTime();
-                long lastMessageReceievedAt = perCorrelationId.timeOutStart;
-                timedOut = (now - lastMessageReceievedAt) > (timeout * 1000000);
-
-                // log.debug("now = " + now);
-                // log.debug("lastMessageReceievedAt = " + lastMessageReceievedAt);
-            }
-            while (!timedOut && !allMessagesReceived);
-
-            if ((numReplies < getExpectedNumPings(totalPingsRequested)) && _verbose)
-            {
-                log.info("Timed out (" + timeout + " ms) before all replies received on id, " + messageCorrelationId);
-            }
-            else if (_verbose)
-            {
-                log.info("Got all replies on id, " + messageCorrelationId);
-            }
-
-            // commitTx(_consumerSession);
-
-            // log.debug("public int pingAndWaitForReply(Message message, int numPings, long timeout): ending");
-
-            return numReplies;
-        }
-        // Ensure that the message countdown latch is always removed from the reply map. The reply map is long lived,
-        // so will be a memory leak if this is not done.
-        finally
-        {
-            // NDC.pop();
-            perCorrelationIds.remove(messageCorrelationId);
-        }
-    }
-
-    /**
-     * Sends the specified number of ping messages and does not wait for correlating replies.
-     *
-     * @param message              The message to send.
-     * @param numPings             The number of pings to send.
-     * @param messageCorrelationId A correlation id to place on all messages sent.
-     *
-     * @throws JMSException All underlying JMSExceptions are allowed to fall through.
-     */
-    public void pingNoWaitForReply(Message message, int numPings, String messageCorrelationId) throws JMSException
-    {
-        /*log.debug("public void pingNoWaitForReply(Message message, int numPings = " + numPings
-            + ", String messageCorrelationId = " + messageCorrelationId + "): called");*/
-
-        // If we are runnning a consumeOnly test then don't send any messages
-        if (_consumeOnly)
-        {
-            return;
-        }
-        
-        if (message == null)
-        {
-            message = getTestMessage(getReplyDestinations().get(0), _messageSize, _persistent);
-        }
-
-        message.setJMSCorrelationID(messageCorrelationId);
-
-        // Set up a committed flag to detect uncommitted messages at the end of the send loop. This may occurr if the
-        // transaction batch size is not a factor of the number of pings. In which case an extra commit at the end is
-        // needed.
-        boolean committed = false;
-
-        // Send all of the ping messages.
-        for (int i = 0; i < numPings; i++)
-        {
-            // Re-timestamp the message.
-            // message.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
-
-            // Send the message, passing in the message count.
-            committed = sendMessage(i, message);
-
-            // Spew out per message timings on every message sonly in verbose mode.
-            /*if (_verbose)
-            {
-                log.info(timestampFormatter.format(new Date()) + ": Pinged at with correlation id, " + messageCorrelationId);
-            }*/
-        }
-
-        // Call commit if the send loop finished before reaching a batch size boundary so there may still be uncommitted messages.
-        if (!committed)
-        {
-            commitTx(_producerSession);
-        }
-    }
-
-    /**
-     * Sends the sepcified message, applies rate limiting and possibly commits the current transaction. The count of
-     * messages sent so far must be specified and is used to round robin the ping destinations (where there are more
-     * than one), and to determine if the transaction batch size has been reached and the sent messages should be
-     * committed.
-     *
-     * @param i       The count of messages sent so far in a loop of multiple calls to this send method.
-     * @param message The message to send.
-     *
-     * @return <tt>true</tt> if the messages were committed, <tt>false</tt> otherwise.
-     *
-     * @throws JMSException All underlyiung JMSExceptions are allowed to fall through.
-     */
-    protected boolean sendMessage(int i, Message message) throws JMSException
-    {
-        try
-        {
-            NDC.push("id" + instanceId + "/prod");
-
-            // log.debug("protected boolean sendMessage(int i = " + i + ", Message message): called");
-            // log.debug("_txBatchSize = " + _txBatchSize);
-
-            // Round robin the destinations as the messages are sent.
-            Destination destination = _pingDestinations.get(i % _pingDestinations.size());
-
-            // Prompt the user to kill the broker when doing failover testing.
-            _failBeforeSend = waitForUserToPromptOnFailure(_failBeforeSend);
-
-            // Get the test setup for the correlation id.
-            String correlationID = message.getJMSCorrelationID();
-            PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationID);
-
-            // If necessary, wait until the max pending message size comes within its limit.
-            if (_maxPendingSize > 0)
-            {
-                synchronized (_sendPauseMonitor)
-                {
-                    // Used to keep track of the number of times that send has to wait.
-                    int numWaits = 0;
-
-                    // The maximum number of waits before the test gives up and fails. This has been chosen to correspond with
-                    // the test timeout.
-                    int waitLimit = (int) (TIMEOUT_DEFAULT / 10000);
-
-                    while (true)
-                    {
-                        // Get the size estimate of sent but not yet received messages.
-                        int unreceived = _unreceived.get();
-                        int unreceivedSize =
-                            (unreceived * ((_messageSize == 0) ? 1 : _messageSize))
-                            / (_isPubSub ? getConsumersPerDestination() : 1);
-
-                        // log.debug("unreceived = " + unreceived);
-                        // log.debug("unreceivedSize = " + unreceivedSize);
-                        // log.debug("_maxPendingSize = " + _maxPendingSize);
-
-                        if (unreceivedSize > _maxPendingSize)
-                        {
-                            // log.debug("unreceived size estimate over limit = " + unreceivedSize);
-
-                            // Fail the test if the send has had to wait more than the maximum allowed number of times.
-                            if (numWaits > waitLimit)
-                            {
-                                String errorMessage =
-                                    "Send has had to wait for the unreceivedSize (" + unreceivedSize
-                                    + ") to come below the maxPendingSize (" + _maxPendingSize + ") more that " + waitLimit
-                                    + " times.";
-                                log.warn(errorMessage);
-                                throw new RuntimeException(errorMessage);
-                            }
-
-                            // Wait on the send pause barrier for the limit to be re-established.
-                            try
-                            {
-                                long start = System.nanoTime();
-                                // _sendPauseMonitor.wait(10000);
-                                _sendPauseMonitor.offer(new Object(), 10000, TimeUnit.MILLISECONDS);
-                                long end = System.nanoTime();
-
-                                // Count the wait only if it was for > 99% of the requested wait time.
-                                if (((float) (end - start) / (float) (10000 * 1000000L)) > 0.99)
-                                {
-                                    numWaits++;
-                                }
-                            }
-                            catch (InterruptedException e)
-                            {
-                                // Restore the interrupted status
-                                Thread.currentThread().interrupt();
-                                throw new RuntimeException(e);
-                            }
-                        }
-                        else
-                        {
-                            break;
-                        }
-                    }
-                }
-            }
-
-            // Send the message either to its round robin destination, or its default destination.
-            // int num = numSent.incrementAndGet();
-            // message.setIntProperty("MSG_NUM", num);
-            setTimestamp(message);
-
-            if (destination == null)
-            {
-                _producer.send(message);
-            }
-            else
-            {
-                _producer.send(destination, message);
-            }
-
-            // Increase the unreceived size, this may actually happen after the message is received.
-            // The unreceived size is incremented by the number of consumers that will get a copy of the message,
-            // in pub/sub mode.
-            if (_maxPendingSize > 0)
-            {
-                int newUnreceivedCount = _unreceived.addAndGet(_isPubSub ? getConsumersPerDestination() : 1);
-                // log.debug("newUnreceivedCount = " + newUnreceivedCount);
-            }
-
-            // Apply message rate throttling if a rate limit has been set up.
-            if (_rateLimiter != null)
-            {
-                _rateLimiter.throttle();
-            }
-
-            // Call commit every time the commit batch size is reached.
-            boolean committed = false;
-
-            // Commit on every transaction batch size boundary. Here i + 1 is the count of actual messages sent.
-            if (((i + 1) % _txBatchSize) == 0)
-            {
-                // log.debug("Trying commit on producer session.");
-                committed = commitTx(_producerSession);
-            }
-
-            return committed;
-        }
-        finally
-        {
-            NDC.clear();
-        }
-    }
-
-    /**
-     * If the specified fail flag is set, this method waits for the user to cause a failure and then indicate to the
-     * test that the failure has occurred, before the method returns.
-     *
-     * @param failFlag The fail flag to test.
-     *
-     * @return The new value for the fail flag. If the {@link #_failOnce} flag is set, then each fail flag is only
-     *         used once, then reset.
-     */
-    private boolean waitForUserToPromptOnFailure(boolean failFlag)
-    {
-        if (failFlag)
-        {
-            if (_failOnce)
-            {
-                failFlag = false;
-            }
-
-            // log.debug("Failing Before Send");
-            waitForUser(KILL_BROKER_PROMPT);
-        }
-
-        return failFlag;
-    }
-
-    /**
-     * Implements a single iteration of the ping loop. This sends the number of pings specified by the transaction batch
-     * size property, and waits for replies to all of them. Any errors cause the publish flag to be cleared, which will
-     * terminate the pinger.
-     */
-    public void pingLoop()
-    {
-        try
-        {
-            // Generate a sample message and time stamp it.
-            Message msg = getTestMessage(_replyDestination, _messageSize, _persistent);
-            // setTimestamp(msg);
-
-            // Send the message and wait for a reply.
-            pingAndWaitForReply(msg, TX_BATCH_SIZE_DEFAULT, TIMEOUT_DEFAULT, null);
-        }
-        catch (JMSException e)
-        {
-            _publish = false;
-            // log.debug("There was a JMSException: " + e.getMessage(), e);
-        }
-        catch (InterruptedException e)
-        {
-            _publish = false;
-            // log.debug("There was an interruption: " + e.getMessage(), e);
-        }
-    }
-
-    /**
-     * Sets a chained message listener. The message listener on this pinger, chains all its messages to the one set
-     * here.
-     *
-     * @param messageListener The chained message listener.
-     */
-    public void setChainedMessageListener(ChainedMessageListener messageListener)
-    {
-        _chainedMessageListener = messageListener;
-    }
-
-    /** Removes any chained message listeners from this pinger. */
-    public void removeChainedMessageListener()
-    {
-        _chainedMessageListener = null;
-    }
-
-    /**
-     * Generates a test message of the specified size, with the specified reply-to destination and persistence flag.
-     *
-     * @param replyQueue  The reply-to destination for the message.
-     * @param messageSize The desired size of the message in bytes.
-     * @param persistent  <tt>true</tt> if the message should use persistent delivery, <tt>false</tt> otherwise.
-     *
-     * @return A freshly generated test message.
-     *
-     * @throws javax.jms.JMSException All underlying JMSException are allowed to fall through.
-     */
-    public Message getTestMessage(Destination replyQueue, int messageSize, boolean persistent) throws JMSException
-    {
-        // return TestMessageFactory.newObjectMessage(_producerSession, replyQueue, messageSize, persistent);
-        return TestUtils.createTestMessageOfSize(_producerSession, messageSize);
-    }
-
-    /**
-     * Sets the current time in nanoseconds as the timestamp on the message.
-     *
-     * @param msg The message to timestamp.
-     *
-     * @throws JMSException Any JMSExceptions are allowed to fall through.
-     */
-    protected void setTimestamp(Message msg) throws JMSException
-    {
-        /*if (((AMQSession)_producerSession).isStrictAMQP())
-        {
-            ((AMQMessage)msg).setTimestampProperty(new AMQShortString(MESSAGE_TIMESTAMP_PROPNAME), System.nanoTime());
-        }
-        else
-        {*/
-        msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
-        // }
-    }
-
-    /**
-     * Extracts the nanosecond timestamp from a message.
-     *
-     * @param msg The message to extract the time stamp from.
-     *
-     * @return The timestamp in nanos.
-     *
-     * @throws JMSException Any JMSExceptions are allowed to fall through.
-     */
-    protected long getTimestamp(Message msg) throws JMSException
-    {
-        /*if (((AMQSession)_producerSession).isStrictAMQP())
-        {
-            Long value = ((AMQMessage)msg).getTimestampProperty(new AMQShortString(MESSAGE_TIMESTAMP_PROPNAME));
-
-            return (value == null) ? 0L : value;
-        }
-        else
-        {*/
-        return msg.getLongProperty(PingPongProducer.MESSAGE_TIMESTAMP_PROPNAME);
-            // }
-    }
-
-    /**
-     * Stops the ping loop by clearing the publish flag. The current loop will complete when it notices that this flag
-     * has been cleared.
-     */
-    public void stop()
-    {
-        _publish = false;
-    }
-
-    /**
-     * Starts the producer and consumer connections.
-     *
-     * @throws JMSException Any JMSExceptions are allowed to fall through.
-     */
-    public void start() throws JMSException
-    {
-        // log.debug("public void start(): called");
-
-        _connection.start();
-        // log.debug("Producer started.");
-
-        for (int i = 0; i < _noOfConsumers; i++)
-        {
-            _consumerConnection[i].start();
-            // log.debug("Consumer " + i + " started.");
-        }
-    }
-
-    /** Implements a ping loop that repeatedly pings until the publish flag becomes false. */
-    public void run()
-    {
-        // Keep running until the publish flag is cleared.
-        while (_publish)
-        {
-            pingLoop();
-        }
-    }
-
-    /**
-     * Callback method, implementing ExceptionListener. This should be registered to listen for exceptions on the
-     * connection, this clears the publish flag which in turn will halt the ping loop.
-     *
-     * @param e The exception that triggered this callback method.
-     */
-    public void onException(JMSException e)
-    {
-        // log.debug("public void onException(JMSException e = " + e + "): called", e);
-        _publish = false;
-    }
-
-    /**
-     * Gets a shutdown hook that will cleanly shut this down when it is running the ping loop. This can be registered
-     * with the runtime system as a shutdown hook.
-     *
-     * @return A shutdown hook for the ping loop.
-     */
-    public Thread getShutdownHook()
-    {
-        return new Thread(new Runnable()
-                {
-                    public void run()
-                    {
-                        stop();
-                    }
-                });
-    }
-
-    /**
-     * Closes all of the producer and consumer connections.
-     *
-     * @throws JMSException All JMSException are allowed to fall through.
-     */
-    public void close() throws JMSException
-    {
-        // log.debug("public void close(): called");
-
-        try
-        {
-            if (_connection != null)
-            {
-                // log.debug("Before close producer connection.");
-                _connection.close();
-                // log.debug("Closed producer connection.");
-            }
-
-            for (int i = 0; i < _noOfConsumers; i++)
-            {
-                if (_consumerConnection[i] != null)
-                {
-                    // log.debug("Before close consumer connection " + i + ".");
-                    _consumerConnection[i].close();
-                    // log.debug("Closed consumer connection " + i + ".");
-                }
-            }
-        }
-        finally
-        {
-            _connection = null;
-            _producerSession = null;
-            _consumerSession = null;
-            _consumerConnection = null;
-            _producer = null;
-            _consumer = null;
-            _pingDestinations = null;
-            _replyDestination = null;
-        }
-    }
-
-    /**
-     * Convenience method to commit the transaction on the specified controlSession. If the controlSession to commit on is not a
-     * transactional controlSession, this method does nothing (unless the failover after send flag is set).
-     *
-     * <p/>If the {@link #_failAfterSend} flag is set, this will prompt the user to kill the broker before the commit is
-     * applied. This flag applies whether the pinger is transactional or not.
-     *
-     * <p/>If the {@link #_failBeforeCommit} flag is set, this will prompt the user to kill the broker before the commit
-     * is applied. If the {@link #_failAfterCommit} flag is set, this will prompt the user to kill the broker after the
-     * commit is applied. These flags will only apply if using a transactional pinger.
-     *
-     * @param session The controlSession to commit
-     *
-     * @return <tt>true</tt> if the controlSession was committed, <tt>false</tt> if it was not.
-     *
-     * @throws javax.jms.JMSException If the commit fails and then the rollback fails.
-     *
-     * @todo Consider moving the fail after send logic into the send method. It is confusing to have it in this commit
-     * method, because commits only apply to transactional pingers, but fail after send applied to transactional and
-     * non-transactional alike.
-     */
-    protected boolean commitTx(Session session) throws JMSException
-    {
-        // log.debug("protected void commitTx(Session session): called");
-
-        boolean committed = false;
-
-        _failAfterSend = waitForUserToPromptOnFailure(_failAfterSend);
-
-        if (session.getTransacted())
-        {
-            // log.debug("Session is transacted.");
-
-            try
-            {
-                _failBeforeCommit = waitForUserToPromptOnFailure(_failBeforeCommit);
-
-                long start = System.nanoTime();
-                session.commit();
-                committed = true;
-                // log.debug("Time taken to commit :" + ((System.nanoTime() - start) / 1000000f) + " ms");
-
-                _failAfterCommit = waitForUserToPromptOnFailure(_failAfterCommit);
-
-                // log.debug("Session Commited.");
-            }
-            catch (JMSException e)
-            {
-                // log.debug("JMSException on commit:" + e.getMessage(), e);
-
-                try
-                {
-                    session.rollback();
-                    // log.debug("Message rolled back.");
-                }
-                catch (JMSException jmse)
-                {
-                    // log.debug("JMSE on rollback:" + jmse.getMessage(), jmse);
-
-                    // Both commit and rollback failed. Throw the rollback exception.
-                    throw jmse;
-                }
-            }
-        }
-
-        return committed;
-    }
-
-    /**
-     * Outputs a prompt to the console and waits for the user to press return.
-     *
-     * @param prompt The prompt to display on the console.
-     */
-    public void waitForUser(String prompt)
-    {
-        System.out.println(prompt);
-
-        try
-        {
-            System.in.read();
-        }
-        catch (IOException e)
-        {
-            // Ignored.
-        }
-
-        System.out.println("Continuing.");
-    }
-
-    /**
-     * Gets the number of consumers that are listening to each destination in the test.
-     *
-     * @return int The number of consumers subscribing to each topic.
-     */
-    public int getConsumersPerDestination()
-    {
-        return _noOfConsumers;
-    }
-
-    /**
-     * Calculates how many pings are expected to be received for the given number sent.
-     *
-     * Note : that if you have set noConsumers to 0 then this will also return 0
-     * in the case of PubSub testing. This is correct as without consumers there
-     * will be no-one to receive the sent messages so they will be unable to respond. 
-     *
-     * @param numpings The number of pings that will be sent.
-     *
-     * @return The number that should be received, for the test to pass.
-     */
-    public int getExpectedNumPings(int numpings)
-    {
-        // Wow, I'm freaking sorry about this return here...
-        return ((_failAfterSend || _failBeforeCommit) ? numpings - 1: numpings) *
-                                    (_isPubSub ? getConsumersPerDestination() : 1);
-    }
-
-    /**
-     * Defines a chained message listener interface that can be attached to this pinger. Whenever this pinger's {@link
-     * PingPongProducer#onMessageWithConsumerNo} method is called, the chained listener set through the {@link
-     * PingPongProducer#setChainedMessageListener} method is passed the message, and the remaining expected count of
-     * messages with that correlation id.
-     *
-     * <p/>Provided only one pinger is producing messages with that correlation id, the chained listener will always be
-     * given unique message counts. It will always be called while the producer waiting for all messages to arrive is
-     * still blocked.
-     */
-    public static interface ChainedMessageListener
-    {
-        /**
-         * Notifies interested listeners about message arrival and important test stats, the number of messages
-         * remaining in the test, and the messages send timestamp.
-         *
-         * @param message        The newly arrived message.
-         * @param remainingCount The number of messages left to complete the test.
-         * @param latency        The nanosecond latency of the message.
-         *
-         * @throws JMSException Any JMS exceptions is allowed to fall through.
-         */
-        public void onMessage(Message message, int remainingCount, long latency) throws JMSException;
-    }
-
-    /**
-     * Holds information on each correlation id. The countdown latch, the current timeout timer... More stuff to be
-     * added to this: read/write lock to make onMessage more concurrent as described in class header comment.
-     */
-    protected static class PerCorrelationId
-    {
-        /** Holds a countdown on number of expected messages. */
-        CountDownLatch trafficLight;
-
-        /** Holds the last timestamp that the timeout was reset to. */
-        Long timeOutStart;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java
deleted file mode 100644
index 009254c612..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java
+++ /dev/null
@@ -1,251 +0,0 @@
-/*
- *
- * 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.requestreply;
-
-import junit.framework.Assert;
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.junit.extensions.AsymptoticTestCase;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-import org.apache.qpid.junit.extensions.util.TestContextProperties;
-
-import javax.jms.*;
-
-/**
- * PingPongTestPerf is a full round trip ping test, that has been written with the intention of being scaled up to run
- * many times simultaneously to simluate many clients/producer/connections. A full round trip ping sends a message from
- * a producer to a conumer, then the consumer replies to the message on a temporary queue.
- *
- * <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of the number
- * of pings specified by the test size and time how long it takes for all of these to complete. This test may be scaled
- * up using a suitable JUnit test runner. See {@link org.apache.qpid.junit.extensions.TKTestRunner} for more
- * information on how to do this.
- *
- * <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a
- * temporary queue for replies. This setup is only established once for all the test repeats, but each test threads
- * gets its own connection/producer/consumer, this is only re-established if the connection is lost.
- *
- * <p/>The test cycle is: Connects to a queue, creates a temporary queue, creates messages containing a property that
- * is the name of the temporary queue, fires off many messages on the original queue and waits for them all to come
- * back on the temporary queue.
- *
- * <p/>Configurable test properties: message size, transacted or not, persistent or not. Broker connection details.
- *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * </table>
- */
-public class PingPongTestPerf extends AsymptoticTestCase
-{
-    private static Logger _logger = Logger.getLogger(PingPongTestPerf.class);
-
-    /** Thread local to hold the per-thread test setup fields. */
-    ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
-
-    // Set up a property reader to extract the test parameters from. Once ContextualProperties is available in
-    // the project dependencies, use it to get property overrides for configurable tests and to notify the test runner
-    // of the test parameters to log with the results. It also providers some basic type parsing convenience methods.
-    // private Properties testParameters = System.getProperties();
-    private ParsedProperties testParameters =
-        TestContextProperties.getInstance(PingPongProducer.defaults /*System.getProperties()*/);
-
-    public PingPongTestPerf(String name)
-    {
-        super(name);
-
-        _logger.debug(testParameters);
-
-        // Sets up the test parameters with defaults.
-        /*testParameters.setPropertyIfNull(PingPongProducer.TX_BATCH_SIZE_PROPNAME,
-            Integer.toString(PingPongProducer.TX_BATCH_SIZE_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.MESSAGE_SIZE_PROPNAME,
-            Integer.toString(PingPongProducer.MESSAGE_SIZE_DEAFULT));
-        testParameters.setPropertyIfNull(PingPongProducer.PING_QUEUE_NAME_PROPNAME,
-            PingPongProducer.PING_QUEUE_NAME_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.PERSISTENT_MODE_PROPNAME,
-            Boolean.toString(PingPongProducer.PERSISTENT_MODE_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.TRANSACTED_PROPNAME,
-            Boolean.toString(PingPongProducer.TRANSACTED_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.BROKER_PROPNAME, PingPongProducer.BROKER_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.USERNAME_PROPNAME, PingPongProducer.USERNAME_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.PASSWORD_PROPNAME, PingPongProducer.PASSWORD_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.VIRTUAL_HOST_PROPNAME, PingPongProducer.VIRTUAL_HOST_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.VERBOSE_PROPNAME,
-            Boolean.toString(PingPongProducer.VERBOSE_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.RATE_PROPNAME, Integer.toString(PingPongProducer.RATE_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.PUBSUB_PROPNAME,
-            Boolean.toString(PingPongProducer.PUBSUB_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.TX_BATCH_SIZE_PROPNAME,
-            Integer.toString(PingPongProducer.TX_BATCH_SIZE_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.TIMEOUT_PROPNAME, Long.toString(PingPongProducer.TIMEOUT_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.DESTINATION_COUNT_PROPNAME,
-            Integer.toString(PingPongProducer.DESTINATION_COUNT_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.FAIL_AFTER_COMMIT_PROPNAME,
-            PingPongProducer.FAIL_AFTER_COMMIT_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.FAIL_BEFORE_COMMIT_PROPNAME,
-            PingPongProducer.FAIL_BEFORE_COMMIT_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.FAIL_AFTER_SEND_PROPNAME,
-            PingPongProducer.FAIL_AFTER_SEND_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.FAIL_BEFORE_SEND_PROPNAME,
-            PingPongProducer.FAIL_BEFORE_SEND_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.FAIL_ONCE_PROPNAME, PingPongProducer.FAIL_ONCE_DEFAULT);
-        testParameters.setPropertyIfNull(PingPongProducer.UNIQUE_DESTS_PROPNAME,
-            Boolean.toString(PingPongProducer.UNIQUE_DESTS_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.ACK_MODE_PROPNAME,
-            Integer.toString(PingPongProducer.ACK_MODE_DEFAULT));
-        testParameters.setPropertyIfNull(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME,
-            PingPongProducer.PAUSE_AFTER_BATCH_DEFAULT);*/
-    }
-
-    /**
-     * Compile all the tests into a test suite.
-     */
-    public static Test suite()
-    {
-        // Build a new test suite
-        TestSuite suite = new TestSuite("Ping-Pong Performance Tests");
-
-        // Run performance tests in read committed mode.
-        suite.addTest(new PingPongTestPerf("testPingPongOk"));
-
-        return suite;
-    }
-
-    private static void setSystemPropertyIfNull(String propName, String propValue)
-    {
-        if (System.getProperty(propName) == null)
-        {
-            System.setProperty(propName, propValue);
-        }
-    }
-
-    public void testPingPongOk(int numPings) throws Exception
-    {
-        // Get the per thread test setup to run the test through.
-        PerThreadSetup perThreadSetup = threadSetup.get();
-
-        // Generate a sample message. This message is already time stamped and has its reply-to destination set.
-        Message msg =
-            perThreadSetup._testPingProducer.getTestMessage(perThreadSetup._testPingProducer.getReplyDestinations().get(0),
-                testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
-                testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
-
-        // Send the message and wait for a reply.
-        int numReplies =
-            perThreadSetup._testPingProducer.pingAndWaitForReply(msg, numPings, PingPongProducer.TIMEOUT_DEFAULT, null);
-
-        // Fail the test if the timeout was exceeded.
-        if (numReplies != numPings)
-        {
-            Assert.fail("The ping timed out, got " + numReplies + " out of " + numPings);
-        }
-    }
-
-    /**
-     * Performs test fixture creation on a per thread basis. This will only be called once for each test thread.
-     */
-    public void threadSetUp()
-    {
-        try
-        {
-            PerThreadSetup perThreadSetup = new PerThreadSetup();
-
-            // Extract the test set up paramaeters.
-            String brokerDetails = testParameters.getProperty(PingPongProducer.BROKER_PROPNAME);
-            String username = testParameters.getProperty(PingPongProducer.USERNAME_PROPNAME);
-            String password = testParameters.getProperty(PingPongProducer.PASSWORD_PROPNAME);
-            String virtualPath = testParameters.getProperty(PingPongProducer.VIRTUAL_HOST_PROPNAME);
-            String destinationName = testParameters.getProperty(PingPongProducer.PING_QUEUE_NAME_PROPNAME);
-            boolean persistent = testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME);
-            boolean transacted = testParameters.getPropertyAsBoolean(PingPongProducer.TRANSACTED_PROPNAME);
-            String selector = testParameters.getProperty(PingPongProducer.SELECTOR_PROPNAME);
-            boolean verbose = testParameters.getPropertyAsBoolean(PingPongProducer.VERBOSE_PROPNAME);
-            boolean pubsub = testParameters.getPropertyAsBoolean(PingPongProducer.PUBSUB_PROPNAME);
-
-            synchronized (this)
-            {
-                // Establish a bounce back client on the ping queue to bounce back the pings.
-                perThreadSetup._testPingBouncer =
-                    new PingPongBouncer(brokerDetails, username, password, virtualPath, destinationName, persistent,
-                        transacted, selector, verbose, pubsub);
-
-                // Start the connections for client and producer running.
-                perThreadSetup._testPingBouncer.getConnection().start();
-
-                // Establish a ping-pong client on the ping queue to send the pings and receive replies with.
-                perThreadSetup._testPingProducer = new PingPongProducer(testParameters);
-                perThreadSetup._testPingProducer.establishConnection(true, true);
-                perThreadSetup._testPingProducer.start();
-            }
-
-            // Attach the per-thread set to the thread.
-            threadSetup.set(perThreadSetup);
-        }
-        catch (Exception e)
-        {
-            _logger.warn("There was an exception during per thread setup.", e);
-        }
-    }
-
-    /**
-     * Performs test fixture clean
-     */
-    public void threadTearDown()
-    {
-        _logger.debug("public void threadTearDown(): called");
-
-        try
-        {
-            // Get the per thread test fixture.
-            PerThreadSetup perThreadSetup = threadSetup.get();
-
-            // Close the pingers so that it cleans up its connection cleanly.
-            synchronized (this)
-            {
-                perThreadSetup._testPingProducer.close();
-                // perThreadSetup._testPingBouncer.close();
-            }
-
-            // Ensure the per thread fixture is reclaimed.
-            threadSetup.remove();
-        }
-        catch (JMSException e)
-        {
-            _logger.warn("There was an exception during per thread tear down.");
-        }
-    }
-
-    protected static class PerThreadSetup
-    {
-        /**
-         * Holds the test ping-pong producer.
-         */
-        private PingPongProducer _testPingProducer;
-
-        /**
-         * Holds the test ping client.
-         */
-        private PingPongBouncer _testPingBouncer;
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Config.java b/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Config.java
deleted file mode 100644
index d5c0979399..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Config.java
+++ /dev/null
@@ -1,326 +0,0 @@
-/*
- *
- * 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.topic;
-
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.config.ConnectorConfig;
-import org.apache.qpid.config.Connector;
-import org.apache.qpid.config.AbstractConfig;
-
-import javax.jms.Connection;
-
-public class Config extends AbstractConfig implements ConnectorConfig
-{
-
-    private String host = "localhost";
-    private int port = 5672;
-    private String factory = null;
-
-    private int payload = 256;
-    private int messages = 1000;
-    private int clients = 1;
-    private int batch = 1;
-    private long delay = 1;
-    private int warmup;
-    private int ackMode= AMQSession.NO_ACKNOWLEDGE;
-    private String clientId;
-    private String subscriptionId;
-    private String selector;
-    private String destinationName;
-    private boolean persistent;
-    private boolean transacted;
-    private int destinationsCount;
-    private int batchSize;
-    private int rate;
-    private boolean ispubsub;
-    private long timeout;
-
-    public Config()
-    {
-    }
-
-    public int getAckMode()
-    {
-        return ackMode;
-    }
-
-    public void setPayload(int payload)
-    {
-        this.payload = payload;
-    }
-
-    public int getPayload()
-    {
-        return payload;
-    }
-
-    void setClients(int clients)
-    {
-        this.clients = clients;
-    }
-
-    int getClients()
-    {
-        return clients;
-    }
-
-    void setMessages(int messages)
-    {
-        this.messages = messages;
-    }
-
-    public int getMessages()
-    {
-        return messages;
-    }
-
-    public int getBatchSize()
-    {
-        return batchSize;
-    }
-
-    public int getRate()
-    {
-        return rate;
-    }
-
-    public int getDestinationsCount()
-    {
-        return destinationsCount;
-    }
-
-    public String getHost()
-    {
-        return host;
-    }
-
-    public void setHost(String host)
-    {
-        this.host = host;
-    }
-
-    public int getPort()
-    {
-        return port;
-    }
-
-    public String getFactory()
-    {
-        return factory;
-    }
-
-    public void setPort(int port)
-    {
-        this.port = port;
-    }
-
-    int getBatch()
-    {
-        return batch;
-    }
-
-    void setBatch(int batch)
-    {
-        this.batch = batch;
-    }
-
-    int getWarmup()
-    {
-        return warmup;
-    }
-
-    void setWarmup(int warmup)
-    {
-        this.warmup = warmup;
-    }
-
-    public long getDelay()
-    {
-        return delay;
-    }
-
-    public void setDelay(long delay)
-    {
-        this.delay = delay;
-    }
-
-    public long getTimeout()
-    {
-        return timeout;
-    }
-
-    public void setTimeout(long time)
-    {
-        this.timeout = time;
-    }
-
-    public String getClientId()
-    {
-        return clientId;
-    }
-
-    public String getSubscriptionId()
-    {
-        return subscriptionId;
-    }
-
-    public String getSelector()
-    {
-        return selector;
-    }
-
-    public String getDestination()
-    {
-        return destinationName;
-    }
-
-    public boolean usePersistentMessages()
-    {
-        return persistent;
-    }
-
-    public boolean isTransacted()
-    {
-        return transacted;
-    }
-
-    public boolean isPubSub()
-    {
-        return ispubsub;
-    }
-
-    public void setOption(String key, String value)
-    {
-        if("-host".equalsIgnoreCase(key))
-        {
-            setHost(value);
-        }
-        else if("-port".equalsIgnoreCase(key))
-        {
-            try
-            {
-                setPort(Integer.parseInt(value));
-            }
-            catch(NumberFormatException e)
-            {
-                throw new RuntimeException("Bad port number: " + value, e);
-            }
-        }
-        else if("-payload".equalsIgnoreCase(key))
-        {
-            setPayload(parseInt("Bad payload size", value));
-        }
-        else if("-messages".equalsIgnoreCase(key))
-        {
-            setMessages(parseInt("Bad message count", value));
-        }
-        else if("-clients".equalsIgnoreCase(key))
-        {
-            setClients(parseInt("Bad client count", value));
-        }
-        else if("-batch".equalsIgnoreCase(key))
-        {
-            setBatch(parseInt("Bad batch count", value));
-        }
-        else if("-delay".equalsIgnoreCase(key))
-        {
-            setDelay(parseLong("Bad batch delay", value));
-        }
-        else if("-warmup".equalsIgnoreCase(key))
-        {
-            setWarmup(parseInt("Bad warmup count", value));
-        }
-        else if("-ack".equalsIgnoreCase(key))
-        {
-            ackMode = parseInt("Bad ack mode", value);
-        }
-        else if("-factory".equalsIgnoreCase(key))
-        {
-            factory = value;
-        }
-        else if("-clientId".equalsIgnoreCase(key))
-        {
-            clientId = value;
-        }
-        else if("-subscriptionId".equalsIgnoreCase(key))
-        {
-            subscriptionId = value;
-        }
-        else if("-persistent".equalsIgnoreCase(key))
-        {
-            persistent = "true".equalsIgnoreCase(value);
-        }
-        else if("-transacted".equalsIgnoreCase(key))
-        {
-            transacted = "true".equalsIgnoreCase(value);
-        }
-        else if ("-destinationscount".equalsIgnoreCase(key))
-        {
-            destinationsCount = parseInt("Bad destinations count", value);
-        }
-        else if ("-batchsize".equalsIgnoreCase(key))
-        {
-            batchSize = parseInt("Bad batch size", value);
-        }
-        else if ("-rate".equalsIgnoreCase(key))
-        {
-            rate = parseInt("MEssage rate", value);
-        }
-        else if("-pubsub".equalsIgnoreCase(key))
-        {
-            ispubsub = "true".equalsIgnoreCase(value);
-        }
-        else if("-selector".equalsIgnoreCase(key))
-        {
-            selector = value;
-        }
-        else if("-destinationname".equalsIgnoreCase(key))
-        {
-            destinationName = value;
-        }
-        else if("-timeout".equalsIgnoreCase(key))
-        {
-            setTimeout(parseLong("Bad timeout data", value));
-        }
-        else
-        {
-            System.out.println("Ignoring unrecognised option: " + key);
-        }
-    }
-
-    static String getAckModeDescription(int ackMode)
-    {
-        switch(ackMode)
-        {
-            case AMQSession.NO_ACKNOWLEDGE: return "NO_ACKNOWLEDGE";
-            case AMQSession.AUTO_ACKNOWLEDGE: return "AUTO_ACKNOWLEDGE";
-            case AMQSession.CLIENT_ACKNOWLEDGE: return "CLIENT_ACKNOWLEDGE";
-            case AMQSession.DUPS_OK_ACKNOWLEDGE: return "DUPS_OK_ACKNOWELDGE";
-            case AMQSession.PRE_ACKNOWLEDGE: return "PRE_ACKNOWLEDGE";
-        }
-        return "AckMode=" + ackMode;
-    }
-
-    public Connection createConnection() throws Exception
-    {
-        return new Connector().createConnection(this);
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Listener.java b/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Listener.java
deleted file mode 100644
index 6dcea42bfe..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Listener.java
+++ /dev/null
@@ -1,303 +0,0 @@
-/*
- *
- * 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.topic;
-
-import java.util.Random;
-
-import javax.jms.*;
-
-import org.apache.log4j.Logger;
-import org.apache.log4j.NDC;
-
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.exchange.ExchangeDefaults;
-
-/**
- * This class has not kept up to date with the topic_listener in the cpp tests. It should provide identical behaviour for
- * cross testing the java and cpp clients.
- *
- * <p/>How the cpp topic_publisher operates:
- * It publishes text messages to the default topic exchange, on virtual host "/test", on the topic "topic_control", for
- * the specified number of test messages to be sent.
- * It publishes a report request message (on same topic), with the header text field "TYPE", value "REPORT_REQUEST",
- * optionally within a transaction, and waits for the specified number of consumers to reply to this request. The
- * listeners should reply to this message on a queue named "response", on virtual host "/test", with some sort of message
- * about the number of messages received and how long it took, although the publisher never looks at the message content.
- * The publisher then send a message (on the same topic), with the header text field "TYPE", value "TERMINATION_REQUEST",
- * which the listener should close its connection and terminate upon receipt of.
- *
- * @todo I've added lots of field table types in the report message, just to check if the other end can decode them
- *       correctly. Not really the right place to test this, so remove them from
- *       {@link #createReportResponseMessage(String)} once a better test exists.
- */
-public class Listener implements MessageListener
-{
-    private static Logger log = Logger.getLogger(Listener.class);
-
-    public static final String CONTROL_TOPIC = "topic_control";
-    public static final String RESPONSE_QUEUE = "response";
-
-    private final Topic _topic;
-    //private final Topic _control;
-
-    private final Queue _response;
-
-    /** Holds the connection to listen on. */
-    private final Connection _connection;
-
-    /** Holds the producer to send control messages on. */
-    private final MessageProducer _controller;
-
-    /** Holds the JMS session. */
-    private final javax.jms.Session _session;
-
-    /** Holds a flag to indicate that a timer has begun on the first message. Reset when report is sent. */
-    private boolean init;
-
-    /** Holds the count of messages received by this listener. */
-    private int count;
-
-    /** Used to hold the start time of the first message. */
-    private long start;
-    private static String clientId;
-
-    Listener(Connection connection, int ackMode, String name) throws Exception
-    {
-        log.debug("Listener(Connection connection = " + connection + ", int ackMode = " + ackMode + ", String name = " + name
-                  + "): called");
-
-        _connection = connection;
-        _session = connection.createSession(false, ackMode);
-
-        if (_session instanceof AMQSession)
-        {
-            _topic = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, CONTROL_TOPIC);
-            //_control = new AMQTopic(CONTROL_TOPIC);
-            _response = new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, RESPONSE_QUEUE);
-        }
-        else
-        {
-            _topic = _session.createTopic(CONTROL_TOPIC);
-            //_control = _session.createTopic(CONTROL_TOPIC);
-            _response = _session.createQueue(RESPONSE_QUEUE);
-        }
-
-        //register for events
-        if (name == null)
-        {
-            log.debug("Calling _factory.createTopicConsumer().setMessageListener(this)");
-            createTopicConsumer().setMessageListener(this);
-        }
-        else
-        {
-            log.debug("Calling createDurableTopicConsumer(name).setMessageListener(this)");
-            createDurableTopicConsumer(name).setMessageListener(this);
-        }
-
-        _connection.start();
-
-        _controller = createControlPublisher();
-        System.out.println("Waiting for messages " + Config.getAckModeDescription(ackMode)
-                           +
-                           ((name == null)
-                            ? "" : (" (subscribed with name " + name + " and client id " + connection.getClientID() + ")"))
-                           + "...");
-    }
-
-    public static void main(String[] argv) throws Exception
-    {
-        clientId = "Listener-" + System.currentTimeMillis();
-
-        NDC.push(clientId);
-
-        Config config = new Config();
-        config.setOptions(argv);
-
-        //Connection con = config.createConnection();
-        Connection con =
-            new AMQConnection("amqp://guest:guest@testid/test?brokerlist='" + config.getHost() + ":" + config.getPort()
-                              + "'");
-
-        if (config.getClientId() != null)
-        {
-            con.setClientID(config.getClientId());
-        }
-
-        new Listener(con, config.getAckMode(), config.getSubscriptionId());
-
-        NDC.pop();
-        NDC.remove();
-    }
-
-    /**
-     * Checks whether or not a text field on a message has the specified value.
-     *
-     * @param m         The message to check.
-     * @param fieldName The name of the field to check.
-     * @param value     The expected value of the field to compare with.
-     *
-     * @return <tt>true</tt>If the specified field has the specified value, <tt>fals</tt> otherwise.
-     *
-     * @throws JMSException Any JMSExceptions are allowed to fall through.
-     */
-    private static boolean checkTextField(Message m, String fieldName, String value) throws JMSException
-    {
-        log.debug("private static boolean checkTextField(Message m = " + m + ", String fieldName = " + fieldName
-                  + ", String value = " + value + "): called");
-
-        String comp = m.getStringProperty(fieldName);
-        log.debug("comp = " + comp);
-
-        boolean result = (comp != null) && comp.equals(value);
-        log.debug("result = " + result);
-
-        return result;
-    }
-
-    public void onMessage(Message message)
-    {
-        NDC.push(clientId);
-
-        log.debug("public void onMessage(Message message = " + message + "): called");
-
-        if (!init)
-        {
-            start = System.nanoTime() / 1000000;
-            count = 0;
-            init = true;
-        }
-
-        try
-        {
-            if (isShutdown(message))
-            {
-                log.debug("Got a shutdown message.");
-                shutdown();
-            }
-            else if (isReport(message))
-            {
-                log.debug("Got a report request message.");
-
-                // Send the report.
-                report();
-                init = false;
-            }
-        }
-        catch (JMSException e)
-        {
-            log.warn("There was a JMSException during onMessage.", e);
-        }
-        finally
-        {
-            NDC.pop();
-        }
-    }
-
-    Message createReportResponseMessage(String msg) throws JMSException
-    {
-        Message message = _session.createTextMessage(msg);
-
-        // Shove some more field table type in the message just to see if the other end can handle it.
-        message.setBooleanProperty("BOOLEAN", true);
-        message.setByteProperty("BYTE", (byte) 5);
-        message.setDoubleProperty("DOUBLE", Math.PI);
-        message.setFloatProperty("FLOAT", 1.0f);
-        message.setIntProperty("INT", 1);
-        message.setShortProperty("SHORT", (short) 1);
-        message.setLongProperty("LONG", (long) 1827361278);
-        message.setStringProperty("STRING", "hello");
-
-        return message;
-    }
-
-    boolean isShutdown(Message m) throws JMSException
-    {
-        boolean result = checkTextField(m, "TYPE", "TERMINATION_REQUEST");
-
-        //log.debug("isShutdown = " + result);
-
-        return result;
-    }
-
-    boolean isReport(Message m) throws JMSException
-    {
-        boolean result = checkTextField(m, "TYPE", "REPORT_REQUEST");
-
-        //log.debug("isReport = " + result);
-
-        return result;
-    }
-
-    MessageConsumer createTopicConsumer() throws Exception
-    {
-        return _session.createConsumer(_topic);
-    }
-
-    MessageConsumer createDurableTopicConsumer(String name) throws Exception
-    {
-        return _session.createDurableSubscriber(_topic, name);
-    }
-
-    MessageProducer createControlPublisher() throws Exception
-    {
-        return _session.createProducer(_response);
-    }
-
-    private void shutdown()
-    {
-        try
-        {
-            _session.close();
-            _connection.stop();
-            _connection.close();
-        }
-        catch (Exception e)
-        {
-            e.printStackTrace(System.out);
-        }
-    }
-
-    private void report()
-    {
-        log.debug("private void report(): called");
-
-        try
-        {
-            String msg = getReport();
-            _controller.send(createReportResponseMessage(msg));
-            log.debug("Sent report: " + msg);
-        }
-        catch (Exception e)
-        {
-            e.printStackTrace(System.out);
-        }
-    }
-
-    private String getReport()
-    {
-        long time = ((System.nanoTime() / 1000000) - start);
-
-        return "Received " + count + " in " + time + "ms";
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/MessageFactory.java b/qpid/java/perftests/src/main/java/org/apache/qpid/topic/MessageFactory.java
deleted file mode 100644
index 4efdc1cb56..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/MessageFactory.java
+++ /dev/null
@@ -1,157 +0,0 @@
-/*
- *
- * 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.topic;
-
-import javax.jms.*;
-
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.exchange.ExchangeDefaults;
-
-/**
- */
-class MessageFactory
-{
-    private static final char[] DATA = "abcdefghijklmnopqrstuvwxyz".toCharArray();
-
-    private final Session _session;
-    private final Topic _topic;
-    private final Topic _control;
-    private final byte[] _payload;
-
-    MessageFactory(Session session) throws JMSException
-    {
-        this(session, 256);
-    }
-
-    MessageFactory(Session session, int size) throws JMSException
-    {
-        _session = session;
-        if (session instanceof AMQSession)
-        {
-            _topic = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, "topic_control");
-            _control = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, "topictest.control");
-        }
-        else
-        {
-            _topic = session.createTopic("topic_control");
-            _control = session.createTopic("topictest.control");
-        }
-
-        _payload = new byte[size];
-
-        for (int i = 0; i < size; i++)
-        {
-            _payload[i] = (byte) DATA[i % DATA.length];
-        }
-    }
-
-    private static boolean checkText(Message m, String s)
-    {
-        try
-        {
-            return (m instanceof TextMessage) && ((TextMessage) m).getText().equals(s);
-        }
-        catch (JMSException e)
-        {
-            e.printStackTrace(System.out);
-
-            return false;
-        }
-    }
-
-    Topic getTopic()
-    {
-        return _topic;
-    }
-
-    Message createEventMessage() throws JMSException
-    {
-        BytesMessage msg = _session.createBytesMessage();
-        msg.writeBytes(_payload);
-
-        return msg;
-    }
-
-    Message createShutdownMessage() throws JMSException
-    {
-        return _session.createTextMessage("SHUTDOWN");
-    }
-
-    Message createReportRequestMessage() throws JMSException
-    {
-        return _session.createTextMessage("REPORT");
-    }
-
-    Message createReportResponseMessage(String msg) throws JMSException
-    {
-        return _session.createTextMessage(msg);
-    }
-
-    boolean isShutdown(Message m)
-    {
-        return checkText(m, "SHUTDOWN");
-    }
-
-    boolean isReport(Message m)
-    {
-        return checkText(m, "REPORT");
-    }
-
-    Object getReport(Message m)
-    {
-        try
-        {
-            return ((TextMessage) m).getText();
-        }
-        catch (JMSException e)
-        {
-            e.printStackTrace(System.out);
-
-            return e.toString();
-        }
-    }
-
-    MessageConsumer createTopicConsumer() throws Exception
-    {
-        return _session.createConsumer(_topic);
-    }
-
-    MessageConsumer createDurableTopicConsumer(String name) throws Exception
-    {
-        return _session.createDurableSubscriber(_topic, name);
-    }
-
-    MessageConsumer createControlConsumer() throws Exception
-    {
-        return _session.createConsumer(_control);
-    }
-
-    MessageProducer createTopicPublisher() throws Exception
-    {
-        return _session.createProducer(_topic);
-    }
-
-    MessageProducer createControlPublisher() throws Exception
-    {
-        return _session.createProducer(_control);
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Publisher.java b/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Publisher.java
deleted file mode 100644
index c3b19b558a..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/Publisher.java
+++ /dev/null
@@ -1,186 +0,0 @@
-/*
- *
- * 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.topic;
-
-import javax.jms.*;
-
-public class Publisher implements MessageListener
-{
-    private final Object _lock = new Object();
-    private final Connection _connection;
-    private final Session _session;
-    private final MessageFactory _factory;
-    private final MessageProducer _publisher;
-    private int _count;
-
-    Publisher(Connection connection, int size, int ackMode, boolean persistent) throws Exception
-    {
-        _connection = connection;
-        _session = _connection.createSession(false, ackMode);
-        _factory = new MessageFactory(_session, size);
-        _publisher = _factory.createTopicPublisher();
-        _publisher.setDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
-        System.out.println("Publishing " + (persistent ? "persistent" : "non-persistent") + " messages of " + size + " bytes, " + Config.getAckModeDescription(ackMode) + ".");
-    }
-
-    private void test(Config config) throws Exception
-    {
-        test(config.getBatch(), config.getDelay(), config.getMessages(), config.getClients(), config.getWarmup());
-    }
-
-    private void test(int batches, long delay, int msgCount, int consumerCount, int warmup) throws Exception
-    {
-        _factory.createControlConsumer().setMessageListener(this);
-        _connection.start();
-
-        if (warmup > 0)
-        {
-            System.out.println("Runing warmup (" + warmup + " msgs)");
-            long time = batch(warmup, consumerCount);
-            System.out.println("Warmup completed in " + time + "ms");
-        }
-
-        long[] times = new long[batches];
-        for (int i = 0; i < batches; i++)
-        {
-            if (i > 0)
-            {
-                Thread.sleep(delay * 1000);
-            }
-            times[i] = batch(msgCount, consumerCount);
-            System.out.println("Batch " + (i + 1) + " of " + batches + " completed in " + times[i] + " ms.");
-        }
-
-        long min = min(times);
-        long max = max(times);
-        System.out.println("min: " + min + ", max: " + max + " avg: " + avg(times, min, max));
-
-        //request shutdown
-        _publisher.send(_factory.createShutdownMessage());
-
-        _connection.stop();
-        _connection.close();
-    }
-
-    private long batch(int msgCount, int consumerCount) throws Exception
-    {
-        _count = consumerCount;
-        long start = System.currentTimeMillis();
-        publish(msgCount);
-        waitForCompletion(consumerCount);
-        return System.currentTimeMillis() - start;
-    }
-
-    private void publish(int count) throws Exception
-    {
-
-        //send events
-        for (int i = 0; i < count; i++)
-        {
-            _publisher.send(_factory.createEventMessage());
-            if ((i + 1) % 100 == 0)
-            {
-                System.out.println("Sent " + (i + 1) + " messages");
-            }
-        }
-
-        //request report
-        _publisher.send(_factory.createReportRequestMessage());
-    }
-
-    private void waitForCompletion(int consumers) throws Exception
-    {
-        System.out.println("Waiting for completion...");
-        synchronized (_lock)
-        {
-            while (_count > 0)
-            {
-                _lock.wait();
-            }
-        }
-    }
-
-
-    public void onMessage(Message message)
-    {
-        System.out.println("Received report " + _factory.getReport(message) + " " + --_count + " remaining");
-        if (_count == 0)
-        {
-            synchronized (_lock)
-            {
-                _lock.notify();
-            }
-        }
-    }
-
-    static long min(long[] times)
-    {
-        long min = times.length > 0 ? times[0] : 0;
-        for (int i = 0; i < times.length; i++)
-        {
-            min = Math.min(min, times[i]);
-        }
-        return min;
-    }
-
-    static long max(long[] times)
-    {
-        long max = times.length > 0 ? times[0] : 0;
-        for (int i = 0; i < times.length; i++)
-        {
-            max = Math.max(max, times[i]);
-        }
-        return max;
-    }
-
-    static long avg(long[] times, long min, long max)
-    {
-        long sum = 0;
-        for (int i = 0; i < times.length; i++)
-        {
-            sum += times[i];
-        }
-
-        int adjustment = 0;
-
-        // Remove min and max if we have run enough batches.
-        if (times.length > 2)
-        {
-            sum -= min;
-            sum -= max;
-            adjustment = 2;
-        }
-
-        return (sum / (times.length - adjustment));
-    }
-
-    public static void main(String[] argv) throws Exception
-    {
-        Config config = new Config();
-        config.setOptions(argv);
-
-        Connection con = config.createConnection();
-        int size = config.getPayload();
-        int ackMode = config.getAckMode();
-        boolean persistent = config.usePersistentMessages();
-        new Publisher(con, size, ackMode, persistent).test(config);
-    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/TopicWithSelectorsTransientVolumeTest.java b/qpid/java/perftests/src/main/java/org/apache/qpid/topic/TopicWithSelectorsTransientVolumeTest.java
deleted file mode 100644
index 1d6ec07bb0..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/TopicWithSelectorsTransientVolumeTest.java
+++ /dev/null
@@ -1,344 +0,0 @@
-/*
- * 
- * 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.topic;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-
-import javax.jms.BytesMessage;
-import javax.jms.DeliveryMode;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.jms.TopicConnection;
-import javax.jms.TopicPublisher;
-import javax.jms.TopicSession;
-import javax.jms.TopicSubscriber;
-import javax.naming.NamingException;
-
-import org.apache.qpid.AMQException;
-import org.apache.qpid.client.AMQQueue;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.AMQTopic;
-import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.test.utils.QpidBrokerTestCase;
-
-public class TopicWithSelectorsTransientVolumeTest extends QpidBrokerTestCase
-{   
-    private static final int NUM_MSG_PER_ITERATION = 50;//must be a multiple of 10
-    private static final int NUM_ITERATIONS = 1000;
-    
-    private static final int NUM_CONSUMERS = 50;
-    private static final int MSG_SIZE = 1024;
-    private static final byte[] BYTE_ARRAY = new byte[MSG_SIZE];
-    
-    private ArrayList<MyMessageSubscriber> _subscribers = new ArrayList<MyMessageSubscriber>();
-    private HashMap<String,Long> _queueMsgCounts = new HashMap<String,Long>();
-    
-    private final static Object _lock=new Object();
-    private boolean _producerFailed;
-    private static int _finishedCount;
-    private static int _failedCount;
-    
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-        init();
-    }
-
-    protected void tearDown() throws Exception
-    {
-        super.tearDown();
-    }
-    
-    private void init()
-    {
-        _finishedCount = 0;
-        _failedCount = 0;
-        _producerFailed = false;
-        _subscribers.clear();
-        _queueMsgCounts.clear();
-    }
-    
-    
-    private Message createMessage(Session session) throws JMSException
-    {
-        BytesMessage message = session.createBytesMessage();
-        message.writeBytes(BYTE_ARRAY);
-        
-        return message;
-    }
-    
-    /**
-     * 1 Topic with 50 subscribers using a selector, and 1 producer sending 50,000 1K messages with 90% selector success ratio.
-     */
-    public void test50SubscribersWith90PercentMatched() throws Exception
-    {
-        Topic topic = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, "test50ConsumersWith10PercentUnmatched");
-
-        System.out.println("Creating consumers");
-
-        MyMessageSubscriber sub;
-
-        for(int i=1; i <= NUM_CONSUMERS; i++)
-        {
-            sub = new MyMessageSubscriber(topic, "consumer" + i, ((9 * NUM_MSG_PER_ITERATION * NUM_ITERATIONS) / 10));
-            _subscribers.add(sub);
-        }
-        
-        System.out.println("Starting consumers");
-        for(MyMessageSubscriber s: _subscribers)
-        {
-            Thread consumer = new Thread(s);
-            consumer.start();
-        }
-
-        System.out.println("Creating producer");
-        MyMessageProducer prod = new MyMessageProducer(topic);
-        
-        long startTime = System.currentTimeMillis();
-        
-        System.out.println("Starting producer");
-        Thread producer = new Thread(prod);
-        producer.start();
-        
-        
-        // Wait for all the messageConsumers to have finished or failed
-        synchronized (_lock)
-        {
-            while (_finishedCount + _failedCount < NUM_CONSUMERS)
-            {
-                try
-                {
-                    _lock.wait();
-                }
-                catch (InterruptedException e)
-                {
-                    //ignore
-                }
-            }
-        }
-        
-        long endTime = System.currentTimeMillis();
-        System.out.println("Elapsed time for messaging: " + (endTime-startTime) + "ms");
-        
-        assertFalse("Producer failed to send all messages", _producerFailed);
-        
-        //check if all messages received by consumers, or if there were failures
-        if (_finishedCount != NUM_CONSUMERS)
-        {
-            fail(_failedCount + " consumers did not recieve all their expected messages");
-        }
-        
-        //check if all queue depths were 0
-        for(String consumer: _queueMsgCounts.keySet())
-        {
-            long depth = _queueMsgCounts.get(consumer);
-            assertEquals(consumer + " subscription queue msg count was not 0", 0, depth);
-        }
-
-    }
-    
-    private class MyMessageProducer implements Runnable
-    {
-        private TopicConnection _connection;
-        private TopicSession _session;
-        private TopicPublisher _messagePublisher;
-
-        public MyMessageProducer(Topic topic) throws JMSException, NamingException
-        {
-            _connection = (TopicConnection) getConnection();
-            _session = (TopicSession) _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-            _messagePublisher = _session.createPublisher(topic);
-        }
-
-        public void run()
-        {
-            try
-            {
-                for(int iter = 0; iter < NUM_ITERATIONS; iter++)
-                {
-                    int i = 0;
-                    
-                    //send 90% matching messages
-                    for (; i < (9 * NUM_MSG_PER_ITERATION)/10; i++)
-                    {
-                        Message message = createMessage(_session);
-                        message.setStringProperty("testprop", "true");
-
-                        _messagePublisher.publish(message, DeliveryMode.NON_PERSISTENT, 
-                                Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
-
-                        Thread.yield();
-                    }
-
-                    //send remaining 10% non-matching messages
-                    for (; i < NUM_MSG_PER_ITERATION; i++)
-                    {
-                        Message message = _session.createMessage();
-                        message.setStringProperty("testprop", "false");
-
-                        _messagePublisher.publish(message, DeliveryMode.NON_PERSISTENT, 
-                                Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE);
-
-                        Thread.yield();
-                    }
-                }
-
-            }
-            catch (Exception exp)
-            {
-                System.out.println("producer: caught an exception, probably exiting before all messages sent");
-                exp.printStackTrace();
-                synchronized (_lock)
-                {
-                    _producerFailed=true;
-                    _lock.notifyAll();
-                }
-            }
-        }
-    }
-    
-    
-    private class MyMessageSubscriber implements Runnable
-    {
-        /* The topic this subscriber is subscribing  to */
-        private Topic _topic;
-        private String _consumerName;
-        private int _outstandingMsgCount;
-        private TopicConnection _connection;
-        private TopicSession _session;
-        private TopicSubscriber _durSub;
-
-        public MyMessageSubscriber(Topic topic, String consumerName, int messageCount) throws JMSException, NamingException
-        {
-            _outstandingMsgCount = messageCount;
-            _topic=topic;
-            _consumerName = consumerName;
-            _connection = (TopicConnection) getConnection();
-            _session = (TopicSession) _connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-            _durSub = _session.createDurableSubscriber(_topic, _consumerName,"testprop='true'", false);
-            _connection.start();
-        }
-
-        public void run()
-        {
-  
-                boolean failed = false;
-                do
-                {
-                    Message m = null;
-                    try
-                    {
-                        m = _durSub.receive(10000);
-                    }
-                    catch (JMSException exp)
-                    {
-                        System.out.println(_consumerName + ": caught an exception handling a received message");
-                        exp.printStackTrace();
-
-                        failed = true;
-                        break;
-                    }
-
-                    Thread.yield();
-
-                    _outstandingMsgCount--;
-                    
-                    if(_outstandingMsgCount % 500 == 0)
-                    {
-                        System.out.println(_consumerName + ": outstanding message count: " + _outstandingMsgCount);
-                    }
-                    
-                    if(m == null)
-                    {
-                        if(_outstandingMsgCount != 0)
-                        {
-                            failed = true;
-                        }
-                        break;
-                    }
-                }
-                while(_outstandingMsgCount > 0);
-                
-                System.out.println(_consumerName + ": outstanding message count: " + _outstandingMsgCount);
-                
-                try
-                {
-                    AMQQueue subcriptionQueue = new AMQQueue(ExchangeDefaults.TOPIC_EXCHANGE_NAME,"clientid" + ":" + _consumerName);
-
-                    ((AMQSession)_session).sync();
-                    Long depth = ((AMQSession)_session).getQueueDepth(subcriptionQueue);
-                    _queueMsgCounts.put(_consumerName, depth);
-                    
-                    System.out.println(_consumerName + ": completion queue msg count: " + depth);
-                }
-                catch (AMQException exp)
-                {
-                    System.out.println(_consumerName + ": caught an exception determining completion queue depth");
-                    exp.printStackTrace();
-                }
-                finally
-                {
-                    try
-                    {
-                        _session.unsubscribe(_consumerName);
-                    }
-                    catch (JMSException e)
-                    {
-                        System.out.println(_consumerName + ": caught an exception whilst unsubscribing");
-                        e.printStackTrace();
-                    }
-                }
-
-                synchronized (_lock)
-                {
-                    if (_outstandingMsgCount == 0 && !failed)
-                    {
-                        _finishedCount++;
-                        System.out.println(_consumerName + ": finished");
-                    }
-                    else
-                    {
-                        _failedCount++;
-                        System.out.println(_consumerName + ": failed");
-                    }
-                    _lock.notifyAll();
-                }
-
-        }
-    }
-
-    //helper method to allow easily running against an external standalone broker
-//    public static void main(String[] args) throws Exception
-//    {
-//        System.setProperty("broker.config", "/dev/null");
-//        System.setProperty("broker", "external");
-//        System.setProperty("java.naming.factory.initial", "org.apache.qpid.jndi.PropertiesFileInitialContextFactory");
-//        System.setProperty("java.naming.provider.url", "test-profiles/test-provider.properties");
-//        
-//        TopicWithSelectorsTransientVolumeTest test = new TopicWithSelectorsTransientVolumeTest();
-//        test.init();
-//        test.test50SubscribersWith90PercentMatched();
-//        test.tearDown();
-//    }
-}
diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/topicselectors.properties b/qpid/java/perftests/src/main/java/org/apache/qpid/topic/topicselectors.properties
deleted file mode 100644
index 1f572af58a..0000000000
--- a/qpid/java/perftests/src/main/java/org/apache/qpid/topic/topicselectors.properties
+++ /dev/null
@@ -1,24 +0,0 @@
-#
-# 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.
-#
-java.naming.factory.initial = org.apache.qpid.jndi.PropertiesFileInitialContextFactory
-
-
-# register some connection factories
-# connectionfactory.[jndiname] = [ConnectionURL]
-connectionfactory.default = amqp://guest:guest@clientid/test?brokerlist='tcp://localhost:5672'
\ No newline at end of file