Merging bug 22331 into default

Author: Matthew Sackman

src/com/rabbitmq/client/Channel.java

@@ -77,35 +77,35 @@ public interface Channel extends ShutdownNotifier {
     /**
      * Close this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
      * and message 'OK'.
-     * 
+     *
      * @throws java.io.IOException if an error is encountered
      */
     void close() throws IOException;
-    
+
     /**
      * Close this channel.
-     * 
+     *
      * @param closeCode the close code (See under "Reply Codes" in the AMQP specification)
      * @param closeMessage a message indicating the reason for closing the connection
      * @throws java.io.IOException if an error is encountered
      */
     void close(int closeCode, String closeMessage) throws IOException;
-    
+
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
      * and message 'OK'.
-     * 
+     *
      * Forces the channel to close and waits for the close operation to complete.
      * Any encountered exceptions in the close operation are silently discarded.
      */
     void abort() throws IOException;
-    
+
     /**
      * Abort this channel.
-     * 
+     *
      * Forces the channel to close and waits for the close operation to complete.
      * Any encountered exceptions in the close operation are silently discarded.
-     */    
+     */
     void abort(int closeCode, String closeMessage) throws IOException;
 
     /**
@@ -254,7 +254,7 @@ Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean passive
      * @throws java.io.IOException if an error is encountered
      */
     Queue.DeclareOk queueDeclare(String queue) throws IOException;
-    
+
     /**
      * Actively declare a non-exclusive, non-autodelete queue
      * The name of the new queue is held in the "queue" field of the {@link com.rabbitmq.client.AMQP.Queue.DeclareOk} result.
@@ -329,7 +329,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean passive, boolean durable, boo
      * @throws java.io.IOException if an error is encountered
      */
     Queue.BindOk queueBind(String queue, String exchange, String routingKey, Map<String, Object> arguments) throws IOException;
-    
+
     /**
      * Unbinds a queue from an exchange, with no extra arguments.
      * @see com.rabbitmq.client.AMQP.Queue.Unbind
@@ -370,7 +370,7 @@ Queue.DeclareOk queueDeclare(String queue, boolean passive, boolean durable, boo
      * @see com.rabbitmq.client.AMQP.Queue.Purge
      * @see com.rabbitmq.client.AMQP.Queue.PurgeOk
      * @param queue the name of the queue
-     * @param nowait whether to await completion of the purge 
+     * @param nowait whether to await completion of the purge
      * @return a purge-confirm method if the purge was executed succesfully
      * @throws java.io.IOException if an error is encountered
      */

src/com/rabbitmq/tools/Tracer.java

@@ -31,13 +31,11 @@
 
 package com.rabbitmq.tools;
 
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.IOException;
+import java.io.*;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.util.HashMap;
+import java.util.concurrent.*;
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.impl.AMQCommand;
@@ -46,6 +44,7 @@
 import com.rabbitmq.client.impl.Frame;
 import com.rabbitmq.utility.BlockingCell;
 
+
 /**
  * AMQP Protocol Analyzer program. Listens on a configurable port and when a
  * connection arrives, makes an outbound connection to a configurable host and
@@ -64,6 +63,64 @@ public class Tracer implements Runnable {
     public static final boolean SUPPRESS_COMMAND_BODIES =
         Boolean.parseBoolean(System.getProperty("com.rabbitmq.tools.Tracer.SUPPRESS_COMMAND_BODIES"));
 
+    public static final boolean SILENT_MODE =
+        Boolean.parseBoolean(System.getProperty("com.rabbitmq.tools.Tracer.SILENT_MODE"));
+
+    final static int LOG_QUEUE_SIZE = 1024 * 1024;
+    final static int BUFFER_SIZE = 10 * 1024 * 1024;
+    final static int MAX_TIME_BETWEEN_FLUSHES = 1000;
+    final static Object FLUSH = new Object();
+
+    private static class AsyncLogger extends Thread{
+        final PrintStream ps;
+        final BlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(LOG_QUEUE_SIZE, true);
+        AsyncLogger(PrintStream ps){
+            this.ps = new PrintStream(new BufferedOutputStream(ps, BUFFER_SIZE), false);
+            start();
+
+            new Thread(){
+                @Override public void run(){
+                    while(true){
+                        try {
+                            Thread.sleep(MAX_TIME_BETWEEN_FLUSHES);
+                            queue.add(FLUSH);
+                        } catch(InterruptedException e) { }
+                    }
+
+                }
+            }.start();
+        }
+
+        void printMessage(Object message){
+            if(message instanceof Throwable){
+                ((Throwable)message).printStackTrace(ps);
+            } else if (message instanceof String){
+                ps.println(message);
+            } else {
+                throw new RuntimeException("Unrecognised object " + message);
+            }
+        }
+
+        @Override public void run(){
+            try {
+                while(true){
+                    Object message = queue.take();
+                    if(message == FLUSH) ps.flush();
+                    else printMessage(message);
+                }
+            } catch (InterruptedException interrupt){
+            }
+        }
+
+        void log(Object message){
+            try {
+              queue.put(message);
+            } catch(InterruptedException ex){
+              throw new RuntimeException(ex);
+            }
+        }
+    }
+
     public static void main(String[] args) {
         int listenPort = args.length > 0 ? Integer.parseInt(args[0]) : 5673;
         String connectHost = args.length > 1 ? args[1] : "localhost";
@@ -85,9 +142,10 @@ public static void main(String[] args) {
         try {
             ServerSocket server = new ServerSocket(listenPort);
             int counter = 0;
+            AsyncLogger logger = new AsyncLogger(System.out);
             while (true) {
                 Socket conn = server.accept();
-                new Tracer(conn, counter++, connectHost, connectPort);
+                new Tracer(conn, counter++, connectHost, connectPort, logger);
             }
         } catch (IOException ioe) {
             ioe.printStackTrace();
@@ -109,7 +167,9 @@ public static void main(String[] args) {
 
     public DataOutputStream oos;
 
-    public Tracer(Socket sock, int id, String host, int port) throws IOException {
+    public AsyncLogger logger;
+
+    public Tracer(Socket sock, int id, String host, int port, AsyncLogger logger) throws IOException {
         this.inSock = sock;
         this.outSock = new Socket(host, port);
         this.id = id;
@@ -118,6 +178,7 @@ public Tracer(Socket sock, int id, String host, int port) throws IOException {
         this.ios = new DataOutputStream(inSock.getOutputStream());
         this.ois = new DataInputStream(outSock.getInputStream());
         this.oos = new DataOutputStream(outSock.getOutputStream());
+        this.logger = logger;
 
         new Thread(this).start();
     }
@@ -135,18 +196,18 @@ public void run() {
             new Thread(outHandler).start();
             Object result = w.uninterruptibleGet();
             if (result instanceof Exception) {
-                ((Exception) result).printStackTrace();
+                logger.log(result);
             }
         } catch (EOFException eofe) {
-            eofe.printStackTrace();
+            logger.log(eofe);
         } catch (IOException ioe) {
-            ioe.printStackTrace();
+            logger.log(ioe);
         } finally {
             try {
                 inSock.close();
                 outSock.close();
             } catch (IOException ioe2) {
-                ioe2.printStackTrace();
+                logger.log(ioe2);
             }
         }
     }
@@ -174,7 +235,7 @@ public Frame readFrame() throws IOException {
         }
 
         public void report(int channel, Object object) {
-            System.out.println("" + System.currentTimeMillis() + ": conn#" + id + " ch#" + channel + (inBound ? " -> " : " <- ") + object);
+            logger.log("" + System.currentTimeMillis() + ": conn#" + id + " ch#" + channel + (inBound ? " -> " : " <- ") + object);
         }
 
         public void reportFrame(Frame f)
@@ -202,7 +263,13 @@ public void reportFrame(Frame f)
 
         public void doFrame() throws IOException {
             Frame f = readFrame();
+
             if (f != null) {
+
+                if(SILENT_MODE){
+                  f.writeTo(o);
+                  return;
+                }
                 if (f.type == AMQP.FRAME_HEARTBEAT) {
                     if ((inBound && !WITHHOLD_INBOUND_HEARTBEATS) ||
                         (!inBound && !WITHHOLD_OUTBOUND_HEARTBEATS))

src/com/rabbitmq/utility/BlockingValueOrException.java

@@ -46,7 +46,7 @@ public void setException(E e) {
     public V uninterruptibleGetValue() throws E {
         return uninterruptibleGet().getValue();
     }
-    
+
     public V uninterruptibleGetValue(int timeout) throws E, TimeoutException {
     	return uninterruptibleGet(timeout).getValue();
     }

src/com/rabbitmq/utility/Utility.java

@@ -76,7 +76,7 @@ public static <T extends Throwable & SensibleClone<T>> T fixStackTrace(T throwab
       throwable.fillInStackTrace();
       // We want to remove fixStackTrace from the trace.
       StackTraceElement[] existing = throwable.getStackTrace();
-      StackTraceElement[] newTrace = new StackTraceElement[existing.length - 1];      
+      StackTraceElement[] newTrace = new StackTraceElement[existing.length - 1];
       System.arraycopy(existing, 1, newTrace, 0, newTrace.length);
       throwable.setStackTrace(newTrace);
       return throwable;
@@ -92,7 +92,7 @@ public static String makeStackTrace(Throwable throwable) {
         try {
             outputStream.close();
         } catch (IOException ex) {
-            // Closing the output stream won't generate an error, and in 
+            // Closing the output stream won't generate an error, and in
             // fact does nothing - just being tidy
             ex.printStackTrace();
         }

src/com/rabbitmq/utility/ValueOrException.java

@@ -34,7 +34,7 @@ public class ValueOrException<V, E extends Throwable & SensibleClone<E>> {
     private final boolean _useValue;
     private final V _value;
     private final E _exception;
-    
+
     /**
      * dual-purpose private constructor: one will be null, and the flag tells which to use
      * @param value the value to wrap, if applicable
@@ -69,7 +69,7 @@ public static <V, E extends Throwable & SensibleClone<E>> ValueOrException<V, E>
     public static <V, E extends Throwable & SensibleClone<E>> ValueOrException<V, E> makeException(E exception) {
         return new ValueOrException<V, E>(null, exception, false);
     }
-    
+
     /** Retrieve value or throw exception
      * @return the wrapped value, if it's a value
      * @throws E the wrapped exception, if it's an exception

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

@@ -38,7 +38,8 @@
 
 
 public class ValueOrExceptionTest extends TestCase {
-    public static class InsufficientMagicException extends Exception implements SensibleClone<InsufficientMagicException> {
+    public static class InsufficientMagicException extends Exception 
+      implements SensibleClone<InsufficientMagicException> {
       public InsufficientMagicException(String message) {
         super(message);
       }
@@ -58,15 +59,19 @@ public static TestSuite suite()
 
     public void testStoresValue() throws InsufficientMagicException {
         Integer value = new Integer(3);
-        ValueOrException<Integer, InsufficientMagicException> valueOrEx = ValueOrException.<Integer, InsufficientMagicException>makeValue(value);
+
+        ValueOrException<Integer, InsufficientMagicException> valueOrEx = 
+            ValueOrException.<Integer, InsufficientMagicException>makeValue(value);
 
         Integer returnedValue = valueOrEx.getValue();
         assertTrue(returnedValue == value);
     }
 
     public void testClonesException() {
-        InsufficientMagicException exception = new InsufficientMagicException("dummy message");
-        ValueOrException<Integer, InsufficientMagicException> valueOrEx = ValueOrException.<Integer, InsufficientMagicException>makeException(exception);
+        InsufficientMagicException exception = 
+            new InsufficientMagicException("dummy message");
+        ValueOrException<Integer, InsufficientMagicException> valueOrEx 
+            = ValueOrException.makeException(exception);
 
         try {
             valueOrEx.getValue();

test/src/com/rabbitmq/client/test/performance/ScalabilityTest.java

@@ -144,7 +144,7 @@ public void print(final int base, final String prefix)
             s.close();
             s = open(prefix, "deletion");
             print(s, base, deletionTimes);
-            s.close(); 
+            s.close();
             s = open(prefix, "routing");
             print(s, base, transpose(routingTimes));
             s.close();