Merge in default

Author: Steve Powell

codegen.py

@@ -264,12 +264,12 @@ def printWritePropertiesTo(c):
                 print "            if (this.%s != null) writer.write%s(this.%s);" % (jfName, jfClass, jfName)
         print "        }"
 
-    def printAppendArgumentDebugStringTo(c):
+    def printAppendPropertyDebugStringTo(c):
         appendList = [ "%s=\")\n               .append(this.%s)\n               .append(\""
                        % (f.name, java_field_name(f.name))
                        for f in c.fields ]
         print
-        print "        public void appendArgumentDebugStringTo(StringBuilder acc) {"
+        print "        public void appendPropertyDebugStringTo(StringBuilder acc) {"
         print "            acc.append(\"(%s)\");" % (", ".join(appendList))
         print "        }"
 
@@ -386,7 +386,7 @@ def printSetter(fieldType, fieldName):
             printSetter(jType, jName)
 
         printWritePropertiesTo(c)
-        printAppendArgumentDebugStringTo(c)
+        printAppendPropertyDebugStringTo(c)
         printPropertiesBuilderClass(c)
 
         print "    }"

src/com/rabbitmq/client/Channel.java

@@ -638,7 +638,7 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * Cancel a consumer. Calls the consumer's {@link Consumer#handleCancelOk}
      * method.
      * @param consumerTag a client- or server-generated consumer tag to establish context
-     * @throws java.io.IOException if an error is encountered
+     * @throws IOException if an error is encountered, or if the consumerTag is unknown
      * @see com.rabbitmq.client.AMQP.Basic.Cancel
      * @see com.rabbitmq.client.AMQP.Basic.CancelOk
      */
@@ -649,12 +649,12 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      * basic.recover is asynchronous; in 0-9-1 it is synchronous, and
      * the new, deprecated method basic.recover_async is asynchronous.
      * <p/>
-     * Equivalent to calling <code>basicRecover(true)</code>, messages 
-     * will be requeued and possibly delivered to a different consumer. 
+     * Equivalent to calling <code>basicRecover(true)</code>, messages
+     * will be requeued and possibly delivered to a different consumer.
      * @see #basicRecover(boolean)
      */
      Basic.RecoverOk basicRecover() throws IOException;
