restart bug22412 off default

Author: Alexandru Scvortov

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/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
@@ -257,6 +275,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 +730,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

@@ -63,6 +63,22 @@ public void handleReturnListenerException(Channel channel, Throwable exception)
         }
     }
 
+    public void handleFlowListenerException(Channel channel, Throwable exception) {
+        // TODO: Convert to logging framework
+        System.err.println("FlowListener.handleFlow threw an exception for channel " +
+                           channel + ":");
+        exception.printStackTrace();
+        try {
+            ((AMQConnection) channel.getConnection()).close(AMQP.INTERNAL_ERROR,
+                                                            "Internal error in FlowListener",
+                                                            false,
+                                                            exception);
+        } catch (IOException ioe) {
+            // Man, this clearly isn't our day.
+            // Ignore the exception? TODO: Log the nested failure
+        }
+    }
+
     public void handleConsumerException(Channel channel,
                                         Throwable exception,
                                         Consumer consumer,

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,