QPID-5970 : [Java Broker] Expose "age of oldest message on queue" to management apis

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

4 files changed, 52 insertions, 13 deletions

diff --git a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageContentServlet.java b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageContentServlet.java
index 074ca865a6..ef33a240a9 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageContentServlet.java
+++ b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageContentServlet.java
@@ -24,6 +24,7 @@ import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
+import org.apache.qpid.server.message.MessageDeletedException;
 import org.apache.qpid.server.message.MessageReference;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.model.Queue;
@@ -121,15 +122,27 @@ public class MessageContentServlet extends AbstractServlet
             {
                 if(_messageNumber == message.getMessageNumber())
                 {
-                    MessageReference reference = message.newReference();
-
-                    _mimeType = message.getMessageHeader().getMimeType();
-                    _size = message.getSize();
-                    _content = new byte[(int)_size];
-                    _found = true;
-                    message.getContent(ByteBuffer.wrap(_content),0);
-                    reference.release();
-                    return true;
+                    try
+                    {
+                        MessageReference reference = message.newReference();
+                        try
+                        {
+                            _mimeType = message.getMessageHeader().getMimeType();
+                            _size = message.getSize();
+                            _content = new byte[(int) _size];
+                            _found = true;
+                            message.getContent(ByteBuffer.wrap(_content), 0);
+                        }
+                        finally
+                        {
+                            reference.release();
+                        }
+                        return true;
+                    }
+                    catch (MessageDeletedException e)
+                    {
+                        // ignore - the message was deleted as we tried too look at it, treat as if no message found
+                    }
                 }
 
             }
diff --git a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
index 531ea1e3c1..9866207234 100644
--- a/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
+++ b/qpid/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java
@@ -35,6 +35,7 @@ import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.map.SerializationConfig;
 import org.apache.qpid.server.consumer.ConsumerImpl;
 import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.MessageDeletedException;
 import org.apache.qpid.server.message.MessageReference;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.model.Queue;
@@ -328,10 +329,23 @@ public class MessageServlet extends AbstractServlet
             {
                 if(_messageNumber == message.getMessageNumber())
                 {
-                    MessageReference reference = message.newReference();
-                    _messageObject = convertToObject(entry, true);
-                    reference.release();
-                    return true;
+                    try
+                    {
+                        MessageReference reference = message.newReference();
+                        try
+                        {
+                            _messageObject = convertToObject(entry, true);
+                        }
+                        finally
+                        {
+                            reference.release();
+                        }
+                        return true;
+                    }
+                    catch (MessageDeletedException e)
+                    {
+                        // ignore - the message has been deleted before we got a chance to look at it
+                    }
                 }
             }
             return false;
diff --git a/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java b/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
index a1fec6f69e..ca092fe6f8 100644
--- a/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
+++ b/qpid/java/broker-plugins/management-jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
@@ -700,4 +700,10 @@ public class QueueMBean extends AMQManagedObject implements ManagedQueue, QueueN
     {
         return _queue.isMessageGroupSharedGroups();
     }
+
+    @Override
+    public Long getOldestMessageAge()
+    {
+        return _queue.getOldestMessageAge();
+    }
 }
diff --git a/qpid/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java b/qpid/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java
index 6f31dc6d04..671920c33d 100644
--- a/qpid/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java
+++ b/qpid/java/broker-plugins/management-jmx/src/test/java/org/apache/qpid/server/jmx/mbeans/QueueMBeanTest.java
@@ -118,6 +118,12 @@ public class QueueMBeanTest extends QpidTestCase
         assertStatistic("consumerCount", 3);
     }
 
+    public void testOldestMessageAge() throws Exception
+    {
+        when(_mockQueue.getOldestMessageAge()).thenReturn(3l);
+        assertStatistic("oldestMessageAge", 3l);
+    }
+
     /**********  Simple Attributes **********/
 
     public void testGetQueueDescription() throws Exception