svn commit: r927894 - /incubator/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/aae/jmx/RemoteJMXServer.java

[cw...@apache.org]

Author: cwiklik
Date: Fri Mar 26 15:00:28 2010
New Revision: 927894

URL: http://svn.apache.org/viewvc?rev=927894&view=rev
Log:
UIMA-1726 This class wraps remote JMX MbeanServer. It enables lookups of objects registered with remote MBeanServer 

Added:
    incubator/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/aae/jmx/RemoteJMXServer.java

Added: incubator/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/aae/jmx/RemoteJMXServer.java
URL: http://svn.apache.org/viewvc/incubator/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/aae/jmx/RemoteJMXServer.java?rev=927894&view=auto
==============================================================================
--- incubator/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/aae/jmx/RemoteJMXServer.java (added)
+++ incubator/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/aae/jmx/RemoteJMXServer.java Fri Mar 26 15:00:28 2010
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.uima.aae.jmx;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+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.broker.jmx.QueueViewMBean;
+import org.apache.uima.UIMAFramework;
+import org.apache.uima.adapter.jms.JmsConstants;
+import org.apache.uima.util.Level;
+
+public class RemoteJMXServer {
+  private static final Class CLASS_NAME = RemoteJMXServer.class;
+
+  private MBeanServerConnection brokerMBeanServer = null;
+
+  private ConcurrentHashMap<String, QueueViewMBean> queueMBeanMap = new ConcurrentHashMap<String, QueueViewMBean>();
+
+  private String brokerName;
+
+  private JMXConnector jmxc = null;
+
+  private volatile boolean initialized = false;
+
+  public boolean isInitialized() {
+    return initialized;
+  }
+
+  /**
+   * 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
+   */
+  public 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";
+
+    JMXServiceURL url = new JMXServiceURL(remoteJmxUrl);
+    jmxc = JMXConnectorFactory.connect(url, null);
+    brokerMBeanServer = jmxc.getMBeanServerConnection();
+    // 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
+      }
+    }
+  }
+  /**
+   * Disconnects from MBeanServer
+   */
+  public void disconnect() {
+    if (jmxc != null) {
+      try {
+        jmxc.close();
+        brokerMBeanServer = null;
+        queueMBeanMap.clear();
+      } catch (IOException e) {
+      }
+    }
+  }
+
+  /**
+   * Tries to fetch total number of MBeans in the MBeanServer. The real goal here is to check if the
+   * server responds. Failure here indicates failed server connection.
+   * 
+   * @return
+   */
+  public boolean isServerAvailable() {
+    try {
+      brokerMBeanServer.getMBeanCount();
+      return true;
+    } catch (Exception e) {
+      return false;
+    }
+  }
+
+  public void attachToTempQueue(String tempQueueName) throws Exception {
+    if (!queueMBeanMap.containsKey(tempQueueName)) {
+      ObjectName uimaServiceTempReplyQueuePattern = composeObjectName("TempQueue", tempQueueName);
+      QueueViewMBean replyQueueMBean = getQueueMBean(tempQueueName,
+              uimaServiceTempReplyQueuePattern);
+      if (replyQueueMBean != null) {
+        queueMBeanMap.put(tempQueueName, replyQueueMBean);
+      }
+    }
+  }
+
+  private ObjectName composeObjectName(String queueType, String queueName) throws Exception {
+
+    ObjectName n = new ObjectName(brokerName + ",Type=" + queueType + ",Destination=" + queueName);
+    return n;
+  }
+
+  /**
+   * Checks if a given queue name exists in remote MBeanServer's registry.
+   * NOTE: The code returns true in case the MBeanServer is not available.
+   * 
+   * @param queueName
+   *          - queue to lookup in the MBeanServer
+   * @return - true if queue exists, false otherwise
+   */
+  public boolean isClientReplyQueueAvailable(String queueName) {
+    try {
+      ObjectName uimaServiceTempReplyQueuePattern = composeObjectName("TempQueue", 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, uimaServiceTempReplyQueuePattern) == null) {
+        return false;
+      }
+    } catch (Exception e) {
+      UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
+              "isClientReplyQueueAvailable", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
+              "UIMAJMS_exception__WARNING", e);    
+    }
+    return true;  // returns true, in case the MBeanServer is not available or the queue exists
+  }
+
+  /**
+   * Creates proxy to a queue MBean using given match pattern
+   * 
+   * @param key
+   * @param matchPattern
+   * @return
+   * @throws Exception
+   */
+  private 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;
+  }
+
+  /**
+   * Replaces ':' with '_'. JMX queries containing ':' are illegal.
+   * 
+   * @param destinationName
+   * @return
+   */
+  public String normalize(String destinationName) {
+    String qN = destinationName.substring(destinationName.indexOf("ID"));
+    return qN.replaceAll(":", "_");
+  }
+
+  public static void main(String[] args) {
+    try {
+      RemoteJMXServer broker = new RemoteJMXServer();
+      broker.initialize("org.apache.activemq", args[0], args[1]);
+      if (broker.isClientReplyQueueAvailable(args[2])) {
+        System.out.println("TempQueue:" + args[2] + " Exists");
+      } else {
+        System.out.println("TempQueue:" + args[2] + " Does not Exist");
+      }
+    } catch (Exception e) {
+      e.printStackTrace();
+    }
+  }
+}