You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2012/03/16 15:33:57 UTC
svn commit: r1301548 -
/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
Author: rgodfrey
Date: Fri Mar 16 14:33:57 2012
New Revision: 1301548
URL: http://svn.apache.org/viewvc?rev=1301548&view=rev
Log:
NO-JIRA : Further updates for Queue config and JMX on the java config branch
Modified:
qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
Modified: qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java?rev=1301548&r1=1301547&r2=1301548&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/broker-plugins/jmx/src/main/java/org/apache/qpid/server/jmx/mbeans/QueueMBean.java Fri Mar 16 14:33:57 2012
@@ -22,6 +22,7 @@
package org.apache.qpid.server.jmx.mbeans;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import javax.management.JMException;
@@ -56,6 +57,10 @@ public class QueueMBean extends AMQManag
private static final OpenType[] MSG_ATTRIBUTE_TYPES;
private static final CompositeType MSG_DATA_TYPE;
private static final TabularType MSG_LIST_DATA_TYPE;
+ private static final CompositeType MSG_CONTENT_TYPE;
+ private static final String[] VIEW_MSG_COMPOSIT_ITEM_NAMES_ARRAY = VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC.toArray(
+ new String[VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC.size()]);
+
static
{
@@ -77,6 +82,19 @@ public class QueueMBean extends AMQManag
MSG_LIST_DATA_TYPE = new TabularType("Messages", "List of messages", MSG_DATA_TYPE,
VIEW_MSGS_TABULAR_UNIQUE_INDEX.toArray(new String[VIEW_MSGS_TABULAR_UNIQUE_INDEX.size()]));
+ OpenType[] msgContentAttrs = new OpenType[] {
+ SimpleType.LONG, // For message id
+ SimpleType.STRING, // For MimeType
+ SimpleType.STRING, // For MimeType
+ new ArrayType(SimpleType.BYTE, true) // For message content
+ };
+
+
+ MSG_CONTENT_TYPE = new CompositeType("Message Content", "AMQ Message Content",
+ VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC.toArray(new String[VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC.size()]),
+ VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC.toArray(new String[VIEW_MSG_CONTENT_COMPOSITE_ITEM_NAMES_DESC.size()]),
+ msgContentAttrs);
+
}
catch (OpenDataException e)
{
@@ -320,7 +338,50 @@ public class QueueMBean extends AMQManag
public CompositeData viewMessageContent(long messageId)
throws IOException, JMException
{
- return null; // TODO
+ QueueEntry entry = getMessage(messageId);
+ if(entry == null)
+ {
+ throw new OperationsException("AMQMessage with message id = " + messageId + " is not in the " + _queue.getName());
+ }
+
+ ServerMessage serverMsg = entry.getMessage();
+ final int bodySize = (int) serverMsg.getSize();
+
+ byte[] msgContent = new byte[bodySize];
+
+ ByteBuffer buf = ByteBuffer.wrap(msgContent);
+ int position = 0;
+
+ while(position < bodySize)
+ {
+ position += serverMsg.getContent(buf, position);
+
+ }
+
+ AMQMessageHeader header = serverMsg.getMessageHeader();
+
+ String mimeType = null, encoding = null;
+ if (header != null)
+ {
+ mimeType = header.getMimeType();
+
+ encoding = header.getEncoding();
+ }
+
+
+ Object[] itemValues = { messageId, mimeType, encoding, msgContent };
+
+ return new CompositeDataSupport(MSG_CONTENT_TYPE, VIEW_MSG_COMPOSIT_ITEM_NAMES_ARRAY, itemValues);
+
+
+ }
+
+ private QueueEntry getMessage(long messageId)
+ {
+ GetMessageVisitor visitor = new GetMessageVisitor(messageId);
+ _queue.visit(visitor);
+ return visitor.getEntry();
+
}
public void deleteMessageFromTop() throws IOException, JMException
@@ -371,6 +432,32 @@ public class QueueMBean extends AMQManag
}
+ private static class GetMessageVisitor implements QueueEntryVisitor
+ {
+
+ private final long _messageNumber;
+ private QueueEntry _entry;
+
+ public GetMessageVisitor(long messageId)
+ {
+ _messageNumber = messageId;
+ }
+
+ public boolean visit(QueueEntry entry)
+ {
+ if(entry.getMessage().getMessageNumber() == _messageNumber)
+ {
+ _entry = entry;
+ return true;
+ }
+ return false;
+ }
+
+ public QueueEntry getEntry()
+ {
+ return _entry;
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org