Merged stable into default

Author: Emile Joubert

scripts/runperftest.sh

@@ -2,7 +2,7 @@
 
 run() {
     echo "=== running with '$2'"
-    sh `dirname $0`/runjava.sh com.rabbitmq.examples.MulticastMain -h $1 -z 10 -i 20 $2
+    sh `dirname $0`/runjava.sh com.rabbitmq.examples.PerfTest -h $1 -z 10 -i 20 $2
     sleep 2
 }
 

src/com/rabbitmq/client/AuthenticationFailureException.java

@@ -0,0 +1,28 @@
+//  The contents of this file are subject to the Mozilla Public License
+//  Version 1.1 (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.mozilla.org/MPL/
+//
+//  Software distributed under the License is distributed on an "AS IS"
+//  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+//  the License for the specific language governing rights and
+//  limitations under the License.
+//
+//  The Original Code is RabbitMQ.
+//
+//  The Initial Developer of the Original Code is GoPivotal, Inc.
+//  Copyright (c) 2013-2013 GoPivotal, Inc.  All rights reserved.
+//
+
+package com.rabbitmq.client;
+
+/**
+ * Thrown when the broker refuses access due to an authentication failure.
+ */
+
+public class AuthenticationFailureException extends PossibleAuthenticationFailureException
+{
+    public AuthenticationFailureException(String reason) {
+        super(reason);
+    }
+}

src/com/rabbitmq/client/BlockedListener.java

@@ -0,0 +1,29 @@
+//  The contents of this file are subject to the Mozilla Public License
+//  Version 1.1 (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.mozilla.org/MPL/
+//
+//  Software distributed under the License is distributed on an "AS IS"
+//  basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+//  the License for the specific language governing rights and
+//  limitations under the License.
+//
+//  The Original Code is RabbitMQ.
+//
+//  The Initial Developer of the Original Code is GoPivotal, Inc.
+//  Copyright (c) 2007-2013 GoPivotal, Inc.  All rights reserved.
+//
+
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+
+/**
+ * Implement this interface in order to be notified of connection block and
+ * unblock events.
+ */
+public interface BlockedListener {
+    void handleBlocked(String reason) throws IOException;
+    void handleUnblocked() throws IOException;
+}

src/com/rabbitmq/client/Connection.java

@@ -214,4 +214,24 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      * operations, use -1 for infinity
      */
     void abort(int closeCode, String closeMessage, int timeout);
+
+    /**
+     * Add a {@link BlockedListener}.
+     * @param listener the listener to add
+     */
+    void addBlockedListener(BlockedListener listener);
+
+    /**
+     * Remove a {@link BlockedListener}.
+     * @param listener the listener to remove
+     * @return <code><b>true</b></code> if the listener was found and removed,
+     * <code><b>false</b></code> otherwise
+     */
+    boolean removeBlockedListener(BlockedListener listener);
+
+    /**
+     * Remove all {@link BlockedListener}s.
+     */
+    void clearBlockedListeners();
+
 }

src/com/rabbitmq/client/PossibleAuthenticationFailureException.java

@@ -31,4 +31,9 @@ public PossibleAuthenticationFailureException(Throwable cause)
         super("Possibly caused by authentication failure");
         super.initCause(cause);
     }
+
+    public PossibleAuthenticationFailureException(String reason)
+    {
+        super(reason);
+    }
 }

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

@@ -21,13 +21,17 @@
 import java.net.InetAddress;
 import java.net.SocketException;
 import java.net.SocketTimeoutException;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeoutException;
 
 import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.AuthenticationFailureException;
