QPID-2177: expose Capacity, FlowResumeCapacity, and FlowOverfull as attributes of the Queue MBeans

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@888248 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 105 insertions, 4 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index c9d20c53e4..a459c64946 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
@@ -186,7 +186,7 @@ public interface AMQQueue extends Managable, Comparable<AMQQueue>, ExchangeRefer
 
     void setFlowResumeCapacity(long flowResumeCapacity);
 
-
+    boolean isOverfull();
 
     void deleteMessageFromTop();
 
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
index eafc73dc97..021128d2fc 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueueMBean.java
@@ -241,6 +241,41 @@ public class AMQQueueMBean extends AMQManagedObject implements ManagedQueue, Que
         return _queue.getQueueDepth();
     }
 
+    public Long getCapacity()
+    {
+        return _queue.getCapacity();
+    }
+
+    public void setCapacity(Long capacity) throws IllegalArgumentException
+    {
+        if( _queue.getFlowResumeCapacity() > capacity )
+        {
+            throw new IllegalArgumentException("Capacity must not be less than FlowResumeCapacity");
+        }
+        
+    	_queue.setCapacity(capacity);
+    }
+
+    public Long getFlowResumeCapacity()
+    {
+        return _queue.getFlowResumeCapacity();
+    }
+
+    public void setFlowResumeCapacity(Long flowResumeCapacity) throws IllegalArgumentException
+    {
+        if( _queue.getCapacity() < flowResumeCapacity )
+        {
+            throw new IllegalArgumentException("FlowResumeCapacity must not exceed Capacity");
+        }
+        
+        _queue.setFlowResumeCapacity(flowResumeCapacity);
+    }
+    
+    public boolean isFlowOverfull()
+    {
+        return _queue.isOverfull();
+    }
+
     /**
      * Checks if there is any notification to be send to the listeners
      */
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index d7d0414936..b4cebda09a 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -1799,8 +1799,14 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
     public void setFlowResumeCapacity(long flowResumeCapacity)
     {
         _flowResumeCapacity = flowResumeCapacity;
+        
+        checkCapacity();
     }
 
+    public boolean isOverfull()
+    {
+        return _overfull.get();
+    }
 
     public Set<NotificationCheck> getNotificationChecks()
     {
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
index 7c1f728664..910c7d42ed 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
@@ -426,4 +426,9 @@ public class MockAMQQueue implements AMQQueue
     {
         return _name.toString();
     }
+
+    public boolean isOverfull()
+    {
+        return false;
+    }
 }
diff --git a/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java b/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
index ff4edb4ddd..7838400cd1 100644
--- a/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
+++ b/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ManagedQueue.java
@@ -68,6 +68,9 @@ public interface ManagedQueue
     String ATTR_MSG_COUNT = "MessageCount";
     String ATTR_QUEUE_DEPTH = "QueueDepth";
     String ATTR_RCVD_MSG_COUNT = "ReceivedMessageCount";
+    String ATTR_CAPACITY = "Capacity";
+    String ATTR_FLOW_OVERFULL = "FlowOverfull";
+    String ATTR_FLOW_RESUME_CAPACITY = "FlowResumeCapacity";
     
     //All attribute names constant
     String[] QUEUE_ATTRIBUTES = new String[]{
@@ -83,7 +86,10 @@ public interface ManagedQueue
             ATTR_ACTIVE_CONSUMER_COUNT,
             ATTR_MSG_COUNT,
             ATTR_QUEUE_DEPTH,
-            ATTR_RCVD_MSG_COUNT
+            ATTR_RCVD_MSG_COUNT,
+            ATTR_CAPACITY,
+            ATTR_FLOW_OVERFULL,
+            ATTR_FLOW_RESUME_CAPACITY
     };
     
     /**
@@ -230,7 +236,56 @@ public interface ManagedQueue
      */
     @MBeanAttribute(name="MaximumQueueDepth", description="The threshold high value(Bytes) for Queue Depth")
     void setMaximumQueueDepth(Long value) throws IOException;
+    
+    
+    /**
+     * Returns the current flow control Capacity of the queue in bytes.
+     * 
+     * @since Qpid JMX API 1.6
+     * @return Capacity at which flow control is enforced
+     * @throws IOException
+     */
+    Long getCapacity() throws IOException;
 
+    /**
+     * Sets the Capacity in bytes above which flow is blocked.
+     * 
+     * @since Qpid JMX API 1.6
+     * @param value the capacity in bytes
+     * @throws IOException
+     * @throws IllegalArgumentException If the given value is less than the queue FloeResumeCapacity
+     */
+    @MBeanAttribute(name="Capacity", description="The flow control Capacity (Bytes) of the queue")
+    void setCapacity(Long value) throws IOException, IllegalArgumentException;
+    
+    /**
+     * Returns the current flow control FlowResumeCapacity of the queue in bytes.
+     * 
+     * @since Qpid JMX API 1.6
+     * @return Capacity below which flow resumes in bytes
+     * @throws IOException
+     */
+    Long getFlowResumeCapacity() throws IOException;
+
+    /**
+     * Sets the FlowResumeCapacity in bytes below which flow resumes.
+     * 
+     * @since Qpid JMX API 1.6
+     * @param value of the resume capacity in bytes
+     * @throws IOException
+     * @throws IllegalArgumentException If the given value exceeds the queue Capacity
+     */
+    @MBeanAttribute(name="FlowResumeCapacity", description="The flow resume Capacity (Bytes) of the queue")
+    void setFlowResumeCapacity(Long value) throws IOException, IllegalArgumentException;
+
+    /**
+     * Indicates whether the Queue is currently considered overfull by the FlowControl system
+     * 
+     * @since Qpid JMX API 1.6
+     * @throws IOException
+     */
+    @MBeanAttribute(name="FlowOverfull", description="true if the queue is considered overfull by the Flow Control system")
+    boolean isFlowOverfull() throws IOException;
 
     //********** Operations *****************//
 
diff --git a/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java b/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java
index 68ecd670c2..9b75945d53 100644
--- a/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java
+++ b/qpid/java/management/common/src/main/java/org/apache/qpid/management/common/mbeans/ServerInformation.java
@@ -43,7 +43,7 @@ public interface ServerInformation
      *  Qpid JMX API 1.1 can be assumed.
      */
     int QPID_JMX_API_MAJOR_VERSION = 1;
-    int QPID_JMX_API_MINOR_VERSION = 5;
+    int QPID_JMX_API_MINOR_VERSION = 6;
     
     
     /**
diff --git a/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/ApplicationRegistry.java b/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/ApplicationRegistry.java
index c22cfc4f45..f3fc135efb 100644
--- a/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/ApplicationRegistry.java
+++ b/qpid/java/management/eclipse-plugin/src/main/java/org/apache/qpid/management/ui/ApplicationRegistry.java
@@ -47,7 +47,7 @@ public abstract class ApplicationRegistry
 
     //max supported broker management interface supported by this release of the management console
     public static final int SUPPORTED_QPID_JMX_API_MAJOR_VERSION = 1;
-    public static final int SUPPORTED_QPID_JMX_API_MINOR_VERSION = 5;
+    public static final int SUPPORTED_QPID_JMX_API_MINOR_VERSION = 6;
     
     public static final String DATA_DIR = System.getProperty("user.home") + File.separator + ".qpidmc";