stable to default

Author: Simon MacMullen

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/impl/AMQConnection.java

@@ -21,13 +21,16 @@
 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.BlockedListener;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Channel;
@@ -81,6 +84,7 @@ 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);
 
         props.put("capabilities", capabilities);
 
@@ -130,6 +134,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 */
 
@@ -597,6 +602,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;
             }
@@ -823,4 +847,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/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

@@ -244,4 +244,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();
+    }
+
+  }
+}

test/src/com/rabbitmq/client/test/server/BlockedConnection.java

@@ -0,0 +1,93 @@
+//  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.test.server;
+
+import com.rabbitmq.client.BlockedListener;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.MessageProperties;
+import com.rabbitmq.client.test.BrokerTestCase;
+import com.rabbitmq.tools.Host;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class BlockedConnection extends BrokerTestCase {
+    protected void releaseResources() throws IOException {
+        try {
+            unblock();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+    public void testBlock() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+
+        Connection connection = connection(latch);
+        block();
+        publish(connection);
+
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+    }
+
+    public void testInitialBlock() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+
+        block();
+        Connection connection = connection(latch);
+        publish(connection);
+
+        assertTrue(latch.await(10, TimeUnit.SECONDS));
+    }
+
+    private void block() throws IOException, InterruptedException {
+        Host.rabbitmqctl("set_vm_memory_high_watermark 0.000000001");
+        setResourceAlarm("disk");
+    }
+
+    private void unblock() throws IOException, InterruptedException {
+        Host.rabbitmqctl("set_vm_memory_high_watermark 0.4");
+        clearResourceAlarm("disk");
+    }
+
+    private Connection connection(final CountDownLatch latch) throws IOException {
+        ConnectionFactory factory = new ConnectionFactory();
+        Connection connection = factory.newConnection();
+        connection.addBlockedListener(new BlockedListener() {
+            public void handleBlocked(String reason) throws IOException {
+                try {
+                    unblock();
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+            }
+
+            public void handleUnblocked() throws IOException {
+                latch.countDown();
+            }
+        });
+        return connection;
+    }
+
+    private void publish(Connection connection) throws IOException {
+        Channel ch = connection.createChannel();
+        ch.basicPublish("", "", MessageProperties.BASIC, "".getBytes());
+    }
+}

test/src/com/rabbitmq/client/test/server/MemoryAlarms.java

@@ -62,15 +62,13 @@ protected void createResources() throws IOException {
 
     @Override
     protected void releaseResources() throws IOException {
-        channel.queueDelete(Q);
-    }
-
-    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);
+        try {
+            clearAllResourceAlarms();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } finally {
+            channel.queueDelete(Q);
+        }
     }
 
     public void testFlowControl() throws IOException, InterruptedException {
@@ -87,7 +85,7 @@ public void testFlowControl() throws IOException, InterruptedException {
         assertNull(c.nextDelivery(3100));
         // once the alarm has cleared the publishes should go through
         clearResourceAlarm("memory");
-        assertNotNull(c.nextDelivery());
+        assertNotNull(c.nextDelivery(3100));
         // everything should be back to normal
         channel.basicCancel(consumerTag);
         basicPublishVolatile(Q);
@@ -108,7 +106,7 @@ public void testOverlappingAlarmsFlowControl() throws IOException, InterruptedEx
         clearResourceAlarm("memory");
         assertNull(c.nextDelivery(100));
         clearResourceAlarm("disk");
-        assertNotNull(c.nextDelivery());
+        assertNotNull(c.nextDelivery(3100));
 
         channel.basicCancel(consumerTag);
         basicPublishVolatile(Q);