Author: eae
Date: Mon Mar 14 14:35:44 2011
New Revision: 1081398
URL: http://svn.apache.org/viewvc?rev=1081398&view=rev
Log:
UIMA-2089
Added:
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/GetMetaRequest.java
Added:
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/GetMetaRequest.java
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/GetMetaRequest.java?rev=1081398&view=auto
==============================================================================
---
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/GetMetaRequest.java
(added)
+++
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/GetMetaRequest.java
Mon Mar 14 14:35:44 2011
@@ -0,0 +1,241 @@
+/**
+ *
+ */
+package org.apache.uima.examples.as;
+
+import java.net.URI;
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jms.Connection;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.TemporaryQueue;
+import javax.jms.TextMessage;
+import javax.management.MBeanServerConnection;
+import javax.management.MBeanServerInvocationHandler;
+import javax.management.ObjectName;
+import javax.management.QueryExp;
+import javax.management.remote.JMXConnector;
+import javax.management.remote.JMXConnectorFactory;
+import javax.management.remote.JMXServiceURL;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.jmx.QueueViewMBean;
+import org.apache.activemq.command.ActiveMQTextMessage;
+import org.apache.uima.aae.message.AsynchAEMessage;
+import org.apache.uima.aae.message.UIMAMessage;
+import org.springframework.jmx.JmxException;
+
+/**
+ *
+ */
+public class GetMetaRequest {
+ private static MBeanServerConnection brokerMBeanServer = null;
+ private static String brokerName;
+ private static JMXConnector jmxc = null;
+ private static boolean initialized = false;
+ private static enum QueueState {exists, existsnot, jmxnot};
+ private static String jmxPort;
+
+ public static void main(String[] args) {
+ if (args.length < 2) {
+ System.err.println("Need arguments: brokerURI serviceName [-verbose]");
+ System.exit(1);
+ }
+ String brokerURI = args[0];
+ String queueName = args[1];
+ boolean printReply = false;
+ if(args.length > 2) {
+ if (args[2].equalsIgnoreCase("-verbose")) {
+ printReply = true;
+ }
+ else {
+ System.err.println("Unknown argument: " + args[2]);
+ System.exit(1);
+ }
+ }
+ Connection connection = null;
+ Session producerSession = null;
+ Queue producerQueue = null;
+ MessageProducer producer = null;
+ MessageConsumer consumer = null;
+ Session consumerSession = null;
+ TemporaryQueue consumerDestination = null;
+ long startTime = 0;
+
+ // Check if JMX server port number was specified
+ jmxPort = System.getProperty("activemq.broker.jmx.port");
+ if ( jmxPort == null || jmxPort.trim().length() == 0 ) {
+ jmxPort = "1099"; // default
+ }
+
+ try {
+ URI target = new URI(brokerURI);
+ String brokerHost = target.getHost();
+
+ attachToRemoteBrokerJMXServer(brokerURI);
+ if (isQueueAvailable(queueName) == QueueState.exists) {
+ System.out.println("Queue "+queueName+" found on "+ brokerURI);
+ System.out.println("Sending getMeta...");
+ }
+ else if (isQueueAvailable(queueName) == QueueState.existsnot) {
+ System.err.println("Queue "+queueName+" does not exist on "+
brokerURI);
+ System.exit(1);
+ }
+ else {
+ System.out.println("Cannot see queues on JMX port
"+brokerHost+":"+jmxPort);
+ System.out.println("Sending getMeta anyway...");
+ }
+
+ ActiveMQConnectionFactory factory = new
ActiveMQConnectionFactory(brokerURI);
+ connection = factory.createConnection();
+ connection.start();
+
+ producerSession = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ producerQueue = producerSession.createQueue(queueName);
+ producer = producerSession.createProducer(producerQueue);
+ consumerSession = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
+ consumerDestination = consumerSession.createTemporaryQueue();
+ consumer = consumerSession.createConsumer(consumerDestination);
+
+ TextMessage msg = producerSession.createTextMessage();
+ msg.setStringProperty(AsynchAEMessage.MessageFrom,
consumerDestination.getQueueName());
+ msg.setStringProperty(UIMAMessage.ServerURI, brokerURI);
+ msg.setIntProperty(AsynchAEMessage.MessageType, AsynchAEMessage.Request);
+ msg.setIntProperty(AsynchAEMessage.Command, AsynchAEMessage.GetMeta);
+ msg.setJMSReplyTo(consumerDestination);
+ msg.setText("");
+ producer.send(msg);
+ startTime = System.nanoTime();
+
+ System.out.println("Sent getMeta request to " + queueName + " at " +
brokerURI);
+ System.out.println("Waiting for reply...");
+ ActiveMQTextMessage reply = (ActiveMQTextMessage) consumer.receive();
+ long waitTime = (System.nanoTime() - startTime)/1000000;
+ System.out.println("Reply received in " + waitTime + " ms");
+ if (printReply) {
+ System.out.println("Reply MessageText: " + reply.getText());
+ }
+ System.exit(0);
+ } catch (Exception e) {
+ System.err.println(e.toString());
+// e.printStackTrace();
+ }
+ }
+ /**
+ * Connects to this service Broker's JMX Server. If unable to connect, this
method
+ * fails silently. The method uses default JMX Port number 1099 to create a
connection
+ * to the Broker's JMX MBean server. The default can be overridden via
System
+ * property 'activemq.broker.jmx.port'. If connection cannot be established
the
+ * method silently fails.
+ *
+ */
+ private static void attachToRemoteBrokerJMXServer(String brokerUri) throws
Exception {
+ // Fetch AMQ jmx domain from system properties. This property is not
required
+ // and the default AMQ jmx is used. The only exception is when the
service is
+ // deployed in a jvm with multiple brokers deployed as it is the case
with jUnit
+ // tests. In such a case, each broker will register self with JMX using a
different
+ // domain.
+ String jmxAMQDomain = System.getProperty("activemq.broker.jmx.domain");
+ if ( jmxAMQDomain == null ) {
+ jmxAMQDomain = "org.apache.activemq"; // default
+ }
+ String brokerHostname="";
+ URI target = new URI(brokerUri);
+ brokerHostname = target.getHost();
+ initialize(jmxAMQDomain, brokerHostname,jmxPort);
+ }
+ /**
+ * Creates a connection to an MBean Server identified by
+ * <code>remoteJMXServerHostName and remoteJMXServerPort</code>
+ *
+ * @param remoteJMXServerHostName
+ * - MBeanServer host name
+ * @param remoteJMXServerPort
+ * - MBeanServer port
+ * @return - none
+ *
+ * @throws Exception
+ */
+ private static void initialize(String jmxDomain, String
remoteJMXServerHostname,
+ String remoteJMXServerPort) throws Exception {
+ // Construct connect string to the JMX MBeanServer
+ String remoteJmxUrl = "service:jmx:rmi:///jndi/rmi://" +
remoteJMXServerHostname + ":"
+ + remoteJMXServerPort + "/jmxrmi";
+
+ try {
+ JMXServiceURL url = new JMXServiceURL(remoteJmxUrl);
+ jmxc = JMXConnectorFactory.connect(url, null);
+ brokerMBeanServer = jmxc.getMBeanServerConnection();
+ // Its possible that the above code succeeds even though the broker runs
+ // with no JMX Connector. At least on windows the above does not throw an
+ // exception as expected. It appears that the broker registers self JVMs
MBeanServer
+ // but it does *not* register any Connections, nor Queues. The code
below
+ // checks if the MBean server we are connected to has any QueueMBeans
registered.
+ // A broker with jmx connector should have queue mbeans registered and
thus
+ // the code below should always succeed. Conversely, a broker with no
jmx connector
+ // reports no queue mbeans.
+
+ // Query broker MBeanServer for QueueMBeans
+ Set queueSet = brokerMBeanServer.queryNames(new ObjectName(jmxDomain
+ + ":*,Type=Queue"), (QueryExp) null);
+ if ( queueSet.isEmpty() ) { // No QueueMBeans, meaning no JMX support
+ throw new JmxException("ActiveMQ Broker Not Configured With JMX
Support");
+ }
+ } catch (Exception e) {
+ return;
+ }
+ // Query JMX Server for Broker MBean. We need the broker's name from an
MBean to construct
+ // queue query string.
+
+ for (Object nameObject : brokerMBeanServer.queryNames(new
ObjectName(jmxDomain
+ + ":*,Type=Broker"), (QueryExp) null)) {
+ ObjectName brokerObjectName = (ObjectName) nameObject;
+ if (brokerObjectName.getCanonicalName().endsWith("Type=Broker")) {
+ // Extract just the name from the canonical name
+ brokerName = brokerObjectName.getCanonicalName().substring(0,
+ brokerObjectName.getCanonicalName().indexOf(","));
+ initialized = true;
+ break; // got the broker name
+ }
+ }
+ }
+
+ private static boolean isServerAvailable() {
+ try {
+ brokerMBeanServer.getMBeanCount();
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private static QueueState isQueueAvailable(String queueName) throws
Exception {
+ if (!initialized) {
+ return QueueState.jmxnot;
+ }
+ ObjectName uimaServiceQueuePattern = new ObjectName(brokerName +
",Type=Queue,Destination=" + queueName);
+ // Tests if queue exists. If a client terminates, the reply queue will be
removed and we
+ // expect null from getQueueMBean()
+ if (isServerAvailable() && getQueueMBean(queueName,
uimaServiceQueuePattern) == null) {
+ return QueueState.existsnot;
+ }
+ return QueueState.exists;
+ }
+
+ private static QueueViewMBean getQueueMBean(String key, ObjectName
matchPattern) throws Exception {
+ // Fetch queue names matching a given pattern.
+ Set<ObjectName> queues = new
HashSet<ObjectName>(brokerMBeanServer.queryNames(matchPattern,
+ null));
+ for (ObjectName name : queues) {
+ // Create and return a proxy to the queue's MBean
+ return (QueueViewMBean)
MBeanServerInvocationHandler.newProxyInstance(brokerMBeanServer,
+ name, QueueViewMBean.class, true);
+ }
+ return null;
+ }
+
+}
