You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2006/02/22 00:14:17 UTC
svn commit: r379619 [29/30] - in /incubator/activemq/trunk: ./
activecluster/ activecluster/src/java/org/apache/activecluster/
activecluster/src/java/org/apache/activecluster/election/
activecluster/src/java/org/apache/activecluster/election/impl/ acti...
Modified: incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/QueueSystemTest.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/QueueSystemTest.java?rev=379619&r1=379618&r2=379619&view=diff
==============================================================================
--- incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/QueueSystemTest.java (original)
+++ incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/QueueSystemTest.java Tue Feb 21 15:12:56 2006
@@ -1,116 +1,116 @@
-package org.activemq.usecases;
-
-public class QueueSystemTest extends SystemTestSupport {
-
- /**
- * Unit test for persistent queue messages with the following settings:
- * 1 Producer, 1 Consumer, 1 Subject, 10 Messages
- *
- * @throws Exception
- */
- ///*
- public void testPersistentQueueMessageA() throws Exception{
- SystemTestSupport st = new SystemTestSupport(false,
- true,
- false,
- 1,
- 1,
- 1,
- 10,
- "testPersistentQueueMessageA()");
- st.doTest();
- }
-
-
- /**
- * Unit test for persistent queue messages with the following settings:
- * 10 Producers, 10 Consumers, 1 Subject, 10 Messages
- *
- * @throws Exception
- */
- public void testPersistentQueueMessageB() throws Exception{
- SystemTestSupport st = new SystemTestSupport(false,
- true,
- false,
- 10,
- 10,
- 1,
- 10,
- "testPersistentQueueMessageB()");
- st.doTest();
- }
-
-
- /**
- * Unit test for persistent queue messages with the following settings:
- * 10 Producers, 10 Consumers, 10 Subjects, 10 Messages
- *
- * @throws Exception
- */
- public void testPersistentQueueMessageC() throws Exception{
- SystemTestSupport st = new SystemTestSupport(false,
- true,
- false,
- 10,
- 10,
- 10,
- 10,
- "testPersistentQueueMessageC()");
- st.doTest();
- }
-
- /**
- * Unit test for non-persistent queue messages with the following settings:
- * 1 Producer, 1 Consumer, 1 Subject, 10 Messages
- *
- * @throws Exception
- */
- public void testNonPersistentQueueMessageA() throws Exception {
- SystemTestSupport st = new SystemTestSupport(false,
- false,
- false,
- 1,
- 1,
- 1,
- 10,
- "testNonPersistentQueueMessageA()");
- st.doTest();
- }
-
- /**
- * Unit test for non-persistent queue messages with the following settings:
- * 10 Producers, 10 Consumers, 1 Subject, 10 Messages
- *
- * @throws Exception
- */
- public void testNonPersistentQueueMessageB() throws Exception{
- SystemTestSupport st = new SystemTestSupport(false,
- false,
- false,
- 10,
- 10,
- 1,
- 10,
- "testNonPersistentQueueMessageB()");
- st.doTest();
- }
-
-
- /**
- * Unit test for non-persistent queue messages with the following settings:
- * 10 Producers, 10 Consumers, 10 Subjects, 10 Messages
- *
- * @throws Exception
- */
- public void testNonPersistentQueueMessageC() throws Exception{
- SystemTestSupport st = new SystemTestSupport(false,
- false,
- false,
- 10,
- 10,
- 10,
- 10,
- "testNonPersistentQueueMessageC()");
- st.doTest();
- }
+package org.activemq.usecases;
+
+public class QueueSystemTest extends SystemTestSupport {
+
+ /**
+ * Unit test for persistent queue messages with the following settings:
+ * 1 Producer, 1 Consumer, 1 Subject, 10 Messages
+ *
+ * @throws Exception
+ */
+ ///*
+ public void testPersistentQueueMessageA() throws Exception{
+ SystemTestSupport st = new SystemTestSupport(false,
+ true,
+ false,
+ 1,
+ 1,
+ 1,
+ 10,
+ "testPersistentQueueMessageA()");
+ st.doTest();
+ }
+
+
+ /**
+ * Unit test for persistent queue messages with the following settings:
+ * 10 Producers, 10 Consumers, 1 Subject, 10 Messages
+ *
+ * @throws Exception
+ */
+ public void testPersistentQueueMessageB() throws Exception{
+ SystemTestSupport st = new SystemTestSupport(false,
+ true,
+ false,
+ 10,
+ 10,
+ 1,
+ 10,
+ "testPersistentQueueMessageB()");
+ st.doTest();
+ }
+
+
+ /**
+ * Unit test for persistent queue messages with the following settings:
+ * 10 Producers, 10 Consumers, 10 Subjects, 10 Messages
+ *
+ * @throws Exception
+ */
+ public void testPersistentQueueMessageC() throws Exception{
+ SystemTestSupport st = new SystemTestSupport(false,
+ true,
+ false,
+ 10,
+ 10,
+ 10,
+ 10,
+ "testPersistentQueueMessageC()");
+ st.doTest();
+ }
+
+ /**
+ * Unit test for non-persistent queue messages with the following settings:
+ * 1 Producer, 1 Consumer, 1 Subject, 10 Messages
+ *
+ * @throws Exception
+ */
+ public void testNonPersistentQueueMessageA() throws Exception {
+ SystemTestSupport st = new SystemTestSupport(false,
+ false,
+ false,
+ 1,
+ 1,
+ 1,
+ 10,
+ "testNonPersistentQueueMessageA()");
+ st.doTest();
+ }
+
+ /**
+ * Unit test for non-persistent queue messages with the following settings:
+ * 10 Producers, 10 Consumers, 1 Subject, 10 Messages
+ *
+ * @throws Exception
+ */
+ public void testNonPersistentQueueMessageB() throws Exception{
+ SystemTestSupport st = new SystemTestSupport(false,
+ false,
+ false,
+ 10,
+ 10,
+ 1,
+ 10,
+ "testNonPersistentQueueMessageB()");
+ st.doTest();
+ }
+
+
+ /**
+ * Unit test for non-persistent queue messages with the following settings:
+ * 10 Producers, 10 Consumers, 10 Subjects, 10 Messages
+ *
+ * @throws Exception
+ */
+ public void testNonPersistentQueueMessageC() throws Exception{
+ SystemTestSupport st = new SystemTestSupport(false,
+ false,
+ false,
+ 10,
+ 10,
+ 10,
+ 10,
+ "testNonPersistentQueueMessageC()");
+ st.doTest();
+ }
}
Propchange: incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/QueueSystemTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/SystemTestSupport.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/SystemTestSupport.java?rev=379619&r1=379618&r2=379619&view=diff
==============================================================================
--- incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/SystemTestSupport.java (original)
+++ incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/SystemTestSupport.java Tue Feb 21 15:12:56 2006
@@ -1,553 +1,553 @@
-package org.activemq.usecases;
-
-import junit.framework.TestCase;
-
-import javax.jms.*;
-import java.util.*;
-import java.net.URI;
-
-import org.apache.activemq.ActiveMQConnection;
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.command.ActiveMQMessage;
-import org.apache.activemq.command.ActiveMQDestination;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.BrokerFactory;
-import org.apache.activemq.broker.TransportConnector;
-
-import org.apache.jorphan.logging.LoggingManager;
-import org.apache.log.Logger;
-
-import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
-
-public class SystemTestSupport extends TestCase implements MessageListener {
- public static final String ACTIVEMQ_SERVER = "ActiveMQ Server";
- public static final boolean TRANSACTED_FALSE = false;
- public static final String TOOL_DEFAULT = "TOOL.DEFAULT";
- public static final String LAST_MESSAGE = "#LAST";
-
- public String userName;
- public String password;
-
- private static int msgCounter = 0;
- public static Map ProducerMap = Collections.synchronizedMap(new HashMap());
-
- private int producerCount = 0;
- private int consumerCount = 0;
- private int subjectCount = 0;
- private int messageCount = 0;
- private boolean isPersistent = true;
- private boolean isDurable = true;
- private boolean isTopic = true;
- private boolean testStillRunning = true;
-
- protected BrokerService broker;
-
- Map consumerMap = new ConcurrentHashMap();
- Map prodNameMap = new TreeMap();
- Map prodMsgMap = new TreeMap();
-
- /**
- * Default constructor
- */
- protected SystemTestSupport(){
- super();
- ProducerMap.clear();
- msgCounter = 0;
- testStillRunning = true;
- }
-
- /**
- * Constructor
- *
- * @param isTopic - true when topic, false when queue.
- * @param isPersistent - true when the delivery mode is persistent.
- * @param isDurable - true when the suscriber is durable(For topic only).
- * @param producerCount - number of producers.
- * @param consumerCount - number of consumers.
- * @param subjectCount - number of destinations.
- * @param messageCount - number of messages to be delivered.
- * @param testTitle - test title/name.
- *
- */
- protected SystemTestSupport(boolean isTopic,
- boolean isPersistent,
- boolean isDurable,
- int producerCount,
- int consumerCount,
- int subjectCount,
- int messageCount,
- String testTitle){
- super();
- this.isTopic = isTopic;
- this.isPersistent = isPersistent;
- this.isDurable = isDurable;
- this.producerCount = producerCount;
- this.consumerCount = consumerCount;
- this.subjectCount = subjectCount;
- this.messageCount = messageCount;
- this.testParameterSettings(testTitle);
-
- ProducerMap.clear();
- consumerMap.clear();
- prodNameMap.clear();
- prodMsgMap.clear();
- msgCounter = 0;
- testStillRunning = true;
- }
-
- /****************************************************
- *
- * Producer section
- *
- ****************************************************/
-
- /**
- * Creates the message producer threads.
- */
- protected void publish() throws JMSException {
- String subjects[] = getSubjects();
-
- for (int i = 0; i < producerCount; i++) {
- final int x = i;
- final String subject = subjects[i % subjects.length];
-
- Thread thread = new Thread() {
- public void run() {
- try {
- publish(x, subject);
-
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- };
-
- thread.start();
- }
- }
-
- /**
- * Creates the producer and send the messages.
- *
- * @param x - producer number.
- * @param subject - the destination where the messages will be sent.
- */
- protected void publish(int x, String subject) throws Exception {
- ConnectionFactory factory = createConnectionFactory();
- ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection(userName, password);
- connection.start();
-
- String messageBody;
-
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination destination = createDestination(session, subject);
- MessageProducer publisher = session.createProducer(destination);
-
- if (isPersistent) {
- publisher.setDeliveryMode(DeliveryMode.PERSISTENT);
- } else {
- publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
- }
-
- StringBuffer sb = new StringBuffer();
-
- //Sending messages
- for (int i = 1; i <= messageCount; ++i) {
- if (messageCount != i)
- messageBody = "#BODY";
- else
- messageBody = LAST_MESSAGE;
-
- sb.append("PROD");
- sb.append(x);
- sb.append(messageBody);
- sb.append("#");
- sb.append(i);
-
- TextMessage message = session.createTextMessage(sb.toString());
- publisher.send(message);
-
- sb.delete(0, sb.length());
- }
-
- publisher.close();
- session.close();
- connection.stop();
- connection = null;
- }
-
- /****************************************************
- *
- * Consumer section
- *
- ****************************************************/
-
- /**
- * Generates the topic/queue destinations.
- *
- * @return String[] - topic/queue destination name.
- */
- protected String[] getSubjects() {
- //Create the subjects.
- String[] subjects = new String[subjectCount];
-
- //Appended to the subject to determine if its a queue or topic.
- String prefix = null;
- if (this.isTopic) {
- prefix = ".TOPIC";
- } else {
- prefix = ".QUEUE";
- }
-
- for (int i = 0; i < subjects.length; i++) {
- subjects[i] = TOOL_DEFAULT + prefix + i;
- }
-
- return subjects;
- }
-
- /**
- * Suscribes the consumers to the topic/queue destinations.
- */
- protected void subscribe() throws Exception {
- String subjects[] = getSubjects();
-
- for (int i = 0; i < consumerCount; i++) {
- String subject = subjects[i % subjectCount];
- subscribe(subject);
- }
- }
-
- /**
- * Suscribes the consumer to the topic/queue specified by the subject.
- *
- * @param subject - the Destination where the consumer waits upon for messages.
- */
- protected void subscribe(String subject) throws Exception {
- ConnectionFactory factory = createConnectionFactory();
- ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection(userName, password);
- connection.start();
-
- Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- Destination destination = createDestination(session, subject);
-
- MessageConsumer consumer = session.createConsumer(destination);
- consumer.setMessageListener(this);
- connection.stop();
- }
-
- /**
- * Processes the received message.
- *
- * @param message - message received by the listener.
- */
- public void onMessage(Message message) {
- try {
- ActiveMQMessage amsg = (ActiveMQMessage) message;
- TextMessage textMessage = (TextMessage) message;
-
- StringBuffer sb = new StringBuffer();
- sb.append(textMessage.getText());
- sb.append("#");
- sb.append(amsg.getJMSMessageID());
-
- msgCounter++;
- String strMsgCounter = String.valueOf(msgCounter);
- ProducerMap.put(strMsgCounter, sb.toString());
-
- } catch (JMSException e) {
- System.out.println("Unable to force deserialize the content " + e);
- }
- }
-
- /**
- * Validates the result of the producing and consumption of messages by the server.
- * It checks for duplicate messages, message count and order.
- */
- protected synchronized void timerLoop() {
- System.out.println("MessagingSystemTest.timerLoop() * started * ");
- Map ProducerTextMap = new HashMap();
- Map currentProducerMap = null;
- String producerName = null;
- String msgBody = null;
- String consumerName = null;
- String ProdSequenceNo = null;
- String mapKey = null;
- int expectedNoOfMessages = messageCount;
- boolean dowhile = true;
-
- while (dowhile) {
- try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
-
- //Retrieve the map containing the received messages data
- currentProducerMap = resetProducerMap();
-
- if (currentProducerMap.size() == 0) {
- dowhile = false;
- }
-
- //Put the map values to another map for parsing.
- for (int i = 1; i <= currentProducerMap.size(); i++) {
- String ProdMsg = (String) currentProducerMap.get(String.valueOf(i));
- producerName = ProdMsg.substring(0, ProdMsg.indexOf("#"));
- msgBody = ProdMsg.substring(ProdMsg.indexOf("#") + 1, ProdMsg.lastIndexOf("#"));
- ProdSequenceNo = ProdMsg.substring(ProdMsg.indexOf("#", ProdMsg.indexOf("#", ProdMsg.indexOf("#")+1)) + 1, ProdMsg.lastIndexOf("#"));
- consumerName = ProdMsg.substring(ProdMsg.lastIndexOf("#"), ProdMsg.length());
-
- if (isTopic) {
- mapKey = consumerName + producerName;
- } else {
- mapKey = producerName;
- }
-
- if (ProducerTextMap.containsKey(mapKey)) {
- //Increment the counter value
- Integer value = (Integer) ProducerTextMap.get(mapKey);
- ProducerTextMap.put(mapKey, new Integer(value.intValue() + 1));
- } else {
- //Put the Producer Name in the map
- ProducerTextMap.put(mapKey, new Integer(1));
- }
-
- Integer messageCounter = (Integer) ProducerTextMap.get(mapKey);
- Integer ProducerSeqID = Integer.valueOf(ProdSequenceNo);
-
- if (isTopic) {
- // Check for duplicate message.
- if (messageCounter.intValue() > expectedNoOfMessages) {
- assertTrue("Should not have received duplicate messages!", messageCounter.intValue() <= expectedNoOfMessages);
- break;
- } else if (LAST_MESSAGE.equals(msgBody)) {
- System.out.println("entered MsgBody.equals(LAST_MESSAGE)..." + mapKey +
- " " + messageCounter.intValue() +"=" + expectedNoOfMessages);
-
- // Validates that the messages received is equal to the number
- // of expected messages
- if (messageCounter.intValue() != expectedNoOfMessages) {
- System.out.println("entered messageCounter.intValue() != expectedNoOfMessages...");
-
- // Checks for message order.
- assertTrue("Should have received messages in order!", messageCounter.intValue() == expectedNoOfMessages);
- if (messageCounter.intValue() != expectedNoOfMessages) {
- System.out.println("Should have received messages in order!");
- }
- break;
- } else if (currentProducerMap.size() == i) {
- System.out.println("MessagingSystemTest.timerLoop() says system_test_pass!!!... ");
- break;
- }
- }
-
- } else {
- //Create map for each consumer
- for (int j = 0 ; j < consumerCount ; j++) {
- if (!consumerMap.containsKey(new String(consumerName))) {
- consumerMap.put(new String(consumerName), new LinkedHashMap());
- }
- }
-
- //create Producer Name Map
- if (!prodNameMap.containsKey(producerName)) {
- prodNameMap.put(producerName, (null));
- }
-
- //Get the current size of consumer
- int seqVal = 0;
- Object[] cObj = consumerMap.keySet().toArray();
- for (int k = 0; k < cObj.length; k++) {
- String cMapKey = (String)cObj[k];
- Map cMapVal = (Map)consumerMap.get(cObj[k]);
- if (cMapKey.equals(consumerName)) {
- seqVal = cMapVal.size();
- break;
- }
- }
-
- //Put object to its designated consumer map
- Object[] consumerObj = consumerMap.keySet().toArray();
- for (int j = 0; j < consumerObj.length; j++) {
- String cMapKey = (String)consumerObj[j];
- Map cMapVal = (LinkedHashMap)consumerMap.get(consumerObj[j]);
- if (cMapKey.equals(consumerName)) {
- cMapVal.put(new Integer(seqVal), (producerName + "/" + ProducerSeqID));
- }
- }
- }
-
- // Add data to table row
- if (!isTopic) {
- String msgKey = consumerName + "#" + producerName + "#" + String.valueOf(ProdSequenceNo);
- String msgVal = String.valueOf(messageCounter) + "#" + msgBody;
-
- if (!prodMsgMap.containsKey(msgKey)) {
- prodMsgMap.put((msgKey), (msgVal));
- }
- }
- }
- }
-
- if (!isTopic) {
- //Validate message sequence
- boolean isMsgNotOrdered = validateMsg(prodNameMap, consumerMap);
- assertFalse("Should have received messages in order!", isMsgNotOrdered);
- }
-
- testStillRunning = false;
- System.out.println("MessagingSystemTest.timerLoop() * ended * ");
- }
-
- /**
- * Returns the message entries and clears the map for another set
- * of messages to be processed.
- *
- * @return Map - messages to be processed.
- */
- protected synchronized Map resetProducerMap() {
- Map copy = Collections.synchronizedMap(new HashMap(ProducerMap));
- ProducerMap.clear();
- msgCounter = 0;
-
- return copy;
- }
-
- /****************************************************
- *
- * Utility section
- *
- ****************************************************/
-
- /**
- * Creates the session destination.
- *
- * @param session - connection session.
- * @param subject - destination name.
- * @return Destination - session destination.
- */
- protected ActiveMQDestination createDestination(Session session, String subject)
- throws JMSException {
- if (isTopic) {
- return (ActiveMQDestination) session.createQueue(subject);
- } else {
- return (ActiveMQDestination) session.createTopic(subject);
- }
- }
-
- protected ConnectionFactory createConnectionFactory() throws Exception {
- return new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
- }
-
- /****************************************************
- *
- * Unit test section
- *
- ****************************************************/
-
- /**
- * Executes the unit test by running the producers and consumers.
- * It checks for duplicate messages, message count and order.
- */
- protected void doTest() throws Exception {
- System.out.println("MessagingSystemTest.doTest() * start *");
-
- //Set up the consumers
- subscribe();
-
- System.out.println("MessagingSystemTest.doTest() after suscribe()...");
-
- //Set up the producers
- publish();
-
- System.out.println("MessagingSystemTest.doTest() after publish()...");
-
- //Run the test
- Thread timer = new Thread() {
- public void run() {
- timerLoop();
- }
- };
- timer.setPriority(Thread.MIN_PRIORITY);
- timer.start();
-
- while (testStillRunning) {
- try {
- if (Thread.currentThread() == timer) {
- Thread.sleep(1000);
- }
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- System.out.println("MessagingSystemTest.doTest() * end *");
- }
-
- /**
- * Validates the messages received.
- *
- * @param prodNameMap
- * @param cMap
- * @return
- */
- protected boolean validateMsg(Map prodNameMap, Map cMap) {
- boolean ret = false;
- Object[] cObj = cMap.keySet().toArray();
- for (int j = 0; j < cObj.length; j++) {
- Map childMap = (Map)cMap.get(cObj[j]);
-
- Object[] nameObj = prodNameMap.keySet().toArray();
- for (int i = 0; i < nameObj.length; i++) {
- String prodName = (String)nameObj[i];
- String tempProdHolder = null;
- String tempProdIDHolder = null;
-
- Object[] childObj = childMap.keySet().toArray();
- for (int k = 0; k < childObj.length; k++) {
- Integer childMapKey = (Integer)childObj[k];
- String childMapVal = (String)childMap.get(childObj[k]);
- String prodVal = childMapVal.substring(0, childMapVal.indexOf("/"));
- String prodIDVal = childMapVal.substring(childMapVal.indexOf("/")+1, childMapVal.length());
-
- if (prodVal.equals(prodName)) {
- if (tempProdHolder == null) {
- tempProdHolder = prodVal;
- tempProdIDHolder = prodIDVal;
- continue;
- }
- if (Integer.parseInt(prodIDVal) > Integer.parseInt(tempProdIDHolder)) {
- tempProdHolder = prodVal;
- tempProdIDHolder = prodIDVal;
- } else {
- ret = true;
- break;
- }
- } else {
- continue;
- }
- }
- }
- }
- return ret;
- }
-
- /**
- * Prints the test settings.
- *
- * @param strTestTitle - unit test name.
- */
- public void testParameterSettings(String strTestTitle) {
- System.out.println(strTestTitle);
- System.out.println("============================================================");
- System.out.println("Test settings:");
- System.out.println("isTopic=" + new Boolean(isTopic).toString());
- System.out.println("isPersistent=" + new Boolean(isPersistent).toString());
- System.out.println("isDurable=" + new Boolean(isDurable).toString());
- System.out.println("producerCount=" + producerCount);
- System.out.println("consumerCount=" + consumerCount);
- System.out.println("subjectCount=" + subjectCount);
- System.out.println("messageCount=" + messageCount);
- System.out.println("");
- }
-}
+package org.activemq.usecases;
+
+import junit.framework.TestCase;
+
+import javax.jms.*;
+import java.util.*;
+import java.net.URI;
+
+import org.apache.activemq.ActiveMQConnection;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerFactory;
+import org.apache.activemq.broker.TransportConnector;
+
+import org.apache.jorphan.logging.LoggingManager;
+import org.apache.log.Logger;
+
+import EDU.oswego.cs.dl.util.concurrent.ConcurrentHashMap;
+
+public class SystemTestSupport extends TestCase implements MessageListener {
+ public static final String ACTIVEMQ_SERVER = "ActiveMQ Server";
+ public static final boolean TRANSACTED_FALSE = false;
+ public static final String TOOL_DEFAULT = "TOOL.DEFAULT";
+ public static final String LAST_MESSAGE = "#LAST";
+
+ public String userName;
+ public String password;
+
+ private static int msgCounter = 0;
+ public static Map ProducerMap = Collections.synchronizedMap(new HashMap());
+
+ private int producerCount = 0;
+ private int consumerCount = 0;
+ private int subjectCount = 0;
+ private int messageCount = 0;
+ private boolean isPersistent = true;
+ private boolean isDurable = true;
+ private boolean isTopic = true;
+ private boolean testStillRunning = true;
+
+ protected BrokerService broker;
+
+ Map consumerMap = new ConcurrentHashMap();
+ Map prodNameMap = new TreeMap();
+ Map prodMsgMap = new TreeMap();
+
+ /**
+ * Default constructor
+ */
+ protected SystemTestSupport(){
+ super();
+ ProducerMap.clear();
+ msgCounter = 0;
+ testStillRunning = true;
+ }
+
+ /**
+ * Constructor
+ *
+ * @param isTopic - true when topic, false when queue.
+ * @param isPersistent - true when the delivery mode is persistent.
+ * @param isDurable - true when the suscriber is durable(For topic only).
+ * @param producerCount - number of producers.
+ * @param consumerCount - number of consumers.
+ * @param subjectCount - number of destinations.
+ * @param messageCount - number of messages to be delivered.
+ * @param testTitle - test title/name.
+ *
+ */
+ protected SystemTestSupport(boolean isTopic,
+ boolean isPersistent,
+ boolean isDurable,
+ int producerCount,
+ int consumerCount,
+ int subjectCount,
+ int messageCount,
+ String testTitle){
+ super();
+ this.isTopic = isTopic;
+ this.isPersistent = isPersistent;
+ this.isDurable = isDurable;
+ this.producerCount = producerCount;
+ this.consumerCount = consumerCount;
+ this.subjectCount = subjectCount;
+ this.messageCount = messageCount;
+ this.testParameterSettings(testTitle);
+
+ ProducerMap.clear();
+ consumerMap.clear();
+ prodNameMap.clear();
+ prodMsgMap.clear();
+ msgCounter = 0;
+ testStillRunning = true;
+ }
+
+ /****************************************************
+ *
+ * Producer section
+ *
+ ****************************************************/
+
+ /**
+ * Creates the message producer threads.
+ */
+ protected void publish() throws JMSException {
+ String subjects[] = getSubjects();
+
+ for (int i = 0; i < producerCount; i++) {
+ final int x = i;
+ final String subject = subjects[i % subjects.length];
+
+ Thread thread = new Thread() {
+ public void run() {
+ try {
+ publish(x, subject);
+
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ };
+
+ thread.start();
+ }
+ }
+
+ /**
+ * Creates the producer and send the messages.
+ *
+ * @param x - producer number.
+ * @param subject - the destination where the messages will be sent.
+ */
+ protected void publish(int x, String subject) throws Exception {
+ ConnectionFactory factory = createConnectionFactory();
+ ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection(userName, password);
+ connection.start();
+
+ String messageBody;
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = createDestination(session, subject);
+ MessageProducer publisher = session.createProducer(destination);
+
+ if (isPersistent) {
+ publisher.setDeliveryMode(DeliveryMode.PERSISTENT);
+ } else {
+ publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+ }
+
+ StringBuffer sb = new StringBuffer();
+
+ //Sending messages
+ for (int i = 1; i <= messageCount; ++i) {
+ if (messageCount != i)
+ messageBody = "#BODY";
+ else
+ messageBody = LAST_MESSAGE;
+
+ sb.append("PROD");
+ sb.append(x);
+ sb.append(messageBody);
+ sb.append("#");
+ sb.append(i);
+
+ TextMessage message = session.createTextMessage(sb.toString());
+ publisher.send(message);
+
+ sb.delete(0, sb.length());
+ }
+
+ publisher.close();
+ session.close();
+ connection.stop();
+ connection = null;
+ }
+
+ /****************************************************
+ *
+ * Consumer section
+ *
+ ****************************************************/
+
+ /**
+ * Generates the topic/queue destinations.
+ *
+ * @return String[] - topic/queue destination name.
+ */
+ protected String[] getSubjects() {
+ //Create the subjects.
+ String[] subjects = new String[subjectCount];
+
+ //Appended to the subject to determine if its a queue or topic.
+ String prefix = null;
+ if (this.isTopic) {
+ prefix = ".TOPIC";
+ } else {
+ prefix = ".QUEUE";
+ }
+
+ for (int i = 0; i < subjects.length; i++) {
+ subjects[i] = TOOL_DEFAULT + prefix + i;
+ }
+
+ return subjects;
+ }
+
+ /**
+ * Suscribes the consumers to the topic/queue destinations.
+ */
+ protected void subscribe() throws Exception {
+ String subjects[] = getSubjects();
+
+ for (int i = 0; i < consumerCount; i++) {
+ String subject = subjects[i % subjectCount];
+ subscribe(subject);
+ }
+ }
+
+ /**
+ * Suscribes the consumer to the topic/queue specified by the subject.
+ *
+ * @param subject - the Destination where the consumer waits upon for messages.
+ */
+ protected void subscribe(String subject) throws Exception {
+ ConnectionFactory factory = createConnectionFactory();
+ ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection(userName, password);
+ connection.start();
+
+ Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ Destination destination = createDestination(session, subject);
+
+ MessageConsumer consumer = session.createConsumer(destination);
+ consumer.setMessageListener(this);
+ connection.stop();
+ }
+
+ /**
+ * Processes the received message.
+ *
+ * @param message - message received by the listener.
+ */
+ public void onMessage(Message message) {
+ try {
+ ActiveMQMessage amsg = (ActiveMQMessage) message;
+ TextMessage textMessage = (TextMessage) message;
+
+ StringBuffer sb = new StringBuffer();
+ sb.append(textMessage.getText());
+ sb.append("#");
+ sb.append(amsg.getJMSMessageID());
+
+ msgCounter++;
+ String strMsgCounter = String.valueOf(msgCounter);
+ ProducerMap.put(strMsgCounter, sb.toString());
+
+ } catch (JMSException e) {
+ System.out.println("Unable to force deserialize the content " + e);
+ }
+ }
+
+ /**
+ * Validates the result of the producing and consumption of messages by the server.
+ * It checks for duplicate messages, message count and order.
+ */
+ protected synchronized void timerLoop() {
+ System.out.println("MessagingSystemTest.timerLoop() * started * ");
+ Map ProducerTextMap = new HashMap();
+ Map currentProducerMap = null;
+ String producerName = null;
+ String msgBody = null;
+ String consumerName = null;
+ String ProdSequenceNo = null;
+ String mapKey = null;
+ int expectedNoOfMessages = messageCount;
+ boolean dowhile = true;
+
+ while (dowhile) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+
+ //Retrieve the map containing the received messages data
+ currentProducerMap = resetProducerMap();
+
+ if (currentProducerMap.size() == 0) {
+ dowhile = false;
+ }
+
+ //Put the map values to another map for parsing.
+ for (int i = 1; i <= currentProducerMap.size(); i++) {
+ String ProdMsg = (String) currentProducerMap.get(String.valueOf(i));
+ producerName = ProdMsg.substring(0, ProdMsg.indexOf("#"));
+ msgBody = ProdMsg.substring(ProdMsg.indexOf("#") + 1, ProdMsg.lastIndexOf("#"));
+ ProdSequenceNo = ProdMsg.substring(ProdMsg.indexOf("#", ProdMsg.indexOf("#", ProdMsg.indexOf("#")+1)) + 1, ProdMsg.lastIndexOf("#"));
+ consumerName = ProdMsg.substring(ProdMsg.lastIndexOf("#"), ProdMsg.length());
+
+ if (isTopic) {
+ mapKey = consumerName + producerName;
+ } else {
+ mapKey = producerName;
+ }
+
+ if (ProducerTextMap.containsKey(mapKey)) {
+ //Increment the counter value
+ Integer value = (Integer) ProducerTextMap.get(mapKey);
+ ProducerTextMap.put(mapKey, new Integer(value.intValue() + 1));
+ } else {
+ //Put the Producer Name in the map
+ ProducerTextMap.put(mapKey, new Integer(1));
+ }
+
+ Integer messageCounter = (Integer) ProducerTextMap.get(mapKey);
+ Integer ProducerSeqID = Integer.valueOf(ProdSequenceNo);
+
+ if (isTopic) {
+ // Check for duplicate message.
+ if (messageCounter.intValue() > expectedNoOfMessages) {
+ assertTrue("Should not have received duplicate messages!", messageCounter.intValue() <= expectedNoOfMessages);
+ break;
+ } else if (LAST_MESSAGE.equals(msgBody)) {
+ System.out.println("entered MsgBody.equals(LAST_MESSAGE)..." + mapKey +
+ " " + messageCounter.intValue() +"=" + expectedNoOfMessages);
+
+ // Validates that the messages received is equal to the number
+ // of expected messages
+ if (messageCounter.intValue() != expectedNoOfMessages) {
+ System.out.println("entered messageCounter.intValue() != expectedNoOfMessages...");
+
+ // Checks for message order.
+ assertTrue("Should have received messages in order!", messageCounter.intValue() == expectedNoOfMessages);
+ if (messageCounter.intValue() != expectedNoOfMessages) {
+ System.out.println("Should have received messages in order!");
+ }
+ break;
+ } else if (currentProducerMap.size() == i) {
+ System.out.println("MessagingSystemTest.timerLoop() says system_test_pass!!!... ");
+ break;
+ }
+ }
+
+ } else {
+ //Create map for each consumer
+ for (int j = 0 ; j < consumerCount ; j++) {
+ if (!consumerMap.containsKey(new String(consumerName))) {
+ consumerMap.put(new String(consumerName), new LinkedHashMap());
+ }
+ }
+
+ //create Producer Name Map
+ if (!prodNameMap.containsKey(producerName)) {
+ prodNameMap.put(producerName, (null));
+ }
+
+ //Get the current size of consumer
+ int seqVal = 0;
+ Object[] cObj = consumerMap.keySet().toArray();
+ for (int k = 0; k < cObj.length; k++) {
+ String cMapKey = (String)cObj[k];
+ Map cMapVal = (Map)consumerMap.get(cObj[k]);
+ if (cMapKey.equals(consumerName)) {
+ seqVal = cMapVal.size();
+ break;
+ }
+ }
+
+ //Put object to its designated consumer map
+ Object[] consumerObj = consumerMap.keySet().toArray();
+ for (int j = 0; j < consumerObj.length; j++) {
+ String cMapKey = (String)consumerObj[j];
+ Map cMapVal = (LinkedHashMap)consumerMap.get(consumerObj[j]);
+ if (cMapKey.equals(consumerName)) {
+ cMapVal.put(new Integer(seqVal), (producerName + "/" + ProducerSeqID));
+ }
+ }
+ }
+
+ // Add data to table row
+ if (!isTopic) {
+ String msgKey = consumerName + "#" + producerName + "#" + String.valueOf(ProdSequenceNo);
+ String msgVal = String.valueOf(messageCounter) + "#" + msgBody;
+
+ if (!prodMsgMap.containsKey(msgKey)) {
+ prodMsgMap.put((msgKey), (msgVal));
+ }
+ }
+ }
+ }
+
+ if (!isTopic) {
+ //Validate message sequence
+ boolean isMsgNotOrdered = validateMsg(prodNameMap, consumerMap);
+ assertFalse("Should have received messages in order!", isMsgNotOrdered);
+ }
+
+ testStillRunning = false;
+ System.out.println("MessagingSystemTest.timerLoop() * ended * ");
+ }
+
+ /**
+ * Returns the message entries and clears the map for another set
+ * of messages to be processed.
+ *
+ * @return Map - messages to be processed.
+ */
+ protected synchronized Map resetProducerMap() {
+ Map copy = Collections.synchronizedMap(new HashMap(ProducerMap));
+ ProducerMap.clear();
+ msgCounter = 0;
+
+ return copy;
+ }
+
+ /****************************************************
+ *
+ * Utility section
+ *
+ ****************************************************/
+
+ /**
+ * Creates the session destination.
+ *
+ * @param session - connection session.
+ * @param subject - destination name.
+ * @return Destination - session destination.
+ */
+ protected ActiveMQDestination createDestination(Session session, String subject)
+ throws JMSException {
+ if (isTopic) {
+ return (ActiveMQDestination) session.createQueue(subject);
+ } else {
+ return (ActiveMQDestination) session.createTopic(subject);
+ }
+ }
+
+ protected ConnectionFactory createConnectionFactory() throws Exception {
+ return new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
+ }
+
+ /****************************************************
+ *
+ * Unit test section
+ *
+ ****************************************************/
+
+ /**
+ * Executes the unit test by running the producers and consumers.
+ * It checks for duplicate messages, message count and order.
+ */
+ protected void doTest() throws Exception {
+ System.out.println("MessagingSystemTest.doTest() * start *");
+
+ //Set up the consumers
+ subscribe();
+
+ System.out.println("MessagingSystemTest.doTest() after suscribe()...");
+
+ //Set up the producers
+ publish();
+
+ System.out.println("MessagingSystemTest.doTest() after publish()...");
+
+ //Run the test
+ Thread timer = new Thread() {
+ public void run() {
+ timerLoop();
+ }
+ };
+ timer.setPriority(Thread.MIN_PRIORITY);
+ timer.start();
+
+ while (testStillRunning) {
+ try {
+ if (Thread.currentThread() == timer) {
+ Thread.sleep(1000);
+ }
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ System.out.println("MessagingSystemTest.doTest() * end *");
+ }
+
+ /**
+ * Validates the messages received.
+ *
+ * @param prodNameMap
+ * @param cMap
+ * @return
+ */
+ protected boolean validateMsg(Map prodNameMap, Map cMap) {
+ boolean ret = false;
+ Object[] cObj = cMap.keySet().toArray();
+ for (int j = 0; j < cObj.length; j++) {
+ Map childMap = (Map)cMap.get(cObj[j]);
+
+ Object[] nameObj = prodNameMap.keySet().toArray();
+ for (int i = 0; i < nameObj.length; i++) {
+ String prodName = (String)nameObj[i];
+ String tempProdHolder = null;
+ String tempProdIDHolder = null;
+
+ Object[] childObj = childMap.keySet().toArray();
+ for (int k = 0; k < childObj.length; k++) {
+ Integer childMapKey = (Integer)childObj[k];
+ String childMapVal = (String)childMap.get(childObj[k]);
+ String prodVal = childMapVal.substring(0, childMapVal.indexOf("/"));
+ String prodIDVal = childMapVal.substring(childMapVal.indexOf("/")+1, childMapVal.length());
+
+ if (prodVal.equals(prodName)) {
+ if (tempProdHolder == null) {
+ tempProdHolder = prodVal;
+ tempProdIDHolder = prodIDVal;
+ continue;
+ }
+ if (Integer.parseInt(prodIDVal) > Integer.parseInt(tempProdIDHolder)) {
+ tempProdHolder = prodVal;
+ tempProdIDHolder = prodIDVal;
+ } else {
+ ret = true;
+ break;
+ }
+ } else {
+ continue;
+ }
+ }
+ }
+ }
+ return ret;
+ }
+
+ /**
+ * Prints the test settings.
+ *
+ * @param strTestTitle - unit test name.
+ */
+ public void testParameterSettings(String strTestTitle) {
+ System.out.println(strTestTitle);
+ System.out.println("============================================================");
+ System.out.println("Test settings:");
+ System.out.println("isTopic=" + new Boolean(isTopic).toString());
+ System.out.println("isPersistent=" + new Boolean(isPersistent).toString());
+ System.out.println("isDurable=" + new Boolean(isDurable).toString());
+ System.out.println("producerCount=" + producerCount);
+ System.out.println("consumerCount=" + consumerCount);
+ System.out.println("subjectCount=" + subjectCount);
+ System.out.println("messageCount=" + messageCount);
+ System.out.println("");
+ }
+}
Propchange: incubator/activemq/trunk/jmeter/src/test/java/org/activemq/usecases/SystemTestSupport.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/pom.xml
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean-management/pom.xml
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean-management/src/java/org/activemq/gbean/ActiveMQBroker.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean-management/src/java/org/activemq/gbean/ActiveMQConnector.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean-management/src/java/org/activemq/gbean/ActiveMQManager.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/pom.xml
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/ActiveMQConnectorGBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/ActiveMQContainer.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/ActiveMQContainerGBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/management/ActiveMQManagerGBean.java
URL: http://svn.apache.org/viewcvs/incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/management/ActiveMQManagerGBean.java?rev=379619&r1=379618&r2=379619&view=diff
==============================================================================
--- incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/management/ActiveMQManagerGBean.java (original)
+++ incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/management/ActiveMQManagerGBean.java Tue Feb 21 15:12:56 2006
@@ -1 +1,283 @@
-/**
*
* Copyright 2004 Protique Ltd
*
* Licensed 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.activemq.gbean.management;
import java.util.Set;
import java.util.Iterator;
import java.util.ArrayList;
import java.util.List;
import java.util.Hashtable;
import javax.management.ObjectName;
import javax.management.MalformedObjectNameException;
import org.activemq.gbean.ActiveMQManager;
import org.activemq.gbean.ActiveMQBroker;
import org.activemq.gbean.ActiveMQConnector;
i
mport org.activemq.gbean.ActiveMQConnectorGBean;
import org.apache.geronimo.gbean.GBeanInfo;
import org.apache.geronimo.gbean.GBeanInfoBuilder;
import org.apache.geronimo.gbean.GBeanQuery;
import org.apache.geronimo.gbean.GBeanData;
import org.apache.geronimo.kernel.Kernel;
import org.apache.geronimo.kernel.GBeanNotFoundException;
import org.apache.geronimo.j2ee.management.impl.Util;
import org.apache.geronimo.j2ee.j2eeobjectnames.NameFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Implementation of the ActiveMQ management interface. These are the ActiveMQ
* mangement features available at runtime.
*
* @version $Revision: 1.0$
*/
public class ActiveMQManagerGBean implements ActiveMQManager {
private static final Log log = LogFactory.getLog(ActiveMQManagerGBean.class.getName());
private Kernel kernel;
public ActiveMQManagerGBean(Kernel kernel) {
this.kernel = kernel;
}
public String[] get
Containers() {
GBeanQuery query = new GBeanQuery(null, ActiveMQBroker.class.getName());
Set set = kernel.listGBeans(query);
String[] results = new String[set.size()];
int i=0;
for (Iterator it = set.iterator(); it.hasNext();) {
ObjectName name = (ObjectName) it.next();
results[i++] = name.getCanonicalName();
}
return results;
}
public String[] getSupportedProtocols() {
// see files in modules/core/src/conf/META-INF/services/org/activemq/transport/server/
return new String[]{"activeio","jabber","multicast","openwire","peer","stomp","tcp","udp","vm",};
}
public String[] getConnectors() {
GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
Set set = kernel.listGBeans(query);
String[] results = new String[set.size()];
int i=0;
for (Iterator it = set.iterator(); it.hasNext();) {
ObjectName nam
e = (ObjectName) it.next();
results[i++] = name.getCanonicalName();
}
return results;
}
public String[] getConnectors(String protocol) {
if(protocol == null) {
return getConnectors();
}
GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
Set set = kernel.listGBeans(query);
List results = new ArrayList();
for (Iterator it = set.iterator(); it.hasNext();) {
ObjectName name = (ObjectName) it.next();
try {
String target = (String) kernel.getAttribute(name, "protocol");
if(target != null && target.equals(protocol)) {
results.add(name.getCanonicalName());
}
} catch (Exception e) {
log.error("Unable to look up protocol for connector '"+name+"'",e);
}
}
return (String[]) results.toArray(new String[results.size()]);
}
public String[] getConnectorsForContainer(String broker) {
try {
ObjectName brokerName = ObjectName.getInstance(broker);
List results = new ArrayList();
GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
Set set = kernel.listGBeans(query); // all ActiveMQ connectors
for (Iterator it = set.iterator(); it.hasNext();) {
ObjectName name = (ObjectName) it.next(); // a single ActiveMQ connector
GBeanData data = kernel.getGBeanData(name);
Set refs = data.getReferencePatterns("activeMQContainer");
for (Iterator refit = refs.iterator(); refit.hasNext();) {
ObjectName ref = (ObjectName) refit.next();
if(ref.isPattern()) {
Set matches = kernel.listGBeans(ref);
if(matches.size() != 1) {
log.error("Unable to com
pare a connector->container reference that's a pattern to a fixed container name: "+ref.getCanonicalName());
} else {
ref = (ObjectName)matches.iterator().next();
if(ref.equals(brokerName)) {
results.add(name.getCanonicalName());
break;
}
}
} else {
if(ref.equals(brokerName)) {
results.add(name.getCanonicalName());
break;
}
}
}
}
return (String[]) results.toArray(new String[results.size()]);
} catch (Exception e) {
throw new IllegalArgumentException("Unable to look up connectors for broker '"+broker+"': "+e);
}
}
public String[] getConnectorsForContainer(String brok
er, String protocol) {
if(protocol == null) {
return getConnectorsForContainer(broker);
}
try {
ObjectName brokerName = ObjectName.getInstance(broker);
List results = new ArrayList();
GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
Set set = kernel.listGBeans(query); // all ActiveMQ connectors
for (Iterator it = set.iterator(); it.hasNext();) {
ObjectName name = (ObjectName) it.next(); // a single ActiveMQ connector
GBeanData data = kernel.getGBeanData(name);
Set refs = data.getReferencePatterns("activeMQContainer");
for (Iterator refit = refs.iterator(); refit.hasNext();) {
ObjectName ref = (ObjectName) refit.next();
boolean match = false;
if(ref.isPattern()) {
Set matches = kernel.listGBeans(ref);
if(matches.size() != 1) {
log.error("Unable to compare a connector->container reference that's a pattern to a fixed container name: "+ref.getCanonicalName());
} else {
ref = (ObjectName)matches.iterator().next();
if(ref.equals(brokerName)) {
match = true;
}
}
} else {
if(ref.equals(brokerName)) {
match = true;
}
}
if(match) {
try {
String testProtocol = (String) kernel.getAttribute(name, "protocol");
if(testProtocol != null && testProtocol.equals(protocol)) {
results.add(name.getCanonicalName());
}
} catch (Exception e) {
log.error("Unable to look up protocol for connector '"+name+"'",e);
}
break;
}
}
}
return (String[]) results.toArray(new String[results.size()]);
} catch (Exception e) {
throw new IllegalArgumentException("Unable to look up connectors for broker '"+broker+"': "+e);
}
}
/**
* Creates a new connector, and returns the ObjectName for it. Note that
* the connector may well require further customization before being fully
* functional (e.g. SSL settings for a secure connector).
*/
public String addConnector(String broker, String uniqueName, String protocol, String host, int port) {
ObjectName brokerName = null;
try {
brokerName = ObjectName.getInstance(broker);
} catch (MalformedObjectNameException e) {
throw new IllegalArgumentException("Unable to parse ObjectName '"+broker+"'");
}
ObjectName name = getConnectorName(brokerName, protocol, host, port, uniqueName);
GBeanData connector = new GBeanData(name, ActiveMQConnectorGBean.GBEAN_INFO);
//todo: if SSL is supported, need to add more properties or use a different GBean?
connector.setAttribute("protocol", protocol);
connector.setAttribute("host", host);
connector.setAttribute("port", new Integer(port));
connector.setReferencePattern("activeMQContainer", brokerName);
ObjectName config = Util.getConfiguration(kernel, brokerName);
try {
kernel.invoke(config, "addGBean", new Object[]{connector, Boolean.FALSE}, new String[]{GBeanData.class.getName(), boolean.class.getName()});
} catch (Exception e) {
log.error("Unable to add GBean ", e);
return null;
}
return name.getCanonicalName
();
}
public void removeConnector(String objectName) {
ObjectName name = null;
try {
name = ObjectName.getInstance(objectName);
} catch (MalformedObjectNameException e) {
throw new IllegalArgumentException("Invalid object name '" + objectName + "': " + e.getMessage());
}
try {
GBeanInfo info = kernel.getGBeanInfo(name);
boolean found = false;
Set intfs = info.getInterfaces();
for (Iterator it = intfs.iterator(); it.hasNext();) {
String intf = (String) it.next();
if (intf.equals(ActiveMQConnector.class.getName())) {
found = true;
}
}
if (!found) {
throw new GBeanNotFoundException(name);
}
ObjectName config = Util.getConfiguration(kernel, name);
kernel.invoke(config, "removeGBean", new Object[]{name}, new String[]{Ob
jectName.class.getName()});
} catch (GBeanNotFoundException e) {
log.warn("No such GBean '" + objectName + "'"); //todo: what if we want to remove a failed GBean?
} catch (Exception e) {
log.error("Unable to remove GBean", e);
}
}
/**
* Generate an ObjectName for a new connector GBean
*/
private ObjectName getConnectorName(ObjectName broker, String protocol, String host, int port, String uniqueName) {
Hashtable table = new Hashtable();
table.put(NameFactory.J2EE_APPLICATION, broker.getKeyProperty(NameFactory.J2EE_APPLICATION));
table.put(NameFactory.J2EE_SERVER, broker.getKeyProperty(NameFactory.J2EE_SERVER));
table.put(NameFactory.J2EE_MODULE, broker.getKeyProperty(NameFactory.J2EE_MODULE));
table.put(NameFactory.J2EE_TYPE, ActiveMQConnector.CONNECTOR_J2EE_TYPE);
String brokerName = broker.getKeyProperty(NameFactory.J2EE_NAME);
table.put("broker", b
rokerName);
table.put(NameFactory.J2EE_NAME, brokerName+"."+protocol+"."+host+(port > -1 ? "."+port : "")+"-"+uniqueName);
try {
return ObjectName.getInstance(broker.getDomain(), table);
} catch (MalformedObjectNameException e) {
throw new IllegalStateException("Never should have failed: " + e.getMessage());
}
}
public static final GBeanInfo GBEAN_INFO;
static {
GBeanInfoBuilder infoFactory = new GBeanInfoBuilder("ActiveMQ Manager", ActiveMQManagerGBean.class);
infoFactory.addAttribute("kernel", Kernel.class, false);
infoFactory.addInterface(ActiveMQManager.class);
infoFactory.setConstructor(new String[]{"kernel"});
GBEAN_INFO = infoFactory.getBeanInfo();
}
public static GBeanInfo getGBeanInfo() {
return GBEAN_INFO;
}
}
\ No newline at end of file
+/**
+ *
+ * Copyright 2004 Protique Ltd
+ *
+ * Licensed 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.activemq.gbean.management;
+
+import java.util.Set;
+import java.util.Iterator;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Hashtable;
+import javax.management.ObjectName;
+import javax.management.MalformedObjectNameException;
+import org.activemq.gbean.ActiveMQManager;
+import org.activemq.gbean.ActiveMQBroker;
+import org.activemq.gbean.ActiveMQConnector;
+import org.activemq.gbean.ActiveMQConnectorGBean;
+import org.apache.geronimo.gbean.GBeanInfo;
+import org.apache.geronimo.gbean.GBeanInfoBuilder;
+import org.apache.geronimo.gbean.GBeanQuery;
+import org.apache.geronimo.gbean.GBeanData;
+import org.apache.geronimo.kernel.Kernel;
+import org.apache.geronimo.kernel.GBeanNotFoundException;
+import org.apache.geronimo.j2ee.management.impl.Util;
+import org.apache.geronimo.j2ee.j2eeobjectnames.NameFactory;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Implementation of the ActiveMQ management interface. These are the ActiveMQ
+ * mangement features available at runtime.
+ *
+ * @version $Revision: 1.0$
+ */
+public class ActiveMQManagerGBean implements ActiveMQManager {
+ private static final Log log = LogFactory.getLog(ActiveMQManagerGBean.class.getName());
+ private Kernel kernel;
+
+ public ActiveMQManagerGBean(Kernel kernel) {
+ this.kernel = kernel;
+ }
+
+ public String[] getContainers() {
+ GBeanQuery query = new GBeanQuery(null, ActiveMQBroker.class.getName());
+ Set set = kernel.listGBeans(query);
+ String[] results = new String[set.size()];
+ int i=0;
+ for (Iterator it = set.iterator(); it.hasNext();) {
+ ObjectName name = (ObjectName) it.next();
+ results[i++] = name.getCanonicalName();
+ }
+ return results;
+ }
+
+ public String[] getSupportedProtocols() {
+ // see files in modules/core/src/conf/META-INF/services/org/activemq/transport/server/
+ return new String[]{"activeio","jabber","multicast","openwire","peer","stomp","tcp","udp","vm",};
+ }
+
+ public String[] getConnectors() {
+ GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
+ Set set = kernel.listGBeans(query);
+ String[] results = new String[set.size()];
+ int i=0;
+ for (Iterator it = set.iterator(); it.hasNext();) {
+ ObjectName name = (ObjectName) it.next();
+ results[i++] = name.getCanonicalName();
+ }
+ return results;
+ }
+
+ public String[] getConnectors(String protocol) {
+ if(protocol == null) {
+ return getConnectors();
+ }
+ GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
+ Set set = kernel.listGBeans(query);
+ List results = new ArrayList();
+ for (Iterator it = set.iterator(); it.hasNext();) {
+ ObjectName name = (ObjectName) it.next();
+ try {
+ String target = (String) kernel.getAttribute(name, "protocol");
+ if(target != null && target.equals(protocol)) {
+ results.add(name.getCanonicalName());
+ }
+ } catch (Exception e) {
+ log.error("Unable to look up protocol for connector '"+name+"'",e);
+ }
+ }
+ return (String[]) results.toArray(new String[results.size()]);
+ }
+
+ public String[] getConnectorsForContainer(String broker) {
+ try {
+ ObjectName brokerName = ObjectName.getInstance(broker);
+ List results = new ArrayList();
+ GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
+ Set set = kernel.listGBeans(query); // all ActiveMQ connectors
+ for (Iterator it = set.iterator(); it.hasNext();) {
+ ObjectName name = (ObjectName) it.next(); // a single ActiveMQ connector
+ GBeanData data = kernel.getGBeanData(name);
+ Set refs = data.getReferencePatterns("activeMQContainer");
+ for (Iterator refit = refs.iterator(); refit.hasNext();) {
+ ObjectName ref = (ObjectName) refit.next();
+ if(ref.isPattern()) {
+ Set matches = kernel.listGBeans(ref);
+ if(matches.size() != 1) {
+ log.error("Unable to compare a connector->container reference that's a pattern to a fixed container name: "+ref.getCanonicalName());
+ } else {
+ ref = (ObjectName)matches.iterator().next();
+ if(ref.equals(brokerName)) {
+ results.add(name.getCanonicalName());
+ break;
+ }
+ }
+ } else {
+ if(ref.equals(brokerName)) {
+ results.add(name.getCanonicalName());
+ break;
+ }
+ }
+ }
+ }
+ return (String[]) results.toArray(new String[results.size()]);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unable to look up connectors for broker '"+broker+"': "+e);
+ }
+ }
+
+ public String[] getConnectorsForContainer(String broker, String protocol) {
+ if(protocol == null) {
+ return getConnectorsForContainer(broker);
+ }
+ try {
+ ObjectName brokerName = ObjectName.getInstance(broker);
+ List results = new ArrayList();
+ GBeanQuery query = new GBeanQuery(null, ActiveMQConnector.class.getName());
+ Set set = kernel.listGBeans(query); // all ActiveMQ connectors
+ for (Iterator it = set.iterator(); it.hasNext();) {
+ ObjectName name = (ObjectName) it.next(); // a single ActiveMQ connector
+ GBeanData data = kernel.getGBeanData(name);
+ Set refs = data.getReferencePatterns("activeMQContainer");
+ for (Iterator refit = refs.iterator(); refit.hasNext();) {
+ ObjectName ref = (ObjectName) refit.next();
+ boolean match = false;
+ if(ref.isPattern()) {
+ Set matches = kernel.listGBeans(ref);
+ if(matches.size() != 1) {
+ log.error("Unable to compare a connector->container reference that's a pattern to a fixed container name: "+ref.getCanonicalName());
+ } else {
+ ref = (ObjectName)matches.iterator().next();
+ if(ref.equals(brokerName)) {
+ match = true;
+ }
+ }
+ } else {
+ if(ref.equals(brokerName)) {
+ match = true;
+ }
+ }
+ if(match) {
+ try {
+ String testProtocol = (String) kernel.getAttribute(name, "protocol");
+ if(testProtocol != null && testProtocol.equals(protocol)) {
+ results.add(name.getCanonicalName());
+ }
+ } catch (Exception e) {
+ log.error("Unable to look up protocol for connector '"+name+"'",e);
+ }
+ break;
+ }
+ }
+ }
+ return (String[]) results.toArray(new String[results.size()]);
+ } catch (Exception e) {
+ throw new IllegalArgumentException("Unable to look up connectors for broker '"+broker+"': "+e);
+ }
+ }
+
+ /**
+ * Creates a new connector, and returns the ObjectName for it. Note that
+ * the connector may well require further customization before being fully
+ * functional (e.g. SSL settings for a secure connector).
+ */
+ public String addConnector(String broker, String uniqueName, String protocol, String host, int port) {
+ ObjectName brokerName = null;
+ try {
+ brokerName = ObjectName.getInstance(broker);
+ } catch (MalformedObjectNameException e) {
+ throw new IllegalArgumentException("Unable to parse ObjectName '"+broker+"'");
+ }
+ ObjectName name = getConnectorName(brokerName, protocol, host, port, uniqueName);
+ GBeanData connector = new GBeanData(name, ActiveMQConnectorGBean.GBEAN_INFO);
+ //todo: if SSL is supported, need to add more properties or use a different GBean?
+ connector.setAttribute("protocol", protocol);
+ connector.setAttribute("host", host);
+ connector.setAttribute("port", new Integer(port));
+ connector.setReferencePattern("activeMQContainer", brokerName);
+ ObjectName config = Util.getConfiguration(kernel, brokerName);
+ try {
+ kernel.invoke(config, "addGBean", new Object[]{connector, Boolean.FALSE}, new String[]{GBeanData.class.getName(), boolean.class.getName()});
+ } catch (Exception e) {
+ log.error("Unable to add GBean ", e);
+ return null;
+ }
+ return name.getCanonicalName();
+ }
+
+ public void removeConnector(String objectName) {
+ ObjectName name = null;
+ try {
+ name = ObjectName.getInstance(objectName);
+ } catch (MalformedObjectNameException e) {
+ throw new IllegalArgumentException("Invalid object name '" + objectName + "': " + e.getMessage());
+ }
+ try {
+ GBeanInfo info = kernel.getGBeanInfo(name);
+ boolean found = false;
+ Set intfs = info.getInterfaces();
+ for (Iterator it = intfs.iterator(); it.hasNext();) {
+ String intf = (String) it.next();
+ if (intf.equals(ActiveMQConnector.class.getName())) {
+ found = true;
+ }
+ }
+ if (!found) {
+ throw new GBeanNotFoundException(name);
+ }
+ ObjectName config = Util.getConfiguration(kernel, name);
+ kernel.invoke(config, "removeGBean", new Object[]{name}, new String[]{ObjectName.class.getName()});
+ } catch (GBeanNotFoundException e) {
+ log.warn("No such GBean '" + objectName + "'"); //todo: what if we want to remove a failed GBean?
+ } catch (Exception e) {
+ log.error("Unable to remove GBean", e);
+ }
+ }
+
+ /**
+ * Generate an ObjectName for a new connector GBean
+ */
+ private ObjectName getConnectorName(ObjectName broker, String protocol, String host, int port, String uniqueName) {
+ Hashtable table = new Hashtable();
+ table.put(NameFactory.J2EE_APPLICATION, broker.getKeyProperty(NameFactory.J2EE_APPLICATION));
+ table.put(NameFactory.J2EE_SERVER, broker.getKeyProperty(NameFactory.J2EE_SERVER));
+ table.put(NameFactory.J2EE_MODULE, broker.getKeyProperty(NameFactory.J2EE_MODULE));
+ table.put(NameFactory.J2EE_TYPE, ActiveMQConnector.CONNECTOR_J2EE_TYPE);
+ String brokerName = broker.getKeyProperty(NameFactory.J2EE_NAME);
+ table.put("broker", brokerName);
+ table.put(NameFactory.J2EE_NAME, brokerName+"."+protocol+"."+host+(port > -1 ? "."+port : "")+"-"+uniqueName);
+ try {
+ return ObjectName.getInstance(broker.getDomain(), table);
+ } catch (MalformedObjectNameException e) {
+ throw new IllegalStateException("Never should have failed: " + e.getMessage());
+ }
+ }
+
+ public static final GBeanInfo GBEAN_INFO;
+
+ static {
+ GBeanInfoBuilder infoFactory = new GBeanInfoBuilder("ActiveMQ Manager", ActiveMQManagerGBean.class);
+ infoFactory.addAttribute("kernel", Kernel.class, false);
+ infoFactory.addInterface(ActiveMQManager.class);
+ infoFactory.setConstructor(new String[]{"kernel"});
+ GBEAN_INFO = infoFactory.getBeanInfo();
+ }
+
+ public static GBeanInfo getGBeanInfo() {
+ return GBEAN_INFO;
+ }
+}
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/gbean/management/ActiveMQManagerGBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/store/cache/SimpleCachePersistenceAdapterGBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/store/jdbc/JDBCPersistenceAdapterGBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/store/jdbc/ResourceManager.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/java/org/activemq/store/journal/JournalPersistenceAdapterGBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/activemq-gbean/src/test/org/activemq/gbean/ConnectorTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jabber/src/java/org/activemq/transport/jabber/JabberWireFormat.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jabber/src/test/org/activemq/transport/jabber/JabberWireFormatTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/java/org/activemq/transport/jrms/JRMSTransportChannel.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/java/org/activemq/transport/jrms/JRMSTransportChannelFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/java/org/activemq/transport/jrms/JRMSTransportServerChannel.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/java/org/activemq/transport/jrms/JRMSTransportServerChannelFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/test/org/activemq/transport/jrms/JRMSQueueSendReceiveTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/test/org/activemq/transport/jrms/JRMSQueueSendReceiveTwoConnectionsTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/test/org/activemq/transport/jrms/JRMSTopicSendReceiveTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/test/org/activemq/transport/jrms/JRMSTopicSendReceiveTwoConnectionsTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/test/org/activemq/transport/jrms/JRMSTransportChannelTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jrms/src/test/org/activemq/transport/jrms/JRMSTwoBrokerTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jxta/src/java/org/activemq/transport/jxta/JxtaTransportChannel.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jxta/src/java/org/activemq/transport/jxta/JxtaTransportChannelFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jxta/src/java/org/activemq/transport/jxta/JxtaTransportServerChannel.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/sandbox/jxta/src/java/org/activemq/transport/jxta/JxtaTransportServerChannelFactory.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/systest/itests/client/src/itest/org/activemq/itests/ResourceAdapterTest.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/systest/itests/ejb/src/java/org/activemq/itest/ejb/JMSTool.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/systest/itests/ejb/src/java/org/activemq/itest/ejb/JMSToolBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/systest/itests/ejb/src/java/org/activemq/itest/ejb/JMSToolHome.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/activemq/trunk/systest/itests/ejb/src/java/org/activemq/itest/ejb/MDBTransferBean.java
------------------------------------------------------------------------------
svn:eol-style = native