-  
+
     /**
      * Ask the broker to resend unacknowledged messages.  In 0-8
      * basic.recover is asynchronous; in 0-9-1 it is synchronous, and

src/com/rabbitmq/client/ConnectionFactory.java

@@ -21,7 +21,6 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import java.net.InetSocketAddress;
 import java.net.Socket;
@@ -43,39 +42,40 @@
  */
 
 public class ConnectionFactory implements Cloneable {
-    
+
     /** Default Executor threads */
-    private static final int    DEFAULT_NUM_CONSUMER_THREADS = 5;
+    @Deprecated
+    public static final int    DEFAULT_NUM_CONSUMER_THREADS = 5;
     /** Default user name */
-    private static final String DEFAULT_USER = "guest";
+    public static final String DEFAULT_USER = "guest";
     /** Default password */
-    private static final String DEFAULT_PASS = "guest";
+    public static final String DEFAULT_PASS = "guest";
     /** Default virtual host */
-    private static final String DEFAULT_VHOST = "/";
+    public static final String DEFAULT_VHOST = "/";
     /** Default maximum channel number;
      *  zero for unlimited */
-    private static final int    DEFAULT_CHANNEL_MAX = 0;
+    public static final int    DEFAULT_CHANNEL_MAX = 0;
     /** Default maximum frame size;
      *  zero means no limit */
-    private static final int    DEFAULT_FRAME_MAX = 0;
+    public static final int    DEFAULT_FRAME_MAX = 0;
     /** Default heart-beat interval;
      *  zero means no heart-beats */
-    private static final int    DEFAULT_HEARTBEAT = 0;
+    public static final int    DEFAULT_HEARTBEAT = 0;
     /** The default host */
-    private static final String DEFAULT_HOST = "localhost";
+    public static final String DEFAULT_HOST = "localhost";
     /** 'Use the default port' port */
-    private static final int    USE_DEFAULT_PORT = -1;
+    public static final int    USE_DEFAULT_PORT = -1;
     /** The default non-ssl port */
-    private static final int    DEFAULT_AMQP_PORT = AMQP.PROTOCOL.PORT;
+    public static final int    DEFAULT_AMQP_PORT = AMQP.PROTOCOL.PORT;
     /** The default ssl port */
-    private static final int    DEFAULT_AMQP_OVER_SSL_PORT = 5671;
+    public static final int    DEFAULT_AMQP_OVER_SSL_PORT = 5671;
     /** The default connection timeout;
      *  zero means wait indefinitely */
-    private static final int    DEFAULT_CONNECTION_TIMEOUT = 0;
+    public static final int    DEFAULT_CONNECTION_TIMEOUT = 0;
+
     /** The default SSL protocol */
     private static final String DEFAULT_SSL_PROTOCOL = "SSLv3";
 
-    private int numConsumerThreads                = DEFAULT_NUM_CONSUMER_THREADS;
     private String username                       = DEFAULT_USER;
     private String password                       = DEFAULT_PASS;
     private String virtualHost                    = DEFAULT_VHOST;
@@ -90,13 +90,15 @@ public class ConnectionFactory implements Cloneable {
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
 
     /** @return number of consumer threads in default {@link ExecutorService} */
+    @Deprecated
     public int getNumConsumerThreads() {
-        return numConsumerThreads;
+        return DEFAULT_NUM_CONSUMER_THREADS;
     }
 
     /** @param numConsumerThreads threads in created private executor service */
+    @Deprecated
     public void setNumConsumerThreads(int numConsumerThreads) {
-        this.numConsumerThreads = numConsumerThreads;
+        throw new IllegalArgumentException("setNumConsumerThreads not supported -- create explicit ExecutorService instead.");
     }
 
     /** @return the default host to use for connections */
@@ -436,10 +438,22 @@ protected FrameHandler createFrameHandler(Address addr)
 
         String hostName = addr.getHost();
         int portNumber = portOrDefault(addr.getPort());
-        Socket socket = factory.createSocket();
-        configureSocket(socket);
-        socket.connect(new InetSocketAddress(hostName, portNumber), connectionTimeout);
-        return createFrameHandler(socket);
+        Socket socket = null;
+        try {
+            socket = factory.createSocket();
+            configureSocket(socket);
+            socket.connect(new InetSocketAddress(hostName, portNumber),
+                    connectionTimeout);
+            return createFrameHandler(socket);
+        } catch (IOException ioe) {
+            quietTrySocketClose(socket);
+            throw ioe;
+        }
+    }
+
+    private static void quietTrySocketClose(Socket socket) {
+        if (socket != null)
+            try { socket.close(); } catch (Exception _) {/*ignore exceptions*/}
     }
 
     protected FrameHandler createFrameHandler(Socket sock)
@@ -464,21 +478,31 @@ protected void configureSocket(Socket socket) throws IOException{
         socket.setTcpNoDelay(true);
     }
 
+    /**
+     * Create a new broker connection
+     * @param addrs an array of known broker addresses (hostname/port pairs) to try in order
+     * @return an interface to the connection
+     * @throws IOException if it encounters a problem
+     */
+    public Connection newConnection(Address[] addrs) throws IOException {
+        return newConnection(null, addrs);
+    }
+
     /**
      * Create a new broker connection
      * @param executor thread execution service for consumers on the connection
      * @param addrs an array of known broker addresses (hostname/port pairs) to try in order
      * @return an interface to the connection
      * @throws IOException if it encounters a problem
      */
-    private Connection newConnection(ExecutorService executor, Address[] addrs)
+    public Connection newConnection(ExecutorService executor, Address[] addrs)
         throws IOException
     {
         IOException lastException = null;
         for (Address addr : addrs) {
             try {
                 FrameHandler frameHandler = createFrameHandler(addr);
-                AMQConnection conn = 
+                AMQConnection conn =
                     new AMQConnection(username,
                                       password,
                                       frameHandler,
@@ -506,7 +530,7 @@ private Connection newConnection(ExecutorService executor, Address[] addrs)
      * @throws IOException if it encounters a problem
      */
     public Connection newConnection() throws IOException {
-        return newConnection(Executors.newFixedThreadPool(this.numConsumerThreads),
+        return newConnection(null,
                              new Address[] {new Address(getHost(), getPort())}
                             );
     }

src/com/rabbitmq/client/impl/AMQChannel.java

@@ -99,9 +99,7 @@ public void handleFrame(Frame frame) throws IOException {
      * @return the wrapped exception
      */
     public static IOException wrap(ShutdownSignalException ex) {
-        IOException ioe = new IOException();
-        ioe.initCause(ex);
-        return ioe;
+        return wrap(ex, null);
     }
 
     public static IOException wrap(ShutdownSignalException ex, String message) {
@@ -153,13 +151,17 @@ public void handleCompleteInboundCommand(AMQCommand command) throws IOException
     public void enqueueRpc(RpcContinuation k)
     {
         synchronized (_channelMutex) {
+            boolean waitClearedInterruptStatus = false;
             while (_activeRpc != null) {
                 try {
                     _channelMutex.wait();
                 } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
+                    waitClearedInterruptStatus = true;
                 }
             }
+            if (waitClearedInterruptStatus) {
+                Thread.currentThread().interrupt();
+            }
             _activeRpc = k;
         }
     }

src/com/rabbitmq/client/impl/AMQCommand.java

@@ -146,7 +146,7 @@ private static StringBuilder contentBodyStringBuilder(byte[] body, boolean suppr
             if (suppressBody) {
                 return new StringBuilder().append(body.length).append(" bytes of payload");
             } else {
-                return new StringBuilder().append('\"').append(body).append('\"');
+                return new StringBuilder().append('\"').append(new String(body, "UTF-8")).append('\"');
             }
         } catch (Exception e) {
             return new StringBuilder().append('|').append(body.length).append('|');