Merged bug24999 into default

Author: Emile Joubert

test/src/com/rabbitmq/examples/TracerConcurrencyTest.java

@@ -20,68 +20,80 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 
-/** 
- * Test that the tracer correctly handles multiple concurrently processing
- * channels. If it doesn't then this code will cause the tracer to break
- * and will be disconneted.
+/**
+ * Java Application. Repeatedly generate (and get) messages on multiple concurrently processing channels.
+ * <p/>
+ * This application connects to localhost port 5673, and is useful for testing {@link com.rabbitmq.tools.Tracer Tracer}.
+ * @see com.rabbitmq.tools.Tracer Tracer
  */
-public class TracerConcurrencyTest{
+public class TracerConcurrencyTest {
 
-  public String uri = "amqp://localhost:5673";
-  public int threadCount = 3;
+    private static final String uri = "amqp://localhost:5673";
+    private static final int THREADCOUNT = 3;
+    private static final String EXCHANGE = "tracer-exchange";
+    private static final String QUEUE = "tracer-queue";
+    private static final String ROUTING_KEY = "";
 
-  private final Object lock = new Object();
+    /**
+     * @param args command-line parameters -- all ignored.
+     * @throws Exception test
+     */
+    public static void main(String[] args) throws Exception {
 
-  public static void main(String[] args) throws Exception{
-    new TracerConcurrencyTest().run();
-  }
+        final Object outputSync = new Object();
 
-  static String EXCHANGE = "tracer-exchange";
-  static String QUEUE = "tracer-queue";
+        final Connection conn = createConnectionAndResources();
 
-  public void run(){
+        for (int i = 0; i < THREADCOUNT; i++) {
+            new TestThread(conn, outputSync).start();
+        }
+    }
 
-    final Connection conn;
-    try {
-      conn = new ConnectionFactory()
-          {{setUri(uri);}}.newConnection();
-      Channel setup = conn.createChannel();
+    private static class TestThread extends Thread {
+        private final Connection conn;
+        private final Object outputSync;
 
-      setup.exchangeDeclare(EXCHANGE, "direct");
-      setup.queueDeclare(QUEUE, false, false, false, null);
-      setup.queueBind(QUEUE,EXCHANGE, "");
-      
-      setup.close();
-    } catch(Exception e){
-      e.printStackTrace();
-      System.exit(1);
-      throw null; // placate the compiler
-    }
+        private TestThread(Connection conn, Object outputSync) {
+            this.conn = conn;
+            this.outputSync = outputSync;
+        }
 
-    for(int i = 0; i < threadCount; i++){
-      new Thread(){
-        @Override public void run(){
-          try {
-            Channel ch = conn.createChannel();
-            while(true){
-                ch.close();
-                ch = conn.createChannel(); 
-                ch.basicPublish(
-                  EXCHANGE,
-                  "", null,
-                  new byte[1024 * 1024]
-                );
-                ch.basicGet(QUEUE, true);
-            }
-          } catch(Exception e){
-            synchronized(lock){
-              e.printStackTrace();
-              System.err.println();
+        @Override
+        public void run() {
+            try {
+                while (true) {
+                    Channel ch = conn.createChannel();
+                    ch.basicPublish(EXCHANGE, ROUTING_KEY, null, new byte[1024 * 1024]);
+                    ch.basicGet(QUEUE, true);
+                    ch.close();
+                }
+            } catch (Exception e) {
+                synchronized (outputSync) {
+                    e.printStackTrace();
+                    System.err.println();
+                }
+                System.exit(1);
             }
-            System.exit(1);
-          } 
         }
-      }.start();
+
+    }
+
+    /**
+     * Create connection and declare exchange and queue for local use.
+     *
+     * @return connection
+     */
+    private static final Connection createConnectionAndResources() throws Exception {
+        ConnectionFactory cf = new ConnectionFactory();
+        cf.setUri(uri);
+        Connection conn = cf.newConnection();
+        Channel setup = conn.createChannel();
+
+        setup.exchangeDeclare(EXCHANGE, "direct");
+        setup.queueDeclare(QUEUE, false, false, false, null);
+        setup.queueBind(QUEUE, EXCHANGE, ROUTING_KEY);
+
+        setup.close();
+        return conn;
     }
-  }
 }