On 5/16/07, Jiang <[EMAIL PROTECTED]> wrote:
bsnyder thanks a lot.
In fact I just want to admin ActiveMQ from remote admin tools(which written
by myself). I want to add/remove destinations on that MQ server . From my
opinion, I think I should first get a referece of the remote server ,then I
can do some operations on the server through that ref ( What I do is not
create the broker really , In fact it has been created and run on the server
when Activemq server starts.I just use it .Maybe what I think is wrong). So
I connect to the broker at first step use the original code BrokerService
service = BrokerFactory.createBroker(new URI("xxxxx")); or maybe the code
you write for example , then I will use the code below:
ActiveMQDestination[] destinations = service.getDestinations(); //to
get destinations list
BrokerView brview = service.getAdminView(); //get the admin
BrokerView
brview.addTopic("Hello"); //add Hello topic to the broker --- Step 1
brview.addTopic("World"); //add World topic to the broker---Step 2
brview.removeTopic("Hello"); //remove Hello topic from the
broker---Step3
brview.removeTopic("World"); //remove World topic from the
broker---Step 4
All the add/remove actions can been seen from the jconsole (after Step1 , we
can seen from jconsole that Hello Topic has been added. after Step2 we can
seen World topic has been added. after Step3 Hello topic can been seen
removed. after Step4 just as Step3 )
bsnyder what I say is right ? (Now the code I wirte above has not been
tested for lack of environment. I will test it late )
OK, I see what you need now. You need to use the methods exposed via
JMX to achieve this task. I'm attaching a file that is an example of
exactly this. Let me know if you have any questions.
Bruce
--
perl -e 'print unpack("u30","D0G)[EMAIL
PROTECTED]&5R\"F)R=6-E+G-N>61E<D\!G;6%I;\"YC;VT*"
);'
Apache Geronimo - http://geronimo.apache.org/
Apache ActiveMQ - http://activemq.org/
Apache ServiceMix - http://servicemix.org/
Castor - http://castor.org/
package com.logicblaze.fuse;
import java.io.IOException;
import java.net.MalformedURLException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.management.AttributeNotFoundException;
import javax.management.InstanceNotFoundException;
import javax.management.IntrospectionException;
import javax.management.MBeanAttributeInfo;
import javax.management.MBeanException;
import javax.management.MBeanInfo;
import javax.management.MBeanServerConnection;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectInstance;
import javax.management.ObjectName;
import javax.management.Query;
import javax.management.QueryExp;
import javax.management.ReflectionException;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
/**
* Programmatic JMX example
*
* Instructions for running this code:
* <ol type="1">
* <li>Start up FUSE</li>
* <li>Execute the build for this test using the following command on the
command line:
* <tt>$ mvn install </tt> </li>
* </ol>
*/
public class AMQJmxExampleTest extends TestCase {
private static final Log log = LogFactory.getLog(AMQJmxExampleTest.class);
public String amqJmxUrl =
"service:jmx:rmi:///jndi/rmi://localhost:1099/jmxrmi";
/**
* Create the test case
*
* @param testName name of the test case
*/
public AMQJmxExampleTest(String testName) {
super(testName);
}
/**
* @return the suite of tests being tested
*/
public static Test suite() {
return new TestSuite(JmxExampleTest.class);
}
public void testApp() {
MBeanServerConnection connection;
String clientServiceName =
"org.apache.activemq:ContainerName=ServiceMix,Type=SystemService,Name=ClientFactory";
String topicName = "Boulder.Colorado";
try {
// Acquire a connection to the MBean server
connection = connect();
// How many MBeans are running?
count(connection);
// Query for a single MBean
query(connection, clientServiceName);
// Query all MBeans
query(connection, "");
// Check for the topic first
Set mbeans = queryForTopic(connection, topicName);
log.info("Located [" + mbeans.size() + "] MBeans");
if(mbeans.size() > 0) {
// Create a new topic on the broker
createTopic(connection, topicName);
}
else {
// Remove the topic from the broker and then create it
removeTopic(connection, topicName);
createTopic(connection, topicName);
}
mbeans = queryForTopic(connection, topicName);
log.info("Located [" + mbeans.size() + "] MBeans");
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public MBeanServerConnection connect()
throws IOException {
JMXConnector connector = null;
MBeanServerConnection connection = null;
String username = "";
String password = "";
Map env = new HashMap();
String[] credentials = new String[] { username, password };
env.put(JMXConnector.CREDENTIALS, credentials);
try {
connector = JMXConnectorFactory.newJMXConnector(new
JMXServiceURL(amqJmxUrl), env);
connector.connect();
connection = connector.getMBeanServerConnection();
} catch (MalformedURLException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (IOException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return connection;
}
public void count(MBeanServerConnection conn)
throws IOException {
int numberOfMBeans = conn.getMBeanCount().intValue();
log.info("Number of MBeans currently running: " + numberOfMBeans);
}
public void query(MBeanServerConnection conn, String query)
throws IOException {
if (conn != null && query != null) {
listMBeans(conn, query);
} else if (conn != null && query.equals("")) {
listAllMBeanNames(conn);
} else {
log.fatal("Unable to connect to ServiceMix");
}
}
public void listMBeans(MBeanServerConnection conn, String query)
throws IOException {
ObjectName name;
Set names = null;
try {
name = new ObjectName(query);
names = conn.queryMBeans(name, name);
} catch (MalformedObjectNameException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NullPointerException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
for(Iterator iter = names.iterator(); iter.hasNext(); ) {
ObjectInstance obj = (ObjectInstance) iter.next();
log.info("+ " + obj.getClassName());
}
}
public void listAllMBeanNames(MBeanServerConnection conn)
throws IOException {
Set names = getAllMBeanNames(conn);
for (Iterator iter = names.iterator(); iter.hasNext();) {
ObjectName objName = (ObjectName) iter.next();
log.info("+ " + objName);
}
}
public void listMBeanAttrs(MBeanServerConnection conn, String query)
throws IOException {
ObjectName objName = null;
try {
objName = new ObjectName(query);
log.info("+ " + objName.getCanonicalName());
MBeanInfo info = getMBeanInfo(conn, objName);
MBeanAttributeInfo[] attrs = info.getAttributes();
for(int i = 0; i < attrs.length; ++i) {
Object obj = conn.getAttribute(objName, attrs[i].getName());
log.info(" - " + attrs[i].getName() + obj);
}
} catch (MalformedObjectNameException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NullPointerException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (AttributeNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InstanceNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MBeanException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ReflectionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public void createTopic(MBeanServerConnection conn, String topicName)
throws IOException {
String brokerNameQuery =
"org.apache.activemq:BrokerName=localhost,Type=Broker";
String addTopicOperationName = "addTopic";
Object[] params = { topicName };
String[] sig = { "java.lang.String" };
doTopicCrud(conn, topicName, brokerNameQuery, addTopicOperationName,
params, sig, "creat");
}
private void removeTopic(MBeanServerConnection conn, String topicName)
throws IOException {
String brokerNameQuery =
"org.apache.activemq:BrokerName=localhost,Type=Broker";
String removeTopicOperationName = "removeTopic";
Object[] params = { topicName };
String[] sig = { "java.lang.String" };
doTopicCrud(conn, topicName, brokerNameQuery, removeTopicOperationName,
params, sig, "remov");
}
private void doTopicCrud(MBeanServerConnection conn, String topicName,
String brokerNameQuery, String operationName, Object[] params,
String[] sig, String verb)
throws IOException {
ObjectName brokerObjName;
try {
log.info( verb + "ing new topic: [" + topicName + "]");
brokerObjName = new ObjectName(brokerNameQuery);
conn.invoke(brokerObjName, operationName, params, sig);
log.info("Topic [" + topicName + "] has been " + verb + "ed");
} catch (MalformedObjectNameException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NullPointerException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InstanceNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MBeanException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ReflectionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
private void printMBeans(String topicName, Set mbeans) {
if (!mbeans.isEmpty()) {
for (Iterator iter = mbeans.iterator(); iter.hasNext();) {
ObjectInstance mbean = (ObjectInstance) iter.next();
log.info("+ " + mbean.getClassName());
}
}
else {
log.info("Unable to locate MBean for " + topicName);
}
}
public Set queryForTopic(MBeanServerConnection conn, String topicName)
throws IOException {
// Was the topic created?
String topicsQuery =
"org.apache.activemq:BrokerName=localhost,Type=Topic,*";
// listMBeans(conn, topicsQuery);
// Use JMX query expressions
QueryExp queryExp = Query.eq(Query.attr("name"),
Query.value(topicName));
ObjectName objName;
Set mbeans = null;
try {
objName = new ObjectName(topicsQuery);
log.info("Querying for " + topicName);
mbeans = conn.queryMBeans(objName, queryExp);
} catch (MalformedObjectNameException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (NullPointerException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
return mbeans;
}
public void listAllMBeanAttrs(MBeanServerConnection conn, Set names)
throws IOException {
for (Iterator iter = names.iterator(); iter.hasNext();) {
ObjectName objName = (ObjectName) iter.next();
log.info("+ " + objName);
MBeanInfo info = getMBeanInfo(conn, objName);
MBeanAttributeInfo[] attrs = info.getAttributes();
if (attrs == null)
continue;
for (int i = 0; i < attrs.length; ++i) {
try {
Object obj = conn.getAttribute(objName, attrs[i].getName());
log.info(" - " + attrs[i].getName() + " = " + obj);
} catch (NullPointerException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (AttributeNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InstanceNotFoundException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (MBeanException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ReflectionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
public void query(MBeanServerConnection conn, QueryExp queryExp)
throws IOException {
log.info("Not yet implemented");
}
// Private ----------------------------------------------------------------
private MBeanInfo getMBeanInfo(MBeanServerConnection conn, ObjectName
objName)
throws IOException {
MBeanInfo info = null;
try {
info = conn.getMBeanInfo((ObjectName) objName);
} catch (InstanceNotFoundException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} catch (IntrospectionException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
} catch (ReflectionException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
return info;
}
private Set getAllMBeanNames(MBeanServerConnection conn)
throws IOException {
return conn.queryNames(null, null);
}
}