+import com.rabbitmq.client.BlockedListener;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Channel;
@@ -81,6 +85,8 @@ public static final Map<String, Object> defaultClientProperties() {
         capabilities.put("exchange_exchange_bindings", true);
         capabilities.put("basic.nack", true);
         capabilities.put("consumer_cancel_notify", true);
+        capabilities.put("connection.blocked", true);
+        capabilities.put("authentication_failure_close", true);
 
         props.put("capabilities", capabilities);
 
@@ -130,6 +136,7 @@ public static final Map<String, Object> defaultClientProperties() {
     private final int requestedFrameMax;
     private final String username;
     private final String password;
+    private final Collection<BlockedListener> blockedListeners = new CopyOnWriteArrayList<BlockedListener>();
 
     /* State modified after start - all volatile */
 
@@ -274,9 +281,10 @@ public AMQConnection(String username,
      * either before, or during, protocol negotiation;
      * sub-classes {@link ProtocolVersionMismatchException} and
      * {@link PossibleAuthenticationFailureException} will be thrown in the
-     * corresponding circumstances. If an exception is thrown, connection
-     * resources allocated can all be garbage collected when the connection
-     * object is no longer referenced.
+     * corresponding circumstances. {@link AuthenticationFailureException}
+     * will be thrown if the broker closes the connection with ACCESS_REFUSED.
+     * If an exception is thrown, connection resources allocated can all be
+     * garbage collected when the connection object is no longer referenced.
      */
     public void start()
         throws IOException
@@ -352,6 +360,16 @@ public void start()
                         response = sm.handleChallenge(challenge, this.username, this.password);
                     }
                 } catch (ShutdownSignalException e) {
+                    Object shutdownReason = e.getReason();
+                    if (shutdownReason instanceof AMQCommand) {
+                        Method shutdownMethod = ((AMQCommand) shutdownReason).getMethod();
+                        if (shutdownMethod instanceof AMQP.Connection.Close) {
+                            AMQP.Connection.Close shutdownClose =  (AMQP.Connection.Close) shutdownMethod;
+                            if (shutdownClose.getReplyCode() == AMQP.ACCESS_REFUSED) {
+                                throw new AuthenticationFailureException(shutdownClose.getReplyText());
+                            }
+                        }
+                    }
                     throw new PossibleAuthenticationFailureException(e);
                 }
             } while (connTune == null);
@@ -597,6 +615,25 @@ public boolean processControlCommand(Command c) throws IOException
             if (method instanceof AMQP.Connection.Close) {
                 handleConnectionClose(c);
                 return true;
+            } else if (method instanceof AMQP.Connection.Blocked) {
+                AMQP.Connection.Blocked blocked = (AMQP.Connection.Blocked) method;
+                try {
+                    for (BlockedListener l : this.blockedListeners) {
+                        l.handleBlocked(blocked.getReason());
+                    }
+                } catch (Throwable ex) {
+                    getExceptionHandler().handleBlockedListenerException(this, ex);
+                }
+                return true;
+            } else if (method instanceof AMQP.Connection.Unblocked) {
+                try {
+                    for (BlockedListener l : this.blockedListeners) {
+                        l.handleUnblocked();
+                    }
+                } catch (Throwable ex) {
+                    getExceptionHandler().handleBlockedListenerException(this, ex);
+                }
+                return true;
             } else {
                 return false;
             }
