Author: gtully
Date: Mon Nov 15 13:20:31 2010
New Revision: 1035264
URL: http://svn.apache.org/viewvc?rev=1035264&view=rev
Log:
applied patch and test from https://issues.apache.org/activemq/browse/AMQ-3029
with thanks
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/CompositeDataConstants.java
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/OpenTypeSupport.java
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/CompositeDataConstants.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/CompositeDataConstants.java?rev=1035264&r1=1035263&r2=1035264&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/CompositeDataConstants.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/CompositeDataConstants.java
Mon Nov 15 13:20:31 2010
@@ -28,6 +28,8 @@ public interface CompositeDataConstants
String BODY_PREVIEW = "BodyPreview";
String CONTENT_MAP = "ContentMap";
String MESSAGE_TEXT = "Text";
+ String MESSAGE_URL = "Url";
+
String ORIGINAL_DESTINATION = "OriginalDestination";
// User properties
Modified:
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/OpenTypeSupport.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/OpenTypeSupport.java?rev=1035264&r1=1035263&r2=1035264&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/OpenTypeSupport.java
(original)
+++
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/jmx/OpenTypeSupport.java
Mon Nov 15 13:20:31 2010
@@ -23,6 +23,8 @@ import static org.apache.activemq.broker
import static
org.apache.activemq.broker.jmx.CompositeDataConstants.JMSXGROUP_SEQ;
import static
org.apache.activemq.broker.jmx.CompositeDataConstants.MESSAGE_TEXT;
import static
org.apache.activemq.broker.jmx.CompositeDataConstants.ORIGINAL_DESTINATION;
+import static
org.apache.activemq.broker.jmx.CompositeDataConstants.MESSAGE_URL;
+
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
@@ -45,6 +47,7 @@ import javax.management.openmbean.Tabula
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.policy.SlowConsumerEntry;
import org.apache.activemq.broker.scheduler.Job;
+import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.activemq.command.ActiveMQBytesMessage;
import org.apache.activemq.command.ActiveMQMapMessage;
import org.apache.activemq.command.ActiveMQMessage;
@@ -432,6 +435,32 @@ public final class OpenTypeSupport {
}
}
+ static class ActiveMQBlobMessageOpenTypeFactory extends
MessageOpenTypeFactory {
+
+ @Override
+ protected String getTypeName() {
+ return ActiveMQBlobMessage.class.getName();
+ }
+
+ @Override
+ protected void init() throws OpenDataException {
+ super.init();
+ addItem(MESSAGE_URL, "Body Url", SimpleType.STRING);
+ }
+
+ @Override
+ public Map<String, Object> getFields(Object o) throws
OpenDataException {
+ ActiveMQBlobMessage m = (ActiveMQBlobMessage)o;
+ Map<String, Object> rc = super.getFields(o);
+ try {
+ rc.put(MESSAGE_URL, "" + m.getURL().toString());
+ } catch (JMSException e) {
+ rc.put(MESSAGE_URL, "");
+ }
+ return rc;
+ }
+ }
+
static class SlowConsumerEntryOpenTypeFactory extends
AbstractOpenTypeFactory {
@Override
protected String getTypeName() {
@@ -466,6 +495,7 @@ public final class OpenTypeSupport {
OPEN_TYPE_FACTORIES.put(ActiveMQTextMessage.class, new
TextMessageOpenTypeFactory());
OPEN_TYPE_FACTORIES.put(Job.class, new JobOpenTypeFactory());
OPEN_TYPE_FACTORIES.put(SlowConsumerEntry.class, new
SlowConsumerEntryOpenTypeFactory());
+ OPEN_TYPE_FACTORIES.put(ActiveMQBlobMessage.class, new
ActiveMQBlobMessageOpenTypeFactory());
}
private OpenTypeSupport() {
Modified:
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java?rev=1035264&r1=1035263&r2=1035264&view=diff
==============================================================================
---
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
(original)
+++
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/jmx/MBeanTest.java
Mon Nov 15 13:20:31 2010
@@ -19,6 +19,7 @@ package org.apache.activemq.broker.jmx;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URI;
+import java.net.URL;
import java.util.HashMap;
import java.util.Map;
@@ -37,12 +38,15 @@ import javax.management.openmbean.Tabula
import junit.textui.TestRunner;
import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.ActiveMQSession;
+import org.apache.activemq.BlobMessage;
import org.apache.activemq.EmbeddedBrokerTestSupport;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.region.BaseDestination;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.broker.region.policy.SharedDeadLetterStrategy;
+import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -575,6 +579,25 @@ public class MBeanTest extends EmbeddedB
Thread.sleep(1000);
}
+
+ protected void useConnectionWithBlobMessage(Connection connection) throws
Exception {
+ connection.setClientID(clientID);
+ connection.start();
+ ActiveMQSession session = (ActiveMQSession)
connection.createSession(transacted, authMode);
+ destination = createDestination();
+ MessageProducer producer = session.createProducer(destination);
+ for (int i = 0; i < MESSAGE_COUNT; i++) {
+ BlobMessage message = session.createBlobMessage(new
URL("http://foo.bar/test";));
+ message.setIntProperty("counter", i);
+ message.setJMSCorrelationID("MyCorrelationID");
+ message.setJMSReplyTo(new ActiveMQQueue("MyReplyTo"));
+ message.setJMSType("MyType");
+ message.setJMSPriority(5);
+ producer.send(message);
+ }
+ Thread.sleep(1000);
+ }
+
protected void echo(String text) {
LOG.info(text);
}
@@ -583,4 +606,34 @@ public class MBeanTest extends EmbeddedB
protected String getSecondDestinationString() {
return "test.new.destination." + getClass() + "." + getName();
}
+
+ // Test for AMQ-3029
+ public void testBrowseBlobMessages() throws Exception {
+ connection = connectionFactory.createConnection();
+ useConnectionWithBlobMessage(connection);
+
+ ObjectName queueViewMBeanName = assertRegisteredObjectName(domain +
":Type=Queue,Destination=" + getDestinationString() + ",BrokerName=localhost");
+
+ QueueViewMBean queue =
(QueueViewMBean)MBeanServerInvocationHandler.newProxyInstance(mbeanServer,
queueViewMBeanName, QueueViewMBean.class, true);
+
+ CompositeData[] compdatalist = queue.browse();
+ int initialQueueSize = compdatalist.length;
+ if (initialQueueSize == 0) {
+ fail("There is no message in the queue:");
+ }
+ else {
+ echo("Current queue size: " + initialQueueSize);
+ }
+ int messageCount = initialQueueSize;
+ String[] messageIDs = new String[messageCount];
+ for (int i = 0; i < messageCount; i++) {
+ CompositeData cdata = compdatalist[i];
+ String messageID = (String) cdata.get("JMSMessageID");
+ assertNotNull("Should have a message ID for message " + i,
messageID);
+
+ messageIDs[i] = messageID;
+ }
+
+ assertTrue("dest has some memory usage", queue.getMemoryPercentUsage()
> 0);
+ }
}
