[hornetq-commits] JBoss hornetq SVN: r9711 - in branches/hornetq-416: src/main/org/hornetq/core/server and 4 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Wed Sep 22 05:07:32 EDT 2010
Author: jmesnil
Date: 2010-09-22 05:07:31 -0400 (Wed, 22 Sep 2010)
New Revision: 9711
Added:
branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSConsumerInfo.java
Modified:
branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSServerControl.java
branches/hornetq-416/src/main/org/hornetq/core/server/Queue.java
branches/hornetq-416/src/main/org/hornetq/core/server/ServerConsumer.java
branches/hornetq-416/src/main/org/hornetq/core/server/ServerSession.java
branches/hornetq-416/src/main/org/hornetq/core/server/impl/QueueImpl.java
branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java
branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
branches/hornetq-416/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java
branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java
branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java
branches/hornetq-416/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java
Log:
https://jira.jboss.org/browse/HORNETQ-416
* add JMSConsumerInfo and JMSServerControl.listConsumersAsJSON()
* add ServerConsumer.getCreationTime() and isBrowseOnly() methods
Added: branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSConsumerInfo.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSConsumerInfo.java (rev 0)
+++ branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSConsumerInfo.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.api.jms.management;
+
+import org.hornetq.utils.json.JSONArray;
+import org.hornetq.utils.json.JSONObject;
+
+/**
+ * Helper class to create Java Objects from the
+ * JSON serialization returned by {@link JMSServerControl#listConsumersAsJSON(String)} and related methods.
+ *
+ * @author <a href="mailto:jmesnil at redhat.com">Jeff Mesnil</a>
+ */
+public class JMSConsumerInfo
+{
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ private final String consumerID;
+
+ private final String connectionID;
+
+ private final String destinationName;
+
+ private final String destinationType;
+
+ private final boolean browseOnly;
+
+ private final long creationTime;
+
+ private final boolean durable;
+
+ private final String filter;
+
+ // Static --------------------------------------------------------
+
+ /**
+ * Returns an array of SubscriptionInfo corresponding to the JSON serialization returned
+ * by {@link TopicControl#listAllSubscriptionsAsJSON()} and related methods.
+ */
+ public static JMSConsumerInfo[] from(final String jsonString) throws Exception
+ {
+ JSONArray array = new JSONArray(jsonString);
+ JMSConsumerInfo[] infos = new JMSConsumerInfo[array.length()];
+ for (int i = 0; i < array.length(); i++)
+ {
+ JSONObject sub = array.getJSONObject(i);
+ JMSConsumerInfo info = new JMSConsumerInfo(sub.getString("consumerID"),
+ sub.getString("connectionID"),
+ sub.getString("destinationName"),
+ sub.getString("destinationType"),
+ sub.getBoolean("browseOnly"),
+ sub.getLong("creationTime"),
+ sub.getBoolean("durable"),
+ sub.optString("filter", null));
+ infos[i] = info;
+ }
+
+ return infos;
+ }
+
+ // Constructors --------------------------------------------------
+
+ private JMSConsumerInfo(final String consumerID,
+ final String connectionID,
+ final String destinationName,
+ final String destinationType,
+ final boolean browseOnly,
+ final long creationTime,
+ final boolean durable,
+ final String filter)
+ {
+ this.consumerID = consumerID;
+ this.connectionID = connectionID;
+ this.destinationName = destinationName;
+ this.destinationType = destinationType;
+ this.browseOnly = browseOnly;
+ this.creationTime = creationTime;
+ this.durable = durable;
+ this.filter = filter;
+ }
+
+ // Public --------------------------------------------------------
+
+ public String getConsumerID()
+ {
+ return consumerID;
+ }
+
+ public String getConnectionID()
+ {
+ return connectionID;
+ }
+
+ public String getDestinationName()
+ {
+ return destinationName;
+ }
+
+ public String getDestinationType()
+ {
+ return destinationType;
+ }
+
+ public boolean isBrowseOnly()
+ {
+ return browseOnly;
+ }
+
+ public long getCreationTime()
+ {
+ return creationTime;
+ }
+
+ /**
+ * @return the durable
+ */
+ public boolean isDurable()
+ {
+ return durable;
+ }
+
+ public String getFilter()
+ {
+ return filter;
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
Modified: branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSServerControl.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSServerControl.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/api/jms/management/JMSServerControl.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -228,6 +228,13 @@
@Operation(desc = "List all the connection IDs", impact = MBeanOperationInfo.INFO)
String[] listConnectionIDs() throws Exception;
+ /**
+ * Lists all the connections connected to this server.
+ * The returned String is a JSON string containing an array of JMSConnectionInfo objects.
+ *
+ * @see JMSConnectionInfo#from(String)
+ */
+ @Operation(desc = "List all JMS connections")
String listConnectionsAsJSON() throws Exception;
/**
@@ -236,5 +243,14 @@
@Operation(desc = "List the sessions for the given connectionID", impact = MBeanOperationInfo.INFO)
String[] listSessions(@Parameter(desc = "a connection ID", name = "connectionID") String connectionID) throws Exception;
+ /**
+ * Lists all the consumers which belongs to the JMS Connection specified by the connectionID.
+ * The returned String is a JSON string containing an array of JMSConsumerInfo objects.
+ *
+ * @see JMSConsumerInfo#from(String)
+ */
+ @Operation(desc = "List all JMS consumers associated to a JMS Connection")
+ String listConsumersAsJSON(@Parameter(desc = "a connection ID", name = "connectionID") String connectionID) throws Exception;
+
}
Modified: branches/hornetq-416/src/main/org/hornetq/core/server/Queue.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/core/server/Queue.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/core/server/Queue.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -157,4 +157,6 @@
void close() throws Exception;
boolean isDirectDeliver();
+
+ SimpleString getAddress();
}
Modified: branches/hornetq-416/src/main/org/hornetq/core/server/ServerConsumer.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/core/server/ServerConsumer.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/core/server/ServerConsumer.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -47,6 +47,10 @@
void forceDelivery(long sequence);
void setTransferring(boolean transferring);
+
+ boolean isBrowseOnly();
+
+ long getCreationTime();
}
Modified: branches/hornetq-416/src/main/org/hornetq/core/server/ServerSession.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/core/server/ServerSession.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/core/server/ServerSession.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -14,6 +14,7 @@
package org.hornetq.core.server;
import java.util.List;
+import java.util.Set;
import javax.transaction.xa.Xid;
@@ -110,4 +111,6 @@
void close(boolean failed) throws Exception;
void setTransferring(boolean transferring);
+
+ Set<ServerConsumer> getServerConsumers();
}
Modified: branches/hornetq-416/src/main/org/hornetq/core/server/impl/QueueImpl.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/core/server/impl/QueueImpl.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -238,6 +238,11 @@
{
return name;
}
+
+ public SimpleString getAddress()
+ {
+ return address;
+ }
public long getID()
{
Modified: branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerConsumerImpl.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -118,6 +118,8 @@
private boolean transferring = false;
+ private final long creationTime;
+
// Constructors ---------------------------------------------------------------------------------
public ServerConsumerImpl(final long id,
@@ -160,6 +162,8 @@
this.strictUpdateDeliveryCount = strictUpdateDeliveryCount;
+ this.creationTime = System.currentTimeMillis();
+
if (browseOnly)
{
browserDeliverer = new BrowserDeliverer(messageQueue.iterator());
@@ -177,7 +181,17 @@
{
return id;
}
+
+ public boolean isBrowseOnly()
+ {
+ return browseOnly;
+ }
+ public long getCreationTime()
+ {
+ return creationTime;
+ }
+
public HandleStatus handle(final MessageReference ref) throws Exception
{
if (availableCredits != null && availableCredits.get() <= 0)
Modified: branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/core/server/impl/ServerSessionImpl.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -16,6 +16,7 @@
import static org.hornetq.api.core.management.NotificationType.CONSUMER_CREATED;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -232,6 +233,11 @@
{
return remotingConnection.getID();
}
+
+ public Set<ServerConsumer> getServerConsumers() {
+ Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values());
+ return Collections.unmodifiableSet(consumersClone);
+ }
public void removeConsumer(final long consumerID) throws Exception
{
Modified: branches/hornetq-416/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java
===================================================================
--- branches/hornetq-416/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -36,8 +36,13 @@
import org.hornetq.api.jms.management.JMSQueueControl;
import org.hornetq.api.jms.management.JMSServerControl;
import org.hornetq.api.jms.management.TopicControl;
+import org.hornetq.core.filter.Filter;
import org.hornetq.core.management.impl.MBeanInfoHelper;
+import org.hornetq.core.server.ServerConsumer;
import org.hornetq.core.server.ServerSession;
+import org.hornetq.jms.client.HornetQDestination;
+import org.hornetq.jms.client.HornetQQueue;
+import org.hornetq.jms.client.HornetQTopic;
import org.hornetq.jms.server.JMSServerManager;
import org.hornetq.spi.core.protocol.RemotingConnection;
import org.hornetq.utils.json.JSONArray;
@@ -90,6 +95,38 @@
}
return trimmed;
}
+
+ private static String[] determineJMSDestination(String coreAddress)
+ {
+ String[] result = new String[2]; // destination name & type
+ if (coreAddress.startsWith(HornetQQueue.JMS_QUEUE_ADDRESS_PREFIX))
+ {
+ result[0] = coreAddress.substring(HornetQQueue.JMS_QUEUE_ADDRESS_PREFIX.length());
+ result[1] = "queue";
+ }
+ else if (coreAddress.startsWith(HornetQQueue.JMS_TEMP_QUEUE_ADDRESS_PREFIX))
+ {
+ result[0] = coreAddress.substring(HornetQQueue.JMS_TEMP_QUEUE_ADDRESS_PREFIX.length());
+ result[1] = "tempqueue";
+ }
+ else if (coreAddress.startsWith(HornetQQueue.JMS_TOPIC_ADDRESS_PREFIX))
+ {
+ result[0] = coreAddress.substring(HornetQQueue.JMS_TOPIC_ADDRESS_PREFIX.length());
+ result[1] = "topic";
+ }
+ else if (coreAddress.startsWith(HornetQQueue.JMS_TEMP_TOPIC_ADDRESS_PREFIX))
+ {
+ result[0] = coreAddress.substring(HornetQQueue.JMS_TEMP_TOPIC_ADDRESS_PREFIX.length());
+ result[1] = "temptopic";
+ }
+ else
+ {
+ System.out.println("JMSServerControlImpl.determineJMSDestination()" + coreAddress);
+ // not related to JMS
+ return null;
+ }
+ return result;
+ }
private static List<Pair<TransportConfiguration, TransportConfiguration>> convertToConnectorPairs(final Object[] liveConnectorsTransportClassNames,
final Object[] liveConnectorTransportParams,
@@ -598,7 +635,74 @@
blockOnIO();
}
}
-
+
+ public String listConsumersAsJSON(String connectionID) throws Exception
+ {
+ checkStarted();
+
+ clearIO();
+
+ try
+ {
+ JSONArray array = new JSONArray();
+
+ Set<RemotingConnection> connections = server.getHornetQServer().getRemotingService().getConnections();
+ for (RemotingConnection connection : connections)
+ {
+ if (connectionID.equals(connection.getID().toString()))
+ {
+ List<ServerSession> sessions = server.getHornetQServer().getSessions(connectionID);
+ for (ServerSession session : sessions)
+ {
+ Set<ServerConsumer> consumers = session.getServerConsumers();
+ for (ServerConsumer consumer : consumers)
+ {
+ JSONObject obj = new JSONObject();
+ obj.put("consumerID", consumer.getID());
+ obj.put("connectionID", connectionID);
+ obj.put("queueName", consumer.getQueue().getName().toString());
+ obj.put("browseOnly", consumer.isBrowseOnly());
+ obj.put("creationTime", consumer.getCreationTime());
+ // JMS consumer with message filter use the queue's filter
+ Filter queueFilter = consumer.getQueue().getFilter();
+ if (queueFilter != null)
+ {
+ obj.put("filter", queueFilter.getFilterString().toString());
+ }
+ String[] destinationInfo = determineJMSDestination(consumer.getQueue().getAddress().toString());
+ if (destinationInfo == null)
+ {
+ continue;
+ }
+ obj.put("destinationName", destinationInfo[0]);
+ obj.put("destinationType", destinationInfo[1]);
+ if (destinationInfo[1].equals("topic")) {
+ try
+ {
+ HornetQDestination.decomposeQueueNameForDurableSubscription(consumer.getQueue().getName().toString());
+ obj.put("durable", true);
+ } catch (IllegalArgumentException e)
+ {
+ obj.put("durable", false);
+ }
+ }
+ else
+ {
+ obj.put("durable", false);
+ }
+ array.put(obj);
+ }
+ }
+ }
+ }
+ return array.toString();
+ }
+ finally
+ {
+ blockOnIO();
+ }
+ }
+
public String[] listSessions(final String connectionID) throws Exception
{
checkStarted();
Modified: branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java
===================================================================
--- branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControl2Test.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -21,11 +21,19 @@
import javax.jms.ConnectionFactory;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
+import javax.jms.MessageConsumer;
+import javax.jms.Queue;
+import javax.jms.QueueBrowser;
+import javax.jms.Session;
+import javax.jms.TemporaryTopic;
+import javax.jms.Topic;
import junit.framework.Assert;
import org.hornetq.api.core.TransportConfiguration;
+import org.hornetq.api.jms.HornetQJMSClient;
import org.hornetq.api.jms.management.JMSConnectionInfo;
+import org.hornetq.api.jms.management.JMSConsumerInfo;
import org.hornetq.api.jms.management.JMSServerControl;
import org.hornetq.core.config.Configuration;
import org.hornetq.core.config.impl.ConfigurationImpl;
@@ -36,6 +44,7 @@
import org.hornetq.core.remoting.impl.netty.NettyConnectorFactory;
import org.hornetq.core.server.HornetQServer;
import org.hornetq.core.server.HornetQServers;
+import org.hornetq.jms.client.HornetQQueue;
import org.hornetq.jms.server.impl.JMSServerManagerImpl;
import org.hornetq.tests.integration.management.ManagementControlHelper;
import org.hornetq.tests.integration.management.ManagementTestBase;
@@ -150,7 +159,195 @@
{
doListConnectionsAsJSON(InVMAcceptorFactory.class.getName(), InVMConnectorFactory.class.getName());
}
+
+ public void testListConsumersAsJSON() throws Exception
+ {
+ String queueName = RandomUtil.randomString();
+ try
+ {
+ startHornetQServer(NettyAcceptorFactory.class.getName());
+ serverManager.createQueue(false, queueName, null, true, queueName);
+ Queue queue = HornetQJMSClient.createQueue(queueName);
+
+ JMSServerControl control = createManagementControl();
+
+ long startTime = System.currentTimeMillis();
+
+ String jsonStr = control.listConnectionsAsJSON();
+ assertNotNull(jsonStr);
+ JMSConnectionInfo[] infos = JMSConnectionInfo.from(jsonStr);
+ assertEquals(0, infos.length);
+
+ ConnectionFactory cf1 = JMSUtil.createFactory(NettyConnectorFactory.class.getName(),
+ JMSServerControl2Test.CONNECTION_TTL,
+ JMSServerControl2Test.PING_PERIOD);
+ Connection connection = cf1.createConnection();
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ TemporaryTopic temporaryTopic = session.createTemporaryTopic();
+
+ // create a regular message consumer
+ MessageConsumer consumer = session.createConsumer(queue);
+
+ jsonStr = control.listConnectionsAsJSON();
+ assertNotNull(jsonStr);
+ infos = JMSConnectionInfo.from(jsonStr);
+ assertEquals(1, infos.length);
+ String connectionID = infos[0].getConnectionID();
+
+ String consJsonStr = control.listConsumersAsJSON(connectionID);
+ assertNotNull(consJsonStr);
+ JMSConsumerInfo[] consumerInfos = JMSConsumerInfo.from(consJsonStr);
+ assertEquals(1, consumerInfos.length);
+ JMSConsumerInfo consumerInfo = consumerInfos[0];
+ assertNotNull(consumerInfo.getConsumerID());
+ assertEquals(connectionID, consumerInfo.getConnectionID());
+ assertEquals(queue.getQueueName(), consumerInfo.getDestinationName());
+ assertEquals("queue", consumerInfo.getDestinationType());
+ assertNull(consumerInfo.getFilter());
+ assertEquals(false, consumerInfo.isBrowseOnly());
+ assertEquals(false, consumerInfo.isDurable());
+ assertTrue(startTime <= consumerInfo.getCreationTime() && consumerInfo.getCreationTime() <= System.currentTimeMillis());
+
+ consumer.close();
+
+ consJsonStr = control.listConsumersAsJSON(connectionID);
+ assertNotNull(consJsonStr);
+ consumerInfos = JMSConsumerInfo.from(consJsonStr);
+ assertEquals(0, consumerInfos.length);
+
+ // create a queue browser
+ QueueBrowser browser = session.createBrowser(queue);
+ // the server resources are created when the browser starts enumerating
+ browser.getEnumeration();
+
+ consJsonStr = control.listConsumersAsJSON(connectionID);
+ assertNotNull(consJsonStr);
+ System.out.println(consJsonStr);
+ consumerInfos = JMSConsumerInfo.from(consJsonStr);
+ assertEquals(1, consumerInfos.length);
+ consumerInfo = consumerInfos[0];
+ assertNotNull(consumerInfo.getConsumerID());
+ assertEquals(connectionID, consumerInfo.getConnectionID());
+ assertEquals(queue.getQueueName(), consumerInfo.getDestinationName());
+ assertEquals("queue", consumerInfo.getDestinationType());
+ assertNull(consumerInfo.getFilter());
+ assertEquals(true, consumerInfo.isBrowseOnly());
+ assertEquals(false, consumerInfo.isDurable());
+ assertTrue(startTime <= consumerInfo.getCreationTime() && consumerInfo.getCreationTime() <= System.currentTimeMillis());
+
+ browser.close();
+
+ // create a regular consumer w/ filter on a temp topic
+ String filter = "color = 'red'";
+ consumer = session.createConsumer(temporaryTopic, filter);
+
+ consJsonStr = control.listConsumersAsJSON(connectionID);
+ assertNotNull(consJsonStr);
+ System.out.println(consJsonStr);
+ consumerInfos = JMSConsumerInfo.from(consJsonStr);
+ assertEquals(1, consumerInfos.length);
+ consumerInfo = consumerInfos[0];
+ assertNotNull(consumerInfo.getConsumerID());
+ assertEquals(connectionID, consumerInfo.getConnectionID());
+ assertEquals(temporaryTopic.getTopicName(), consumerInfo.getDestinationName());
+ assertEquals("temptopic", consumerInfo.getDestinationType());
+ assertEquals(filter, consumerInfo.getFilter());
+ assertEquals(false, consumerInfo.isBrowseOnly());
+ assertEquals(false, consumerInfo.isDurable());
+ assertTrue(startTime <= consumerInfo.getCreationTime() && consumerInfo.getCreationTime() <= System.currentTimeMillis());
+
+ consumer.close();
+
+ connection.close();
+ }
+ finally
+ {
+ if (serverManager != null)
+ {
+ serverManager.destroyQueue(queueName);
+ serverManager.stop();
+ }
+
+ if (server != null)
+ {
+ server.stop();
+ }
+ }
+ }
+
+ /**
+ * test for durable subscriber
+ */
+ public void testListConsumersAsJSON2() throws Exception
+ {
+ String topicName = RandomUtil.randomString();
+ String clientID = RandomUtil.randomString();
+ String subName = RandomUtil.randomString();
+
+ try
+ {
+ startHornetQServer(NettyAcceptorFactory.class.getName());
+ serverManager.createTopic(false, topicName, topicName);
+ Topic topic = HornetQJMSClient.createTopic(topicName);
+
+ JMSServerControl control = createManagementControl();
+
+ long startTime = System.currentTimeMillis();
+
+ String jsonStr = control.listConnectionsAsJSON();
+ assertNotNull(jsonStr);
+ JMSConnectionInfo[] infos = JMSConnectionInfo.from(jsonStr);
+ assertEquals(0, infos.length);
+
+ ConnectionFactory cf1 = JMSUtil.createFactory(NettyConnectorFactory.class.getName(),
+ JMSServerControl2Test.CONNECTION_TTL,
+ JMSServerControl2Test.PING_PERIOD);
+ Connection connection = cf1.createConnection();
+ connection.setClientID(clientID);
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ // create a durable subscriber
+ MessageConsumer consumer = session.createDurableSubscriber(topic, subName);
+
+ jsonStr = control.listConnectionsAsJSON();
+ assertNotNull(jsonStr);
+ infos = JMSConnectionInfo.from(jsonStr);
+ assertEquals(1, infos.length);
+ String connectionID = infos[0].getConnectionID();
+
+ String consJsonStr = control.listConsumersAsJSON(connectionID);
+ assertNotNull(consJsonStr);
+ JMSConsumerInfo[] consumerInfos = JMSConsumerInfo.from(consJsonStr);
+ assertEquals(1, consumerInfos.length);
+ JMSConsumerInfo consumerInfo = consumerInfos[0];
+ assertNotNull(consumerInfo.getConsumerID());
+ assertEquals(connectionID, consumerInfo.getConnectionID());
+ assertEquals(topic.getTopicName(), consumerInfo.getDestinationName());
+ assertEquals("topic", consumerInfo.getDestinationType());
+ assertNull(consumerInfo.getFilter());
+ assertEquals(false, consumerInfo.isBrowseOnly());
+ assertEquals(true, consumerInfo.isDurable());
+ assertTrue(startTime <= consumerInfo.getCreationTime() && consumerInfo.getCreationTime() <= System.currentTimeMillis());
+
+ consumer.close();
+
+ connection.close();
+ }
+ finally
+ {
+ if (serverManager != null)
+ {
+ serverManager.destroyTopic(topicName);
+ serverManager.stop();
+ }
+
+ if (server != null)
+ {
+ server.stop();
+ }
+ }
+ }
// Package protected ---------------------------------------------
// Protected -----------------------------------------------------
@@ -246,7 +443,7 @@
{
assertNotNull(info.getConnectionID());
assertNotNull(info.getClientAddress());
- assertTrue(startTime < info.getCreationTime() && info.getCreationTime() < System.currentTimeMillis());
+ assertTrue(startTime <= info.getCreationTime() && info.getCreationTime() <= System.currentTimeMillis());
}
ConnectionFactory cf2 = JMSUtil.createFactory(connectorFactory,
@@ -262,7 +459,7 @@
{
assertNotNull(info.getConnectionID());
assertNotNull(info.getClientAddress());
- assertTrue(startTime < info.getCreationTime() && info.getCreationTime() < System.currentTimeMillis());
+ assertTrue(startTime <= info.getCreationTime() && info.getCreationTime() <= System.currentTimeMillis());
}
connection.close();
Modified: branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java
===================================================================
--- branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/tests/src/org/hornetq/tests/integration/jms/server/management/JMSServerControlUsingJMSTest.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -238,6 +238,11 @@
{
return (String)proxy.invokeOperation("listConnectionsAsJSON");
}
+
+ public String listConsumersAsJSON(String connectionID) throws Exception
+ {
+ return (String)proxy.invokeOperation("listConsumersAsJSON", connectionID);
+ }
public String[] listRemoteAddresses() throws Exception
{
Modified: branches/hornetq-416/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java
===================================================================
--- branches/hornetq-416/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-09-21 22:27:14 UTC (rev 9710)
+++ branches/hornetq-416/tests/src/org/hornetq/tests/unit/core/postoffice/impl/FakeQueue.java 2010-09-22 09:07:31 UTC (rev 9711)
@@ -348,6 +348,12 @@
return name;
}
+ public SimpleString getAddress()
+ {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
/* (non-Javadoc)
* @see org.hornetq.core.server.Queue#getID()
*/
More information about the hornetq-commits
mailing list