@@ -621,7 +658,7 @@ public boolean processControlCommand(Command c) throws IOException
     }
 
     public void handleConnectionClose(Command closeCommand) {
-        ShutdownSignalException sse = shutdown(closeCommand, false, null, false);
+        ShutdownSignalException sse = shutdown(closeCommand, false, null, _inConnectionNegotiation);
         try {
             _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
         } catch (IOException _) { } // ignore
@@ -823,4 +860,16 @@ public AMQCommand transformReply(AMQCommand command) {
     private String getHostAddress() {
         return getAddress() == null ? null : getAddress().getHostAddress();
     }
+
+    public void addBlockedListener(BlockedListener listener) {
+        blockedListeners.add(listener);
+    }
+
+    public boolean removeBlockedListener(BlockedListener listener) {
+        return blockedListeners.remove(listener);
+    }
+
+    public void clearBlockedListeners() {
+        blockedListeners.clear();
+    }
 }

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

@@ -394,7 +394,7 @@ private void releaseChannel() {
                 }
                 if (callback != null) {
                     try {
-                        callback.handleCancel(consumerTag);
+                        this.dispatcher.handleCancel(callback, consumerTag);
                     } catch (Throwable ex) {
                         getConnection().getExceptionHandler().handleConsumerException(this,
                                                                                       ex,

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

@@ -100,6 +100,25 @@ public void run() {
         });
     }
 
+    public void handleCancel(final Consumer delegate, final String consumerTag) {
+        executeUnlessShuttingDown(
+        new Runnable() {
+      public void run() {
+                try {
+                    delegate.handleCancel(consumerTag);
+                } catch (Throwable ex) {
+                    connection.getExceptionHandler().handleConsumerException(
+                            channel,
+                            ex,
+                            delegate,
+                            consumerTag,
+                            "handleCancel");
+                }
+      }
+    });
+  }
+
+
     public void handleRecoverOk(final Consumer delegate, final String consumerTag) {
         executeUnlessShuttingDown(
         new Runnable() {

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

@@ -49,6 +49,10 @@ public void handleConfirmListenerException(Channel channel, Throwable exception)
         handleChannelKiller(channel, exception, "ConfirmListener.handle{N,A}ck");
     }
 
+    public void handleBlockedListenerException(Connection connection, Throwable exception) {
+        handleConnectionKiller(connection, exception, "BlockedListener");
+    }
+
     public void handleConsumerException(Channel channel, Throwable exception,
                                         Consumer consumer, String consumerTag,
                                         String methodName)
@@ -76,4 +80,22 @@ protected void handleChannelKiller(Channel channel, Throwable exception, String
             channel.getConnection().abort(AMQP.INTERNAL_ERROR, "Internal error closing channel for " + what);
         }
     }
+
+    protected void handleConnectionKiller(Connection connection, Throwable exception, String what) {
+        // TODO: log the exception
+        System.err.println("DefaultExceptionHandler: " + what + " threw an exception for connection "
+                + connection + ":");
+        exception.printStackTrace();
+        try {
+            connection.close(AMQP.REPLY_SUCCESS, "Closed due to exception from " + what);
+        } catch (AlreadyClosedException ace) {
+            // noop
+        } catch (IOException ioe) {
+            // TODO: log the failure
+            System.err.println("Failure during close of connection " + connection + " after " + exception
+                    + ":");
+            ioe.printStackTrace();
+            connection.abort(AMQP.INTERNAL_ERROR, "Internal error closing connection for " + what);
+        }
+    }
 }

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

@@ -63,6 +63,16 @@ public interface ExceptionHandler {
      */
     void handleConfirmListenerException(Channel channel, Throwable exception);
 
+    /**
+     * Perform any required exception processing for the situation
+     * when the driver thread for the connection has called a
+     * BlockedListener's method, and that method has
+     * thrown an exception.
+     * @param connection the Connection that held the BlockedListener
+     * @param exception the exception thrown by the BlockedListener
+     */
+    void handleBlockedListenerException(Connection connection, Throwable exception);
+
     /**
      * Perform any required exception processing for the situation
      * when the driver thread for the connection has called a method

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

@@ -242,6 +242,10 @@ public void handleConfirmListenerException(Channel ch, Throwable ex) {
             fail("handleConfirmListenerException: " + ex);
         }
 
+        public void handleBlockedListenerException(Connection conn, Throwable ex) {
+            fail("handleBlockedListenerException: " + ex);
+        }
+
         public void handleConsumerException(Channel ch,
                                             Throwable ex,
                                             Consumer c,

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

@@ -248,4 +248,17 @@ protected void deleteExchange(String x) throws IOException {
     protected void deleteQueue(String q) throws IOException {
         channel.queueDelete(q);
     }
+
+    protected void clearAllResourceAlarms() throws IOException, InterruptedException {
+        clearResourceAlarm("memory");
+        clearResourceAlarm("disk");
+    }
+
+    protected void setResourceAlarm(String source) throws IOException, InterruptedException {
+        Host.executeCommand("cd ../rabbitmq-test; make set-resource-alarm SOURCE=" + source);
+    }
+
+    protected void clearResourceAlarm(String source) throws IOException, InterruptedException {
+        Host.executeCommand("cd ../rabbitmq-test; make clear-resource-alarm SOURCE=" + source);
+    }
 }

test/src/com/rabbitmq/client/test/functional/BasicGet.java

@@ -0,0 +1,49 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.AlreadyClosedException;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.test.BrokerTestCase;
+
+import java.io.IOException;
+
+public class BasicGet extends BrokerTestCase {
+  public void testBasicGetWithEnqueuedMessages() throws IOException, InterruptedException {
+    assertTrue(channel.isOpen());
+    String q = channel.queueDeclare().getQueue();
+
+    basicPublishPersistent("msg".getBytes("UTF-8"), q);
+    Thread.sleep(250);
+
+    assertNotNull(channel.basicGet(q, true));
+    channel.queuePurge(q);
+    assertNull(channel.basicGet(q, true));
+    channel.queueDelete(q);
+  }
+
+  public void testBasicGetWithEmptyQueue() throws IOException, InterruptedException {
+    assertTrue(channel.isOpen());
+    String q = channel.queueDeclare().getQueue();
+
+    assertNull(channel.basicGet(q, true));
+    channel.queueDelete(q);
+  }
+
+  public void testBasicGetWithClosedChannel() throws IOException, InterruptedException {
+    assertTrue(channel.isOpen());
+    String q = channel.queueDeclare().getQueue();
+
+    channel.close();
+    assertFalse(channel.isOpen());
+    try {
+      channel.basicGet(q, true);
+      fail("expected basic.get on a closed channel to fail");
+    } catch (AlreadyClosedException e) {
+      // passed
+    } finally {
+      Channel tch = connection.createChannel();
+      tch.queueDelete(q);
+      tch.close();
+    }
+
+  }
+}