Merged bug22412 into default

Author: Emile Joubert

src/com/rabbitmq/client/Channel.java

@@ -101,6 +101,11 @@ public interface Channel extends ShutdownNotifier {
      */
     FlowOk flow(boolean active) throws IOException;
 
+    /**
+     * Return the current Channel.Flow settings.
+     */
+    FlowOk getFlow();
+
     /**
      * Abort this channel with the {@link com.rabbitmq.client.AMQP#REPLY_SUCCESS} close code
      * and message 'OK'.
@@ -130,6 +135,18 @@ public interface Channel extends ShutdownNotifier {
      */
     void setReturnListener(ReturnListener listener);
 
+    /**
+     * Return the current {@link FlowListener}.
+     * @return an interface to the current flow listener.
+     */
+    FlowListener getFlowListener();
+
+    /**
+     * Set the current {@link FlowListener}.
+     * @param listener the listener to use, or null indicating "don't use one".
+     */
+    void setFlowListener(FlowListener listener);
+
     /**
      * Request specific "quality of service" settings.
      *

src/com/rabbitmq/client/FlowListener.java

@@ -0,0 +1,43 @@
+//   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 Developers of the Original Code are LShift Ltd,
+//   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+//   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+//   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+//   Technologies LLC, and Rabbit Technologies Ltd.
+//
+//   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
+//   Ltd. Portions created by Cohesive Financial Technologies LLC are
+//   Copyright (C) 2007-2010 Cohesive Financial Technologies
+//   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+//   (C) 2007-2010 Rabbit Technologies Ltd.
+//
+//   All Rights Reserved.
+//
+//   Contributor(s): ______________________________________.
+//
+
+package com.rabbitmq.client;
+
+import java.io.IOException;
+
+/**
+ * Implement this interface in order to be notified of Channel.Flow
+ * events.
+ */
+public interface FlowListener {
+    void handleFlow(boolean active)
+        throws IOException;
+}

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

@@ -37,6 +37,7 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.FlowListener;
 import com.rabbitmq.client.GetResponse;
 import com.rabbitmq.client.MessageProperties;
 import com.rabbitmq.client.ReturnListener;
@@ -93,6 +94,10 @@ public class ChannelN extends AMQChannel implements com.rabbitmq.client.Channel
      */
     public volatile ReturnListener returnListener = null;
 
+    /** Reference to the currently-active FlowListener, or null if there is none.
+     */
+    public volatile FlowListener flowListener = null;
+
     /**
      * Construct a new channel on the given connection with the given
      * channel number. Usually not called directly - call
@@ -130,6 +135,19 @@ public void setReturnListener(ReturnListener listener) {
         returnListener = listener;
     }
 
+    /** Returns the current FlowListener. */
+    public FlowListener getFlowListener() {
+        return flowListener;
+    }
+
+    /**
+     * Sets the current FlowListener.
+     * A null argument is interpreted to mean "do not use a flow listener".
+     */
+    public void setFlowListener(FlowListener listener) {
+        flowListener = listener;
+    }
+
     /**
      * Protected API - sends a ShutdownSignal to all active consumers.
      * @param signal an exception signalling channel shutdown
@@ -171,7 +189,8 @@ public void releaseChannelNumber() {
 
     /**
      * Protected API - Filters the inbound command stream, processing
-     * Basic.Deliver, Basic.Return and Channel.Close specially.
+     * Basic.Deliver, Basic.Return, Channel.Flow and Channel.Close
+     * specially.
      */
     @Override public boolean processAsync(Command command) throws IOException
     {
@@ -257,6 +276,14 @@ public void releaseChannelNumber() {
                     transmit(new Channel.FlowOk(channelFlow.active));
                     _channelMutex.notifyAll();
                 }
+                FlowListener l = getFlowListener();
+                if (l != null) {
+                    try {
+                        l.handleFlow(channelFlow.active);
+                    } catch (Throwable ex) {
+                        _connection.getExceptionHandler().handleFlowListenerException(this, ex);
+                    }
+                }
                 return true;
             } else {
                 return false;
@@ -704,4 +731,10 @@ public Tx.RollbackOk txRollback()
     public Channel.FlowOk flow(final boolean a) throws IOException {
         return (Channel.FlowOk) exnWrappingRpc(new Channel.Flow() {{active = a;}}).getMethod();
     }
+
+    /** Public API - {@inheritDoc} */
+    public Channel.FlowOk getFlow() {
+        return new Channel.FlowOk(!_blockContent);
+    }
+
 }

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

@@ -48,40 +48,40 @@ public void handleUnexpectedConnectionDriverException(Connection conn, Throwable
     }
 
     public void handleReturnListenerException(Channel channel, Throwable exception) {
-        // TODO: Convert to logging framework
-        System.err.println("ReturnListener.handleBasicReturn threw an exception for channel " +
-                           channel + ":");
-        exception.printStackTrace();
-        try {
-            ((AMQConnection) channel.getConnection()).close(AMQP.INTERNAL_ERROR,
-                                                            "Internal error in ReturnListener",
-                                                            false,
-                                                            exception);
-        } catch (IOException ioe) {
-            // Man, this clearly isn't our day.
-            // Ignore the exception? TODO: Log the nested failure
-        }
+        handleChannelKiller(channel, exception, "ReturnListener.handleBasicReturn");
     }
 
-    public void handleConsumerException(Channel channel,
-                                        Throwable exception,
-                                        Consumer consumer,
-                                        String consumerTag,
+    public void handleFlowListenerException(Channel channel, Throwable exception) {
+        handleChannelKiller(channel, exception, "FlowListener.handleFlow");
+    }
+
+    public void handleConsumerException(Channel channel, Throwable exception,
+                                        Consumer consumer, String consumerTag,
                                         String methodName)
+    {
+        handleChannelKiller(channel, exception, "Consumer " + consumer
+                                              + " (" + consumerTag + ")"
+                                              + " method " + methodName
+                                              + " for channel " + channel);
+    }
+
+    protected void handleChannelKiller(Channel channel,
+                                       Throwable exception,
+                                       String what)
     {
         // TODO: Convert to logging framework
-        System.err.println("Consumer " + consumer + " method " + methodName + " for channel " +
-                           channel + " threw an exception:");
+        System.err.println(what + " threw an exception for channel " +
+                           channel + ":");
         exception.printStackTrace();
         try {
             ((AMQConnection) channel.getConnection()).close(AMQP.INTERNAL_ERROR,
-                                                            "Internal error in Consumer " +
-                                                              consumerTag,
+                                                            "Internal error in " + what,
                                                             false,
                                                             exception);
         } catch (IOException ioe) {
             // Man, this clearly isn't our day.
             // Ignore the exception? TODO: Log the nested failure
         }
+
     }
 }

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

@@ -57,6 +57,16 @@ public interface ExceptionHandler {
      */
     void handleReturnListenerException(Channel channel, Throwable exception);
 
+    /**
+     * Perform any required exception processing for the situation
+     * when the driver thread for the connection has called a 
+     * FlowListener's handleFlow method, and that method has
+     * thrown an exeption.
+     * @param channel the ChannelN that held the FlowListener
+     * @param exception the exception thrown by FlowListener.handleFlow
+     */
+    void handleFlowListenerException(Channel channel, 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

@@ -200,6 +200,10 @@ public void handleReturnListenerException(Channel ch, Throwable ex) {
             fail("handleReturnListenerException: " + ex);
         }
 
+        public void handleFlowListenerException(Channel ch, Throwable ex) {
+            fail("handleFlowListenerException: " + ex);
+        }
+
         public void handleConsumerException(Channel ch,
                                             Throwable ex,
                                             Consumer c,