svn commit: r1081398 - /uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/examples/as/GetMetaRequest.java

[ea...@apache.org]

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;
+  }
+
+}