You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by ds...@apache.org on 2015/07/06 23:47:39 UTC
[14/18] incubator-geode git commit: GEODE-80: Imported databrowser
from geode-1.0.0-SNAPSHOT-2.src.tar
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/VersionMismatchException.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/VersionMismatchException.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/VersionMismatchException.java
new file mode 100644
index 0000000..985632c
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/VersionMismatchException.java
@@ -0,0 +1,35 @@
+/*=========================================================================
+ * (c)Copyright 2002-2009, GemStone Systems, Inc. All Rights Reserved.
+ * 1260 NW Waterhouse Ave., Suite 200, Beaverton, OR 97006
+ * All Rights Reserved.
+ * =======================================================================*/
+package com.gemstone.gemfire.mgmt.DataBrowser.connection;
+
+/**
+ * @author hgadre
+ *
+ */
+public class VersionMismatchException extends ConnectionFailureException {
+
+ /**
+ *
+ */
+ private static final long serialVersionUID = 1L;
+
+ public VersionMismatchException() {
+ super();
+ }
+
+ public VersionMismatchException(String arg0, Throwable arg1) {
+ super(arg0, arg1);
+ }
+
+ public VersionMismatchException(String arg0) {
+ super(arg0);
+ }
+
+ public VersionMismatchException(Throwable arg0) {
+ super(arg0);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/DummyMemberDiscoveryImpl.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/DummyMemberDiscoveryImpl.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/DummyMemberDiscoveryImpl.java
new file mode 100644
index 0000000..d330807
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/DummyMemberDiscoveryImpl.java
@@ -0,0 +1,107 @@
+/*=========================================================================
+ * (c)Copyright 2002-2009, GemStone Systems, Inc. All Rights Reserved.
+ * 1260 NW Waterhouse Ave., Suite 200, Beaverton, OR 97006
+ * All Rights Reserved.
+ * =======================================================================*/
+package com.gemstone.gemfire.mgmt.DataBrowser.connection.internal;
+
+import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.client.PoolFactory;
+import com.gemstone.gemfire.cache.client.PoolManager;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ClientConfiguration;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GFMemberDiscovery;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnectionListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireMemberListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.CacheServerInfo;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.GemFireMember;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberFactory;
+import com.gemstone.gemfire.mgmt.DataBrowser.utils.LogUtil;
+
+/**
+ * @author hgadre
+ *
+ */
+public class DummyMemberDiscoveryImpl implements GFMemberDiscovery {
+
+ private GemFireMember member;
+ private Pool pool;
+
+ public DummyMemberDiscoveryImpl(ClientConfiguration conf) {
+ member = MemberFactory.createGemFireMember(conf);
+ preparePoolConnection();
+ }
+
+ private void preparePoolConnection() {
+ if(this.pool != null) {
+ throw new IllegalStateException();
+ }
+
+ PoolFactory factory = PoolManager.createFactory();
+
+ if(this.member.getType() == GemFireMember.GEMFIRE_CACHE_SERVER) {
+ for(int i = 0 ; i < this.member.getCacheServers().length ; i++) {
+ CacheServerInfo info = this.member.getCacheServers()[i];
+ factory.addServer(info.getBindAddress(), info.getPort());
+ LogUtil.fine("Adding Cache Server :"+info);
+ }
+
+ } else {
+ for(int i = 0 ; i < this.member.getCacheServers().length ; i++) {
+ CacheServerInfo info = this.member.getCacheServers()[i];
+ factory.addLocator(info.getBindAddress(), info.getPort());
+ }
+ }
+
+ this.pool = factory.create("GFE_DATABROWSER_POOL");
+ }
+
+
+
+ public void addGemFireMemberListener(GemFireMemberListener listener) {
+
+ }
+
+ public void close() {
+ }
+
+ public GemFireMember getMember(String id) {
+ return member;
+ }
+
+ public GemFireMember[] getMembers() {
+ return new GemFireMember[] {member};
+ }
+
+ public QueryService getQueryService(GemFireMember mbr) {
+ // TODO - MGH: we will need to find the pool for this member ?
+ if(this.pool == null) {
+ throw new IllegalStateException("pool should not be null");
+ }
+
+ return this.pool.getQueryService();
+ }
+
+ public void removeGemFireMemberListener(GemFireMemberListener listener) {
+ }
+
+
+ public void addConnectionNotificationListener(GemFireConnectionListener listener) {
+ }
+
+ public void removeConnectionNotificationListener(GemFireConnectionListener listener) {
+ }
+
+ public String getGemFireSystemVersion() {
+ return "UNKNOWN VERSION";
+ }
+
+ public long getRefreshInterval() {
+ return -1;
+ }
+
+ public void setRefreshInterval(long time) {
+ // Do nothing.
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/GemFireClientConnectionImpl.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/GemFireClientConnectionImpl.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/GemFireClientConnectionImpl.java
new file mode 100644
index 0000000..2b03b1a
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/GemFireClientConnectionImpl.java
@@ -0,0 +1,293 @@
+/*=========================================================================
+ * (c)Copyright 2002-2011, GemStone Systems, Inc. All Rights Reserved.
+ * 1260 NW Waterhouse Ave., Suite 200, Beaverton, OR 97006
+ * All Rights Reserved.
+ * =======================================================================*/
+package com.gemstone.gemfire.mgmt.DataBrowser.connection.internal;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.gemstone.gemfire.cache.Cache;
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqListener;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.distributed.DistributedSystem;
+import com.gemstone.gemfire.internal.GemFireVersion;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GFMemberDiscovery;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnection;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnectionListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireMemberListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.GemFireMember;
+import com.gemstone.gemfire.mgmt.DataBrowser.prefs.DataBrowserPreferences;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.IntrospectionRepository;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.QueryExecutionException;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.QueryResult;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.QueryUtil;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQException;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQQuery;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQQueryImpl;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQResult;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.internal.PdxHelper;
+import com.gemstone.gemfire.mgmt.DataBrowser.utils.LogUtil;
+
+public class GemFireClientConnectionImpl implements GemFireConnection {
+
+ static final String GEMFIRE_LIMIT_AVL = "6.*";
+ private DistributedSystem system;
+ private Cache cache;
+ private GFMemberDiscovery discovery;
+ private Map<String, CQQuery> cqs;
+ private volatile boolean closed;
+ private boolean limitClauseAvl = false;
+
+ public GemFireClientConnectionImpl(DistributedSystem sys,
+ GFMemberDiscovery dscvry) {
+ if (sys == null) {
+ throw new IllegalArgumentException(
+ "A valid GemFire client connection is required.");
+ } else if (dscvry == null) {
+ throw new IllegalArgumentException(
+ "A valid member discovery mechanism must be provided.");
+ }
+
+ this.system = sys;
+ this.cache = createCache(this.system);
+ this.discovery = dscvry;
+ this.cqs = Collections.synchronizedMap(new HashMap<String, CQQuery>());
+ this.closed = false;
+
+ String version = discovery.getGemFireSystemVersion();
+
+ /**
+ * A check is added to verify if the current version of GemFire supports LIMIT clause.
+ * LIMIT clause is available from GemFire 6.0 onwards.
+ **/
+ if(null == version) {
+ this.limitClauseAvl = false;
+ } else {
+ this.limitClauseAvl = version.matches(GEMFIRE_LIMIT_AVL);
+ }
+ String gemFireJarVersion = GemFireVersion.getGemFireVersion();
+ LogUtil.info("Version of GemFire jar in classpath: "+gemFireJarVersion);
+ LogUtil.info("The GemFire system version is :"+version);
+ LogUtil.info("This version of GemFire system supports LIMIT clause :"+this.limitClauseAvl);
+
+ //See comment for IntrospectionRepository.removePdxIntrospectionResults.
+ //Done only if Pdx is supported
+ IntrospectionRepository repos = IntrospectionRepository.singleton();
+ if (PdxHelper.getInstance().isPdxSupported()) {
+ if (repos.removePdxIntrospectionResults()) {
+ LogUtil.fine("Cleared cached PDX Type information");
+ }
+ }
+ }
+
+ public void close() {
+ if (this.system != null) {
+ this.closed = true;
+
+ //Stop All CQs.
+ List<String> currCqs = new ArrayList<String>();
+ currCqs.addAll(this.cqs.keySet());
+ for(String name : currCqs) {
+ try {
+ closeCQ(name);
+ }
+ catch (CQException e) {
+ LogUtil.warning("Failed to close CQ with name :" + name);
+ }
+ }
+
+ this.discovery.close();
+ this.cache.close();
+ this.system.disconnect();
+ this.cache = null;
+ this.system = null;
+ }
+ }
+
+ public QueryResult executeQuery(String query, GemFireMember member)
+ throws QueryExecutionException {
+ boolean param = limitClauseAvl && DataBrowserPreferences.getEnforceLimitToQuery();
+ QueryService service = getQueryService(member);
+ return QueryUtil.executeQuery(service, query, param);
+ }
+
+ public CQQuery newCQ(String name, String queryStr, GemFireMember member)
+ throws QueryExecutionException {
+ if(this.closed) {
+ throw new IllegalStateException("This connection is closed.");
+ }
+
+ if(member == null)
+ throw new IllegalArgumentException("Please specify a non-null member attribute");
+ if(name == null)
+ throw new IllegalArgumentException("Please specify a non-null name attribute for CQ.");
+
+ if(!member.isNotifyBySubscriptionEnabled()) {
+ throw new CQException("notify-by-subscription is not enabled on this server "+member.getId());
+ }
+
+ if(this.cqs.containsKey(name)) {
+ LogUtil.info("A CQ with name :"+name+" already exists. Hence closing the this CQ.");
+ closeCQ(name);
+ }
+
+ QueryService service = getQueryService(member);
+
+ CQResult result = new CQResult();
+ CqListener[] listeners = new CqListener[] {result};
+
+ CqAttributesFactory cqAf = new CqAttributesFactory();
+ cqAf.initCqListeners(listeners);
+
+ CqAttributes attrs = cqAf.create();
+
+ try {
+ CqQuery query = service.newCq(name, queryStr, attrs);
+ CQQueryImpl wrapper = new CQQueryImpl(member, query, result);
+
+ this.cqs.put(name, wrapper);
+
+ return wrapper;
+ }
+ catch (Exception e) {
+ throw new CQException(e);
+ }
+ }
+
+ public CQQuery getCQ(String name) {
+ if(name == null)
+ throw new IllegalArgumentException("Please specify a non-null name attribute for CQ.");
+
+ return this.cqs.get(name);
+ }
+
+
+ public void closeCQ(String name) throws CQException {
+ CQQuery query = getCQ(name);
+
+ if((query != null) && (!query.isClosed())) {
+ query.close();
+ LogUtil.info("A CQ with name :"+name+" is closed.");
+ }
+
+ //Make sure to remove the reference of CQ.
+ this.cqs.remove(name);
+ }
+
+ public void stopCQ(String name) throws CQException {
+ CQQuery query = getCQ(name);
+ if((query != null) && (query.isRunning())) {
+ query.stop();
+ }
+ }
+
+ public void addGemFireMemberListener(GemFireMemberListener listener) {
+ this.discovery.addGemFireMemberListener(listener);
+ }
+
+ public GemFireMember getMember(String id) {
+ return this.discovery.getMember(id);
+ }
+
+ public GemFireMember[] getMembers() {
+ return this.discovery.getMembers();
+ }
+
+ public QueryService getQueryService(GemFireMember member) {
+ return this.discovery.getQueryService(member);
+ }
+
+ public void removeGemFireMemberListener(GemFireMemberListener listener) {
+ this.discovery.removeGemFireMemberListener(listener);
+ }
+
+
+ public void addConnectionNotificationListener(GemFireConnectionListener listener) {
+ this.discovery.addConnectionNotificationListener(listener);
+ }
+
+
+ public void removeConnectionNotificationListener(GemFireConnectionListener listener) {
+ this.discovery.removeConnectionNotificationListener(listener);
+ }
+
+ public String getGemFireSystemVersion() {
+ return this.discovery.getGemFireSystemVersion();
+ }
+
+ public long getRefreshInterval() {
+ return this.discovery.getRefreshInterval();
+ }
+
+ public void setRefreshInterval(long time) {
+ this.discovery.setRefreshInterval(time);
+ }
+
+ /**
+ * Create cache. For version prior to 6.6 uses
+ * CacheFactory.create(DistributedSystem). For 6.6 & later it uses reflection
+ * to create the cache. Using reflection we also set configuration required to
+ * receive PdxInstance for PdxSerializable objects. The cache is loner as per
+ * the properties set in DistributedSystem.
+ *
+ * @param system
+ * Distributed System instance
+ * @return Cache instance
+ */
+ private Cache createCache(DistributedSystem system) {
+ Cache cache = null;
+ boolean pdxSupported = PdxHelper.getInstance().isPdxSupported();
+
+ if (pdxSupported) {
+ // Not using ClientCacheFactory as 6.0 didn't have it.
+ CacheFactory factory = new CacheFactory();
+ Class<? extends CacheFactory> klass = factory.getClass();
+ Exception failure = null;
+ try {
+ Class<?>[] booleanArgs = new Class<?>[] {boolean.class};
+ Object[] booleanVals = new Object[] {Boolean.TRUE};
+
+ Method setPdxIgnoreUnreadFields = klass.getMethod("setPdxIgnoreUnreadFields", booleanArgs);
+ Method setPdxReadSerialized = klass.getMethod("setPdxReadSerialized", booleanArgs);
+ Method create = klass.getMethod("create", new Class<?>[0]);
+
+ //CacheFactory.setPdxIgnoreUnreadFields(boolean) - as Data Browser will only be reading data
+ setPdxIgnoreUnreadFields.invoke(factory, booleanVals);
+ //CacheFactory.setPdxReadSerialized(boolean)
+ setPdxReadSerialized.invoke(factory, booleanVals);
+ //CacheFactory.create()
+ cache = (Cache) create.invoke(factory, new Object[0]);
+ } catch (NoSuchMethodException e) {
+ failure = e;
+ } catch (IllegalAccessException e) {
+ failure = e;
+ } catch (InvocationTargetException e) {
+ failure = e;
+ } finally {
+ if (failure != null) {
+ LogUtil.error("Failed to create GemFire(" + CacheFactory.getVersion()
+ + ") Cache.", failure);
+ throw new UnsupportedOperationException("Failed to create cache " +
+ "with PDX Support. Verify GemFire product directory is accessible.");
+ }
+ }
+ } else {
+ cache = CacheFactory.create(system);
+ }
+ LogUtil.info("GemFire("+CacheFactory.getVersion()+") Cache is created. " +
+ "PDX support is" + (pdxSupported ? "" : " NOT") + " available.");
+
+ return cache;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXCallExecutor.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXCallExecutor.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXCallExecutor.java
new file mode 100644
index 0000000..d6dddd1
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXCallExecutor.java
@@ -0,0 +1,125 @@
+/*
+ * =========================================================================
+ * (c) Copyright 2002-2011, GemStone Systems, Inc. All Rights Reserved.
+ * 1260 NW Waterhouse Ave., Suite 200, Beaverton, OR 97006
+ * All Rights Reserved.
+ * =========================================================================
+ */
+
+package com.gemstone.gemfire.mgmt.DataBrowser.connection.internal;
+
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionFactory;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionFailureException;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.DSConfiguration;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnection;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.JMXOperationFailureException;
+import com.gemstone.gemfire.mgmt.DataBrowser.utils.LogUtil;
+
+/**
+ * This is an utility class which can be used to invoke JMX operations with
+ * timeout value.
+ *
+ * @author hgadre
+ */
+public class JMXCallExecutor {
+
+ public static final int DEFAULT_OPERATION_TIMEOUT = 300000;
+
+ private static DaemonThreadFactory factory = new DaemonThreadFactory();
+
+ private long operationTimeout = DEFAULT_OPERATION_TIMEOUT; // In seconds
+
+ /**
+ * The constructor.
+ *
+ * @param opTimeout
+ * the timeout value of the operation.
+ */
+ public JMXCallExecutor(long opTimeout) {
+ operationTimeout = opTimeout;
+ }
+
+ /**
+ * Default constructor.
+ */
+ public JMXCallExecutor() {
+ }
+
+
+ /**
+ * This method provides a timeout based facility to make a connection to the JMX Manager.
+ * @param config DSConfig object.
+ * @return a connected JMXBuilder instance.
+ * @throws JMXOperationFailureException in case of timeouts or exceptions
+ */
+ public GemFireConnection makeConnection(final DSConfiguration config)
+ throws ConnectionFailureException, JMXOperationFailureException {
+ ExecutorService executor = Executors.newSingleThreadExecutor(factory);
+ final BlockingQueue<Object> mailbox = new ArrayBlockingQueue<Object>(1);
+ executor.submit(new Runnable() {
+ public void run() {
+ try {
+ GemFireConnection connection = ConnectionFactory.createGemFireConnection(config);
+ if (!mailbox.offer(connection)) {
+ LogUtil.info("The user has cancelled JMX connect operation. ");
+ connection.close();
+ }
+ else {
+ LogUtil.info("The user has successfully connected to JMX Manager at " + config.getHost() + " : " + config.getPort());
+ }
+
+ } catch(ConnectionFailureException ex) {
+ mailbox.offer(ex);
+ } catch (Throwable t) {
+ mailbox.offer(new JMXOperationFailureException(t));
+ }
+ }
+ });
+
+ Object result;
+ try {
+ LogUtil.info("Starting to poll for the results of connect to JMX manager operation.");
+ result = mailbox.poll(operationTimeout, TimeUnit.MILLISECONDS);
+ if (result == null) {
+ if (!mailbox.offer(""))
+ result = mailbox.take();
+ }
+ } catch (InterruptedException e) {
+ throw new JMXOperationFailureException(e.getMessage(), e);
+ } finally {
+ executor.shutdownNow();
+ }
+
+ if (result instanceof JMXOperationFailureException) {
+ throw (JMXOperationFailureException)result;
+ } else if(result instanceof ConnectionFailureException) {
+ throw (ConnectionFailureException)result;
+ }
+
+ if (result == null || !(result instanceof GemFireConnection))
+ throw new JMXOperationFailureException(
+ "Could not connect to GemFire JMX Manager at " + config.getHost() + ":"
+ + config.getPort() + " in " + operationTimeout + " milliseconds.");
+
+ return (GemFireConnection)result;
+ }
+
+ /**
+ * This class provides a facility to create daemon threads.
+ */
+ private static class DaemonThreadFactory implements ThreadFactory {
+ public Thread newThread(Runnable r) {
+ Thread t = Executors.defaultThreadFactory().newThread(r);
+ t.setName("JMXCallExecutorThread");
+ t.setDaemon(true);
+ return t;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXDiscoveryImpl.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXDiscoveryImpl.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXDiscoveryImpl.java
new file mode 100644
index 0000000..fc47d15
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXDiscoveryImpl.java
@@ -0,0 +1,916 @@
+/*=========================================================================
+ * (c)Copyright 2002-2011, GemStone Systems, Inc. All Rights Reserved.
+ * 1260 NW Waterhouse Ave., Suite 200, Beaverton, OR 97006
+ * All Rights Reserved.
+ * =======================================================================*/
+package com.gemstone.gemfire.mgmt.DataBrowser.connection.internal;
+
+import java.io.IOException;
+import java.io.InvalidClassException;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import javax.management.JMException;
+import javax.management.JMX;
+import javax.management.ListenerNotFoundException;
+import javax.management.MBeanAttributeInfo;
+import javax.management.MBeanInfo;
+import javax.management.MBeanOperationInfo;
+import javax.management.MBeanServerConnection;
+import javax.management.Notification;
+import javax.management.NotificationListener;
+import javax.management.ObjectName;
+import javax.management.remote.JMXConnectionNotification;
+import javax.management.remote.JMXConnector;
+import javax.management.remote.JMXConnectorFactory;
+import javax.management.remote.JMXServiceURL;
+import javax.rmi.ssl.SslRMIClientSocketFactory;
+
+import com.gemstone.gemfire.admin.ConfigurationParameter;
+import com.gemstone.gemfire.admin.GemFireMemberStatus;
+import com.gemstone.gemfire.admin.RegionSubRegionSnapshot;
+import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.client.PoolFactory;
+import com.gemstone.gemfire.cache.client.PoolManager;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.internal.SocketCreator;
+import com.gemstone.gemfire.management.CacheServerMXBean;
+import com.gemstone.gemfire.management.DistributedSystemMXBean;
+import com.gemstone.gemfire.management.GemFireProperties;
+import com.gemstone.gemfire.management.MemberMXBean;
+import com.gemstone.gemfire.management.internal.JmxManagerLocatorRequest;
+import com.gemstone.gemfire.management.internal.JmxManagerLocatorResponse;
+import com.gemstone.gemfire.management.internal.MBeanJMXAdapter;
+import com.gemstone.gemfire.management.internal.cli.shell.JMXConnectionException;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionClosedException;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionFailureException;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionTerminatedEvent;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GFMemberDiscovery;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnectionListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireMemberListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.VersionMismatchException;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.CacheServerInfo;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.GemFireMember;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberStatus;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberConfigurationPrms;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberCrashedEvent;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberFactory;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberJoinedEvent;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberLeftEvent;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberUpdatedEvent;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.region.RegionFactory;
+import com.gemstone.gemfire.mgmt.DataBrowser.prefs.DataBrowserPreferences;
+import com.gemstone.gemfire.mgmt.DataBrowser.utils.LogUtil;
+
+public class JMXDiscoveryImpl implements GFMemberDiscovery, NotificationListener {
+ public static final String JMX_URI = "service:jmx:rmi://{0}/jndi/rmi://{0}:{1}/jmxconnector";
+ public static final String JMX_URI_MANAGER = "service:jmx:rmi://{0}/jndi/rmi://{0}:{1}/jmxrmi";
+ public static final String MBEAN_PROPERTY_BEAN_ID = "id";
+ public static final String MBEAN_PROPERTY_BEAN_TYPE = "type";
+ public static final String MBEAN_GEMFIRE_APPLICATION_TYPE = "Application";
+ public static final String MBEAN_GEMFIRE_CACHEVM_TYPE = "CacheVm";
+ public static final String MBEAN_DOMAIN_GEMFIRE_NAME = "GemFire";
+ public static final String MBEAN_AGENT_TYPE = "Agent";
+
+ // MGH: FindBugs reports a false positive for most of these private variables
+ // as being unused.
+ private com.gemstone.gemfire.mgmt.DataBrowser.model.DistributedSystem system_model;
+ private JMXConnector connector;
+ private MBeanServerConnection connServer;
+ private ObjectName agent;
+ private ObjectName ds;
+ private Map<String, Pool> connections;
+ private Map<String, ObjectName> jmxinfo;
+ private Timer updateTimer;
+ private JMXNotificationListener jmxNotifListener;
+ private List<GemFireMemberListener> memberListeners;
+ private List<GemFireConnectionListener> connectionListeners;
+ private volatile boolean terminated;
+ private long refreshInterval;
+ private ObjectName[] servers;
+ private DistributedSystemMXBean distributedSystemMXBeanProxy;
+ private MemberMXBean managerMemberMXBeanProxy = null;
+
+ public JMXDiscoveryImpl(String hostName, int port, String gfeSysName, String userName, String password)
+ throws ConnectionFailureException {
+ system_model = new com.gemstone.gemfire.mgmt.DataBrowser.model.DistributedSystem(
+ gfeSysName);
+ connections = Collections.synchronizedMap(new HashMap<String, Pool>());
+ jmxinfo = Collections.synchronizedMap(new HashMap<String, ObjectName>());
+ memberListeners = new ArrayList<GemFireMemberListener>();
+ connectionListeners = new ArrayList<GemFireConnectionListener>();
+
+ establishConnectionUsingJMXManager(hostName, port, userName, password);
+ //establishConnection(hostName, port);
+ jmxNotifListener = new JMXNotificationListener(this);
+
+ jmxNotifListener.start();
+
+ setRefreshInterval(DataBrowserPreferences.getDSRefreshInterval());
+ }
+
+ boolean isTerminated() {
+ return terminated;
+ }
+
+ MBeanServerConnection getConnection() {
+ if (!isTerminated())
+ return this.connServer;
+
+ return null;
+ }
+
+ ObjectName getAdminDistributedSystem() {
+ return this.ds;
+ }
+
+ public void addGemFireMemberListener(GemFireMemberListener listener) {
+ this.memberListeners.add(listener);
+ }
+
+ public void addConnectionNotificationListener(
+ GemFireConnectionListener listener) {
+ this.connectionListeners.add(listener);
+ }
+
+ public void removeConnectionNotificationListener(
+ GemFireConnectionListener listener) {
+ this.connectionListeners.remove(listener);
+ }
+
+ public void close() {
+ if (this.connServer != null) {
+ this.terminated = true;
+ this.updateTimer.cancel();
+ this.jmxNotifListener.stop();
+ try {
+ connector.close();
+ } catch (IOException e) {
+ LogUtil.warning("Exception while closing RMI connection", e);
+ }
+
+ this.connector = null;
+ this.connServer = null;
+
+ LogUtil.info("Connection to the GemFire system is closed...");
+ }
+ }
+
+
+ public DistributedSystemMXBean getDistributedSystemMXBeanProxy() {
+ return distributedSystemMXBeanProxy;
+ }
+
+ public GemFireMember getMember(String id) {
+ return this.system_model.getMember(id);
+ }
+
+ public GemFireMember[] getMembers() {
+ return this.system_model.getMembers().toArray(new GemFireMember[0]);
+ }
+
+ public QueryService getQueryService(GemFireMember member) {
+ Pool pool = null;
+
+ if (!this.connections.containsKey(member.getId())) {
+ int socketReadTimeout = DataBrowserPreferences.getSocketReadTimeoutInterval();
+ String memberName = member.getName();
+ StringBuilder cacheServersHostPort = new StringBuilder();
+
+ LogUtil.info("Preparing Pool for member :" + memberName);
+ PoolFactory factory = PoolManager.createFactory();
+ factory.setSubscriptionEnabled(true);
+ factory.setReadTimeout(socketReadTimeout);
+ CacheServerInfo[] cacheServers = member.getCacheServers();
+ for (int i = 0; i < cacheServers.length; i++) {
+ factory.addServer(cacheServers[i].getBindAddress(), cacheServers[i].getPort());
+ cacheServersHostPort.append(cacheServers[i]);
+ if (i != cacheServers.length - 1) {
+ cacheServersHostPort.append(",");
+ }
+ }
+ pool = factory.create(member.getId() + "-Pool");
+ LogUtil.fine("Prepared Pool for member :" + memberName
+ + " with client pool read timeout(ms) : " + socketReadTimeout
+ + ", Cache Servers : " + cacheServersHostPort);
+ this.connections.put(member.getId(), pool);
+ return pool.getQueryService();
+ }
+
+ pool = this.connections.get(member.getId());
+ QueryService service = pool.getQueryService();
+ return service;
+ }
+
+ public void removeGemFireMemberListener(GemFireMemberListener listener) {
+ this.memberListeners.remove(listener);
+ }
+
+ // TODO MGH - change the signature to eliminate the return value as it served
+ // no purpose
+ // The implementation does not adhere to the signature of this method. A
+ // VersionMismatchException
+ // will never be seen by a caller in a catch statement.
+ // FIXIT
+ protected void registerDSMembers() throws ConnectionClosedException,
+ JMException, IOException, ConnectionFailureException {
+ String[] params = {}, signature = {};
+
+ List<ObjectName> result = new ArrayList<ObjectName>();
+
+ // Execute manageSystemMemberApplications
+ ObjectName[] rets = (ObjectName[]) invokeOperation("manageSystemMemberApplications", ds, params, signature);
+
+ if (null != rets) {
+ result.addAll(Arrays.asList(rets));
+ LogUtil.info("Member Applications Added ==>" + result.size());
+ }
+
+ rets = (ObjectName[]) invokeOperation("manageCacheVms", ds, params, signature);
+
+ if (null != rets) {
+ result.addAll(Arrays.asList(rets));
+ LogUtil.info("CacheVms Added ==>" + result.size());
+ }
+
+ for(ObjectName member : result) {
+ registerDSMember(member);
+ }
+ }
+
+
+ protected void cleanupMember(String memberId, boolean isCrashed) {
+ memberId = memberId.replaceAll("-", ":").replaceFirst(":", "-");
+ Pool pool = this.connections.remove(memberId);
+ if (pool != null && !pool.isDestroyed())
+ pool.destroy();
+
+ this.jmxinfo.remove(memberId);
+
+ GemFireMember member = this.system_model.removeMember(memberId);
+ if (member != null) {
+ for (GemFireMemberListener listener : this.memberListeners) {
+ if (isCrashed) {
+ listener.memberEventReceived(new MemberCrashedEvent(member));
+ } else {
+ listener.memberEventReceived(new MemberLeftEvent(member));
+ }
+ }
+ }
+ }
+
+ protected boolean registerDSMemberMbean(ObjectName member) throws Exception{
+ List<CacheServerInfo> cacheServers = new ArrayList<CacheServerInfo>();
+ MemberMXBean memberMbean = JMX.newMXBeanProxy(connServer, member, MemberMXBean.class);
+ String memberId = memberMbean.getId();
+
+ if(memberMbean.isServer()){
+ for(ObjectName server : servers) {
+ String serverName = server.getKeyProperty("member");
+ LogUtil.fine("server: " + serverName);
+ // check if the cache server belongs to this member
+ if(serverName.equals(member.getKeyProperty("member"))){
+ CacheServerMXBean cmbean = JMX.newMXBeanProxy(connServer, server, CacheServerMXBean.class);
+ CacheServerInfo cs_info = new CacheServerInfo(
+ getCacheServerAddress(cmbean, memberMbean), cmbean.getPort(), true);
+ cacheServers.add(cs_info);
+ LogUtil.fine("Server " + serverName + " on member " + memberMbean.getId() +
+ ": " + cs_info.toString());
+ }
+ }
+ GemFireProperties props = memberMbean.listGemFireProperties();
+
+ GemFireMember gfMember = MemberFactory.createGemFireMemberMbean(memberId, createConfiguration(props), cacheServers,
+ createStatus(props, memberMbean), RegionFactory.getRootRegion(
+ memberMbean.getMember(), distributedSystemMXBeanProxy, connServer,
+ memberMbean.getRootRegionNames()));
+
+ if (system_model.containsMember(memberId)) {
+ LogUtil.info("Updating member :" + memberMbean.getMember());
+ system_model.addMember(gfMember);
+
+ for (GemFireMemberListener listener : this.memberListeners) {
+ listener.memberEventReceived(new MemberUpdatedEvent(gfMember));
+ }
+ } else {
+ LogUtil.info("Adding member :" + memberMbean.getMember());
+ system_model.addMember(gfMember);
+
+ for (GemFireMemberListener listener : this.memberListeners) {
+ listener.memberEventReceived(new MemberJoinedEvent(gfMember));
+ }
+ }
+ if(!isTerminated())
+ jmxinfo.put(memberId, member);
+ }
+
+ return true;
+ }
+
+ private String getCacheServerAddress(CacheServerMXBean csmbean, MemberMXBean mbean) {
+ // If cache server bind-address is set use that
+ String cacheServerAddress = csmbean.getBindAddress();
+ if (cacheServerAddress == null || cacheServerAddress.isEmpty()) {
+ // Otherwise, if the system member bind-address is set, use that
+ cacheServerAddress = mbean.listGemFireProperties().getBindAddress();
+ if (cacheServerAddress == null || cacheServerAddress.isEmpty()) {
+ // Default to mbean's host attribute
+ cacheServerAddress = mbean.getHost();
+ }
+ }
+ return cacheServerAddress;
+ }
+
+ private List<MemberConfigurationPrms> createConfiguration(GemFireProperties props){
+ List<MemberConfigurationPrms> confs = new ArrayList<MemberConfigurationPrms>();
+ if(props.getSecurityClientAuthenticator() != null && !props.getSecurityClientAuthenticator().equals("")){
+ MemberConfigurationPrms prm = new MemberConfigurationPrms();
+ prm.setName("security-client-authenticator");
+ prm.setValue(props.getSecurityClientAuthenticator());
+ confs.add(prm);
+ }
+ return confs;
+ }
+
+ private MemberStatus createStatus(GemFireProperties props, MemberMXBean mbean){
+ MemberStatus status = new MemberStatus();
+ String bindAddress = props.getBindAddress();
+ status.setBindAddress(bindAddress);
+ try {
+ // For host address: if bind-address is set, use that; otherwise use
+ // member's host attribute from mbean
+ status.setHostAddress((bindAddress != null && bindAddress.length() > 0)
+ ? InetAddress.getByName(bindAddress)
+ : InetAddress.getByName(mbean.getHost()));
+ } catch (UnknownHostException e) {
+ LogUtil.warning("Unable to determine host address for the member with ID: " + mbean.getId(), e);
+ }
+ status.setMemberName(mbean.getName());
+ status.setServerPort(props.getTcpPort());//not sure if this should be tcp
+
+ return status;
+
+ }
+
+ protected boolean registerDSMember(ObjectName member)
+ throws VersionMismatchException {
+ String[] params = {}, signature = {};
+ ObjectName cache;
+ String memberID = member.getKeyProperty(MBEAN_PROPERTY_BEAN_ID);
+ String type = member.getKeyProperty(MBEAN_PROPERTY_BEAN_TYPE);
+
+ String methodLogPrefix = "JMXHelper registerDSMember [" + memberID + "]: ";
+ LogUtil.fine(methodLogPrefix + "start");
+
+ try {
+ if (!MBEAN_GEMFIRE_APPLICATION_TYPE.equalsIgnoreCase(type) &&
+ !MBEAN_GEMFIRE_CACHEVM_TYPE.equalsIgnoreCase(type) ) {
+ LogUtil.warning("Member :"+memberID+" has an unknown type :"+type);
+ return false;
+ }
+
+ boolean hasCache = ((Boolean)getAttribute(member, "hasCache")).booleanValue();
+ if (hasCache) {
+ LogUtil.info("Member Detected =>"+memberID);
+
+ cache = (ObjectName) invokeOperation("manageCache", member,
+ params, signature);
+ GemFireMemberStatus snapshot = (GemFireMemberStatus) invokeOperation(
+ "getSnapshot", cache, params, signature);
+ // This is the only type we want.
+ if (snapshot.getIsServer()) {
+ RegionSubRegionSnapshot r_snapshot = (RegionSubRegionSnapshot) invokeOperation(
+ "getRegionSnapshot", cache, params, signature);
+ List<MemberConfigurationPrms> config_prms = getMemberConfiguration(member);
+ List<CacheServerInfo> cs_info = getCacheServerInfo(cache);
+
+ GemFireMember _member = MemberFactory.createGemFireMember(memberID,
+ config_prms, cs_info, snapshot, r_snapshot);
+
+ // If member exists already, just update it.
+ if (system_model.containsMember(memberID)) {
+ LogUtil.info("Updating member :" + _member.getName());
+ system_model.addMember(_member);
+
+ for (GemFireMemberListener listener : this.memberListeners) {
+ listener.memberEventReceived(new MemberUpdatedEvent(_member));
+ }
+ } else {
+ LogUtil.info("Adding member :" + _member.getName());
+ system_model.addMember(_member);
+
+ for (GemFireMemberListener listener : this.memberListeners) {
+ listener.memberEventReceived(new MemberJoinedEvent(_member));
+ }
+ }
+ } else {
+ LogUtil.info("Skipping the member :" + memberID
+ + " since it is not a cache server.");
+ }
+ }
+ } catch (ConnectionClosedException e) {
+ LogUtil.warning("Connection the GemFire system closed abruptly. "+e.getMessage());
+ return false;
+ } catch (VersionMismatchException ex) {
+ LogUtil.error("Failed to register/update the member " + memberID, ex);
+ throw ex;
+ } catch (JMException ex) {
+ LogUtil.error("Failed to register/update the member " + memberID, ex);
+ return false;
+ } catch (IOException ex) {
+ LogUtil.error("Failed to register/update the member " + memberID, ex);
+ return false;
+ } finally {
+ // Keep track of the ObjectName for this member, even in case of
+ // exception.
+ // If the problem is temporary, this will enable us to update the member.
+ LogUtil.fine(methodLogPrefix + "end");
+ if(!isTerminated())
+ jmxinfo.put(memberID, member);
+ }
+
+ return true;
+ }
+
+ // TODO MGH - This method needs cleaning up. If we look or 5.8 and return
+ private List<MemberConfigurationPrms> getMemberConfiguration(
+ ObjectName objName) throws ConnectionClosedException {
+ MBeanInfo info = getMBeanInfo(connServer, objName);
+ List<MemberConfigurationPrms> result = new ArrayList<MemberConfigurationPrms>();
+ MBeanOperationInfo op = getOperationInfo(info, "getConfiguration");
+
+ // This operation is exposed in GemFire 5.8. If this operation is available,
+ // then use it
+ // Otherwise fetch these values from the MBean attributes (which is more
+ // expensive than the operation).
+ if (op != null) {
+ String[] params = {}, signature = {};
+
+ try {
+ ConfigurationParameter[] paramters = (ConfigurationParameter[]) invokeOperation(
+ "getConfiguration", objName, params, signature);
+ for (int i = 0; i < paramters.length; i++) {
+ MemberConfigurationPrms prm = new MemberConfigurationPrms();
+ prm.setName(paramters[i].getName());
+ prm.setValue(paramters[i].getValue());
+ result.add(prm);
+ }
+ return result;
+ } catch(ConnectionClosedException ex) {
+ throw ex;
+
+ } catch (Exception e) {
+ LogUtil.warning("Unable to execute getConfiguration operation on "
+ + objName, e);
+ }
+ }
+
+ if (info != null) {
+ MBeanAttributeInfo[] attrs = info.getAttributes();
+ for (int i = 0; i < attrs.length; i++) {
+ String name = attrs[i].getName();
+
+ try {
+ Object value = getAttribute(objName, attrs[i].getName());
+ MemberConfigurationPrms prm = new MemberConfigurationPrms();
+ prm.setName(name);
+ prm.setValue(value);
+ result.add(prm);
+ } catch (Exception e) {
+ LogUtil.warning("Unable to get the value of attribute"
+ + attrs[i].getName() + " for object " + objName, e);
+ }
+ }
+ }
+
+ return result;
+ }
+
+ private List<CacheServerInfo> getCacheServerInfo(ObjectName cache) {
+ List<CacheServerInfo> result = new ArrayList<CacheServerInfo>();
+ String[] params = {}, signature = {};
+
+ MBeanInfo cacheMBeanInfo = getMBeanInfo(connServer, cache);
+ String manageCacheServersOpName = "manageCacheServers";
+ MBeanOperationInfo manageCacheServersOp = getOperationInfo(cacheMBeanInfo, manageCacheServersOpName);
+
+ if (manageCacheServersOp != null) {
+ try {
+ //1. 1st attempt
+ ObjectName[] cacheServers = (ObjectName[]) invokeOperation(manageCacheServersOpName, cache, params, signature);
+ // Check if 2nd attempt is needed.
+ if (cacheServers != null && cacheServers.length == 0) {
+ MBeanOperationInfo refreshOp = getOperationInfo(cacheMBeanInfo, "refresh");
+
+ //2. Call refresh before 2nd attempt
+ /* Refresh to ensure that SystemMemberCache MBean is refreshed before
+ * reading notifyBySubscription from SystemMemberCacheServer MBean */
+ if (refreshOp != null) {
+ try {
+ invokeOperation("refresh", cache, params, signature);
+ } catch (Exception e) {
+ LogUtil.warning("Unable to execute refresh operation on " + cache, e);
+ }
+ //3. 2nd attempt
+ LogUtil.fine("Attempting again to invoke '" + manageCacheServersOpName + "' after refreshing Cache MBean");
+ cacheServers = (ObjectName[]) invokeOperation(manageCacheServersOpName, cache, params, signature);
+ }
+ }
+
+ for (int i = 0; i < cacheServers.length; i++) {
+ try {
+ String bindAddress = (String) getAttribute(cacheServers[i], "bindAddress");
+ int port = ((Integer)getAttribute(cacheServers[i], "port")).intValue();
+ boolean notifyBySubscription = ((Boolean)getAttribute(cacheServers[i], "notifyBySubscription")).booleanValue();
+
+ CacheServerInfo cs_info = new CacheServerInfo(bindAddress, port,
+ notifyBySubscription);
+ result.add(cs_info);
+
+ } catch (Exception ex) {
+ LogUtil.warning(
+ "Unable to retrieve information about the cacheServer :"
+ + cacheServers[i], ex);
+ }
+ }
+
+ return result;
+ } catch (Exception e) {
+ LogUtil.warning("Unable to execute manageCacheServers operation on "
+ + cache, e);
+ }
+ }
+
+ return result;
+ }
+
+ private void establishConnectionUsingJMXManager(String hostName, int port, String userName, String password)throws ConnectionFailureException{
+ try{
+ Map<String, Object> env = new HashMap<String, Object>();
+ if (userName != null && userName.length() > 0) {
+ env.put(JMXConnector.CREDENTIALS, new String[]{userName, password});
+ }
+ if (System.getProperty("javax.net.ssl.keyStore") != null || System.getProperty("javax.net.ssl.trustStore") != null) {
+ // use ssl to connect
+ env.put("com.sun.jndi.rmi.factory.socket", new SslRMIClientSocketFactory());
+ }
+
+ int timeOut = new Long(DataBrowserPreferences.getConnectionTimeoutInterval()).intValue();
+ String jmxManagerHost = null;
+ int jmxManagerPort = 0;
+ try {
+ JmxManagerLocatorResponse locRes = JmxManagerLocatorRequest.send(hostName, port, timeOut);
+ jmxManagerHost = locRes.getHost();
+ jmxManagerPort = locRes.getPort();
+ }
+ catch (Exception e) {
+ LogUtil.error("Exception while connecting to Locator on host: " + hostName + " and port: " + port+ " : " + e);
+ LogUtil.info("Trying to connect to a JMX Manager on host: "+ hostName + " and port: " + port);
+ //try to connect JMX Manager on same host
+ jmxManagerHost = hostName;
+ jmxManagerPort = port;
+ }
+
+ if(jmxManagerHost == null || jmxManagerPort == 0){
+ throw new ConnectionFailureException("JMX manager not found");
+ }
+
+ String url = MessageFormat.format(JMX_URI_MANAGER, new Object[] {checkAndConvertToCompatibleIPv6Syntax(jmxManagerHost),
+ String.valueOf(jmxManagerPort) });
+ JMXServiceURL jmxurl = new JMXServiceURL(url);
+
+ connector = JMXConnectorFactory.connect(jmxurl, env);
+ connServer = connector.getMBeanServerConnection();
+ connector.connect();
+ LogUtil.info("Connected to the Gemfire distributed system");
+ ds = MBeanJMXAdapter.getDistributedSystemName();
+ distributedSystemMXBeanProxy = JMX.newMXBeanProxy(connServer, ds, DistributedSystemMXBean.class);
+ ObjectName managerMemberObjectName = null;
+ if (distributedSystemMXBeanProxy == null || !JMX.isMXBeanInterface(DistributedSystemMXBean.class)) {
+ connector.close();
+ throw new JMXConnectionException(JMXConnectionException.MANAGER_NOT_FOUND_EXCEPTION);
+ } else {
+ managerMemberObjectName = distributedSystemMXBeanProxy.getMemberObjectName();
+ if (managerMemberObjectName == null || !JMX.isMXBeanInterface(MemberMXBean.class)) {
+ connector.close();
+ throw new JMXConnectionException(JMXConnectionException.MANAGER_NOT_FOUND_EXCEPTION);
+ } else {
+ managerMemberMXBeanProxy = JMX.newMXBeanProxy(connServer, managerMemberObjectName, MemberMXBean.class);
+ }
+ }
+ connector.addConnectionNotificationListener(this, null, null);
+ servers = distributedSystemMXBeanProxy.listCacheServerObjectNames();
+ if(servers.length == 0){
+ LogUtil.info("No cacheservers found in the distributed system");
+ }
+ ObjectName[] members = distributedSystemMXBeanProxy.listMemberObjectNames();
+ if(members.length == 0){
+ LogUtil.info("No members found in the distributed system");
+ }
+ for(ObjectName member : members) {
+ registerDSMemberMbean(member);
+ }
+ } catch(Exception e){
+ throw new ConnectionFailureException(e);
+ }
+
+ }
+
+ // JMX Connection establishment.
+ private void establishConnection(String hostName, int port)
+ throws ConnectionFailureException {
+ try {
+ String url = MessageFormat.format(JMX_URI, new Object[] {checkAndConvertToCompatibleIPv6Syntax(hostName),
+ String.valueOf(port) });
+ JMXServiceURL jmxurl = new JMXServiceURL(url);
+ connector = JMXConnectorFactory.connect(jmxurl, null);
+ connServer = connector.getMBeanServerConnection();
+
+ String[] domains = connServer.getDomains();
+ for (int i = 0; i < domains.length; i++) {
+ LogUtil.info("DOMAIN :" + domains[i]);
+ if (MBEAN_DOMAIN_GEMFIRE_NAME.equals(domains[i])) {
+ Set<ObjectName> objectNames = connServer.queryNames(null,
+ new ObjectName(MBEAN_DOMAIN_GEMFIRE_NAME + ":*"));
+ for (ObjectName obj : objectNames) {
+ String type = obj.getKeyProperty(MBEAN_PROPERTY_BEAN_TYPE);
+ if (MBEAN_AGENT_TYPE.equalsIgnoreCase(type)) {
+ agent = obj;
+ break;
+ }
+ }
+ }
+ }
+
+ if (agent == null) {
+ throw new ConnectionFailureException(
+ "Could not find the JMX Agent MBean...");
+ }
+
+ LogUtil.info("AGENT :" + agent);
+
+ String[] params = {}, signature = {};
+ ds = (ObjectName) invokeOperation("connectToSystem", agent, params, signature);
+
+ if (ds == null) {
+ throw new ConnectionFailureException(
+ "Could not connect to GemFire Distributed System...");
+ }
+
+ connector.addConnectionNotificationListener(this, null, null);
+
+ registerDSMembers();
+ } catch (VersionMismatchException e) {
+ throw e;
+ } catch (ConnectionFailureException e) {
+ throw e;
+ } catch (ConnectionClosedException e) {
+ throw new ConnectionFailureException(e, false);
+ } catch (JMException e) {
+ throw new ConnectionFailureException(e, false);
+ } catch (IOException e) {
+ throw new ConnectionFailureException(e, true);
+ }
+ }
+
+ // JMX Operation invocation.
+ public Object invokeOperation(String name, ObjectName objName, Object[] params, String[] signature)
+ throws VersionMismatchException, ConnectionClosedException, JMException, IOException {
+
+ Object result;
+ try {
+ LogUtil.fine("JMXHelper:JMXOperation [" + name + ", on " + objName
+ + "]: started");
+
+ if(isTerminated()) {
+// LogUtil.warning("Underlaying GemFire/JMX connection is closed. Can not proceed with the JMX operation [."+name+"] execution.");
+ throw new ConnectionClosedException("Underlaying GemFire/JMX connection is closed. Can not proceed with the JMX operation [."+name+"] execution.");
+ }
+
+ result = connServer.invoke(objName, name, params, signature);
+ } catch (IOException ex1) {
+ Throwable cause = ex1.getCause();
+ // while ((cause != null) && !(cause instanceof InvalidClassException))
+ while ((null != cause)
+ && !(cause.getClass().equals(InvalidClassException.class))) {
+ cause = cause.getCause();
+ }
+
+ if(( null != cause ) && cause.getClass().equals(InvalidClassException.class)) {
+ throw new VersionMismatchException(ex1);
+ }
+
+ throw ex1;
+
+ } finally {
+ LogUtil.fine("JMXHelper:JMXOperation [" + name + ", on " + objName + "]: ended");
+ }
+
+ return result;
+ }
+
+ protected Object getAttribute(ObjectName objName, String name)
+ throws VersionMismatchException, ConnectionClosedException, JMException,
+ IOException {
+ Object result;
+
+ try {
+ LogUtil.fine("JMXHelper::getAttribute [" + name + ", on " + objName + "]: started");
+
+ if(isTerminated()) {
+ LogUtil.warning("Underlaying GemFire/JMX connection is closed. Can not retrieve the JMX attribute [."+name+"].");
+ throw new ConnectionClosedException();
+ }
+
+ result = connServer.getAttribute(objName, name);
+
+ } catch (IOException e) {
+ Throwable cause = e.getCause();
+ while ((null != cause)
+ && !(cause.getClass().equals(InvalidClassException.class))) {
+ cause = cause.getCause();
+ }
+
+ if ((null != cause)
+ && cause.getClass().equals(InvalidClassException.class)) {
+ throw new VersionMismatchException(e);
+ }
+
+ throw e;
+ } finally {
+ LogUtil.fine("JMXHelper::getAttribute [" + name + ", on " + objName + "]: ended");
+ }
+
+ return result;
+ }
+
+ // TODO MGH - Fix this awkward code. Why the finally statement when the
+ // exception is eaten?
+ protected static MBeanInfo getMBeanInfo(MBeanServerConnection mbsc,
+ ObjectName objName) {
+ MBeanInfo result = null;
+
+ try {
+ result = mbsc.getMBeanInfo(objName);
+ } catch (Exception e) {
+ LogUtil.warning("Unable to get MBeanInfo for object :" + objName, e);
+ }
+
+ return result;
+ }
+
+ protected static MBeanOperationInfo getOperationInfo(MBeanInfo info,
+ String name) {
+ MBeanOperationInfo[] operations = info.getOperations();
+ for (int i = 0; i < operations.length; i++) {
+ if (operations[i].getName().equals(name))
+ return operations[i];
+ }
+
+ return null;
+ }
+
+ public static String getMemberID(ObjectName obj) {
+ return obj.getKeyProperty(MBEAN_PROPERTY_BEAN_ID);
+ }
+
+ public static String getMemberMbeanID(ObjectName obj) {
+ return obj.getKeyProperty("member");
+ }
+
+ public void handleNotification(Notification notification, Object handback) {
+ if (notification instanceof JMXConnectionNotification) {
+ String type = notification.getType();
+ if (JMXConnectionNotification.CLOSED.equalsIgnoreCase(type)
+ || JMXConnectionNotification.FAILED.equalsIgnoreCase(type)) {
+
+ LogUtil.warning("RECEIVED JMXConnectionNotification of type " + type
+ + ".");
+
+ terminated = true;
+ connServer = null;
+
+ if (connector != null) {
+ try {
+ connector.removeConnectionNotificationListener(this);
+ } catch (ListenerNotFoundException e) { // This is impossible to
+ // happen.
+ LogUtil.warning(e.getLocalizedMessage());
+ }
+ }
+
+ for (GemFireConnectionListener listener : this.connectionListeners) {
+ LogUtil.info("CLOSE_LISTENER :" + listener.getClass());
+ listener.connectionEventReceived(new ConnectionTerminatedEvent());
+ }
+
+ }
+ }
+
+ }
+
+ public String getGemFireSystemVersion() {
+ String result = "NO_VERSION";
+
+ try {
+ // result = (String)this.connServer.getAttribute(agent, "version");
+ result = managerMemberMXBeanProxy.getVersion();
+ } catch (Exception e) {
+ LogUtil.warning("Failed to get the GemFire System version information...", e);
+ return result;
+ }
+
+ String temp = result;
+ int index = temp.indexOf("Native version");
+
+ if(index > 0) {
+ temp = result.substring(0, index);
+ }
+
+ String [] arr = temp.split("\\s+");
+
+ if(arr.length < 3)
+ return result;
+ else
+ return arr[2];
+ }
+
+ /**
+ * If the given host address contains a ":", considers it as an IPv6 address &
+ * returns the host based on RFC2732 requirements i.e. surrounds the given
+ * host address string with square brackets. If ":" is not found in the given
+ * string, simply returns the same string.
+ *
+ * @param hostAddress
+ * host address to check if it's an IPv6 address
+ * @return for an IPv6 address returns compatible host address otherwise
+ * returns the same string
+ */
+ public static String checkAndConvertToCompatibleIPv6Syntax(String hostAddress) {
+ //if host string contains ":", considering it as an IPv6 Address
+ //Conforming to RFC2732 - http://www.ietf.org/rfc/rfc2732.txt
+ if (hostAddress.indexOf(":") != -1) {
+ LogUtil.info("IPv6 host address detected, using IPv6 syntax for host in JMX connection URL");
+ hostAddress = "[" + hostAddress + "]";
+ LogUtil.info("Compatible host address is : "+hostAddress);
+ }
+ return hostAddress;
+ }
+
+ public long getRefreshInterval() {
+ return this.refreshInterval;
+ }
+
+ public void setRefreshInterval(long time) {
+ if(this.refreshInterval != time) {
+ LogUtil.info("Updated the refreshInterval to "+time);
+ this.refreshInterval = time;
+
+ if(null != this.updateTimer)
+ this.updateTimer.cancel();
+
+ UpdateTask task = new UpdateTask();
+ this.updateTimer = new Timer("DataBrowser-MemberUpdator", true);
+ this.updateTimer.schedule(task, refreshInterval, refreshInterval);
+
+ } else {
+ LogUtil.info("refreshInterval is same. Hence not updating...");
+ }
+ }
+
+ private class UpdateTask extends TimerTask {
+ @Override
+ public void run() {
+ LogUtil.info("Updating GemFire system members.");
+ try {
+ List<ObjectName> members = new ArrayList<ObjectName>();
+ synchronized (JMXDiscoveryImpl.this.jmxinfo) {
+ members.addAll(JMXDiscoveryImpl.this.jmxinfo.values());
+ }
+
+ for (ObjectName member : members) {
+ if(isTerminated()) {
+ LogUtil.warning("Since the underlaying GemFire/JMX connection is closed. Hence the member update task is complete");
+ return;
+ }
+ registerDSMemberMbean(member);
+ }
+ } catch (Exception e1) {
+ LogUtil.error("Failed to get information about GemFire system members.", e1);
+ return;
+ }
+ LogUtil.info("Updated GemFire system members.");
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXNotificationListener.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXNotificationListener.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXNotificationListener.java
new file mode 100644
index 0000000..8f13b8c
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/connection/internal/JMXNotificationListener.java
@@ -0,0 +1,196 @@
+/*=========================================================================
+ * (c)Copyright 2002-2011, GemStone Systems, Inc. All Rights Reserved.
+ * 1260 NW Waterhouse Ave., Suite 200, Beaverton, OR 97006
+ * All Rights Reserved.
+ * =======================================================================*/
+package com.gemstone.gemfire.mgmt.DataBrowser.connection.internal;
+
+import java.util.ArrayList;
+import java.util.Date;
+
+import javax.management.Notification;
+import javax.management.NotificationListener;
+import javax.management.ObjectName;
+
+import com.gemstone.gemfire.admin.jmx.internal.AdminDistributedSystemJmxImpl;
+import com.gemstone.gemfire.management.internal.beans.ResourceNotification;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionFailureException;
+import com.gemstone.gemfire.mgmt.DataBrowser.utils.LogUtil;
+
+public class JMXNotificationListener implements NotificationListener, Runnable {
+
+ private volatile boolean _stop;
+ protected ArrayList< Notification > _notifications;
+ protected Thread _notifier;
+ private JMXDiscoveryImpl discovery;
+
+ JMXNotificationListener(JMXDiscoveryImpl dscvry) {
+ this.discovery = dscvry;
+ _notifications = new ArrayList<Notification>();
+ _stop = false;
+ _notifier = new Thread(this, "JMXNotificationListener");
+ _notifier
+ .setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
+ public void uncaughtException(Thread t, Throwable e) {
+ LogUtil.error("Uncaught Exception " + t.getId(), e);
+ }
+ });
+ }
+
+ protected void addNotification(Notification notification) {
+ synchronized (this._notifications) {
+ _notifications.add(notification);
+ }
+
+ LogUtil.fine("\tAdded a new Notification :" + notification.getType());
+ }
+
+ protected int getNotificationsListSize() {
+ synchronized (this._notifications) {
+ return _notifications.size();
+ }
+ }
+
+ protected Notification getNextNotification() {
+ Notification ntfyRet = null;
+ synchronized (this._notifications) {
+ if (_notifications.size() > 0) {
+ ntfyRet = (Notification) _notifications.remove(0);
+ }
+ }
+
+ return ntfyRet;
+ }
+
+ protected void processNotification(Notification notification)
+ throws Exception {
+ String type = notification.getType();
+
+ if (type.equals(AdminDistributedSystemJmxImpl.NOTIF_MEMBER_JOINED)) {
+ discovery.registerDSMember((ObjectName) notification.getSource());
+ } else if (type.equals(AdminDistributedSystemJmxImpl.NOTIF_MEMBER_LEFT)) {
+ ObjectName mem = (ObjectName) notification.getSource();
+ String id = JMXDiscoveryImpl.getMemberID(mem);
+ discovery.cleanupMember(id, false);
+
+ } else if (type.equals(AdminDistributedSystemJmxImpl.NOTIF_MEMBER_CRASHED)) {
+ ObjectName mem = (ObjectName) notification.getSource();
+ String id = JMXDiscoveryImpl.getMemberID(mem);
+ discovery.cleanupMember(id, true);
+ }
+
+ }
+
+ protected void processMBeanNotification(Notification notification)
+ throws Exception {
+ String type = notification.getType();
+ if (type.equals(ResourceNotification.CACHE_SERVER_STOPPED)) {
+ discovery.cleanupMember((String)notification.getSource(), false);
+ } else if (type.equals(ResourceNotification.CACHE_MEMBER_DEPARTED)) {
+ discovery.cleanupMember((String)notification.getSource(), true);
+ }
+ ObjectName source = discovery.getDistributedSystemMXBeanProxy().fetchMemberObjectName((String)notification.getSource());
+ if (type.equals(ResourceNotification.CACHE_MEMBER_JOINED)) {
+ discovery.registerDSMemberMbean(source);
+ } else if (type.equals(ResourceNotification.CACHE_SERVER_STARTED)) {
+ discovery.registerDSMemberMbean(source);
+ }
+
+ }
+
+ public void handleNotification(Notification notification, Object handback) {
+ String type = notification.getType();
+ Date time = new Date(notification.getTimeStamp());
+ String memID = notification.getMessage();
+ long uniqueNum = notification.getSequenceNumber();
+
+// LogUtil.info("Received a Notification for " + "Type = " + type + " "
+ // + "Time = " + time.toString() + "\n\t" + "MemID = " + memID + " "
+ // + "Unique# = " + uniqueNum);
+
+ addNotification(notification);
+ synchronized (_notifier) {
+ _notifier.notify();
+ }
+ }
+
+ public void run() {
+ // Enter waiting
+ while (true && !_stop) {
+ boolean interrupted = Thread.interrupted();
+
+ try {
+ if (!interrupted) {
+ // Sit in a loop and process all notifications
+ Notification notification;
+ while ((notification = getNextNotification()) != null && !_stop) {
+ //processNotification(notification);
+ processMBeanNotification(notification);
+ }
+
+ if (!_stop) {
+ // Enter wait loop
+ synchronized (_notifier) {
+ _notifier.wait();
+ }
+ }
+ }
+ } catch (InterruptedException exp1) {
+ // Clear the status via Exception, since we're waiting
+ interrupted = true;
+ LogUtil
+ .fine("DSNotificaionListener: Thread's interrupt status post last interrupted "
+ + interrupted);
+
+ } catch (Exception exp2) {
+ // TODO: Need a mechanism to stop the execution.
+ LogUtil.error("Exception in JMX notification listener", exp2);
+
+ } finally {
+ if (interrupted) {
+ _stop = true;
+
+ LogUtil
+ .warning("JMXNotificaionListener: Thread got an interrupt. Should stop ");
+ }
+ }
+ }
+
+ LogUtil.fine("JMXNotificaionListener: Thread out of run loop: Stop");
+ }
+
+ public void stop() {
+ synchronized (_notifier) {
+ if (discovery.getConnection() != null) {
+ try {
+ discovery.getConnection().removeNotificationListener(
+ discovery.getAdminDistributedSystem(), this);
+ } catch (Exception e) {
+ LogUtil
+ .warning("Exception while removing JMX connection notifier", e);
+ }
+ }
+ _stop = true;
+ _notifier.notify(); // Should run the loop & exit
+ }
+ }
+
+ // TODO - MGH Is this correct? If an exception is thrown by addNotificationListener
+ // the thread does not start? Why are multiple threads started? Why does the is the Thread
+ // created in the ctor of this class a named thread while this one is not?
+ public void start() throws ConnectionFailureException {
+ if (_stop) {
+ _notifier = new Thread(this);
+ _stop = false;
+ }
+
+ try {
+ discovery.getConnection().addNotificationListener(discovery.getAdminDistributedSystem(), this, null, null);
+ } catch (Exception ex) {
+ String message = "Unable to listen for GemFire system membership changes. Reason : "+ex.getMessage();
+ throw new ConnectionFailureException(message, ex);
+ }
+
+ _notifier.start();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/CQConfiguarationPrms.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/CQConfiguarationPrms.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/CQConfiguarationPrms.java
new file mode 100644
index 0000000..9b4b8b0
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/CQConfiguarationPrms.java
@@ -0,0 +1,44 @@
+package com.gemstone.gemfire.mgmt.DataBrowser.controller;
+
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQQuery;
+
+/**
+ *
+ * @author mjha
+ *
+ */
+public class CQConfiguarationPrms extends QueryConfigurationPrms {
+
+ private ICQueryEventListener cQeventListner_;
+
+ private String name_;
+
+ private CQQuery cQuery_;
+
+ public CQConfiguarationPrms(String queryName,
+ ICQueryEventListener cQeventListner) {
+ name_ = queryName;
+ cQeventListner_ = cQeventListner;
+ }
+
+ public String getName() {
+ return name_;
+ }
+
+ @Override
+ public boolean isCQ() {
+ return true;
+ }
+
+ public CQQuery getCQuery() {
+ return cQuery_;
+ }
+
+ public void setCQuery(CQQuery query) {
+ cQuery_ = query;
+ }
+
+ public ICQueryEventListener getCQeventListner() {
+ return cQeventListner_;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DSSnapShot.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DSSnapShot.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DSSnapShot.java
new file mode 100644
index 0000000..cddbde0
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DSSnapShot.java
@@ -0,0 +1,51 @@
+package com.gemstone.gemfire.mgmt.DataBrowser.controller;
+
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnection;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.GemFireMember;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.region.GemFireRegion;
+
+/**
+ * Respesents the snap shot of DS at any time
+ *
+ * @author mjha
+ *
+ */
+public class DSSnapShot {
+
+ private GemFireConnection connection;
+
+ public DSSnapShot(GemFireConnection cnxn) {
+ this.connection = cnxn;
+ }
+
+ /**
+ * This method returns the list of members connected to a given GemFire
+ * system.
+ *
+ * @return List of GemFire members.
+ */
+ public GemFireMember[] getMembers() {
+ return connection.getMembers();
+ }
+
+ /**
+ * This method returns the information about a specified member.
+ *
+ * @param id
+ * The unique identifier of the member for which we need information.
+ * @return Member information.
+ */
+ public GemFireMember getMember(String id) {
+ return connection.getMember(id);
+ }
+
+ /**
+ * This method returns a list of all regions defined on this GemFire member.
+ *
+ * @return list of all defined regions.
+ */
+ public GemFireRegion[] getAllRegions(GemFireMember member) {
+ return member.getAllRegions();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DataBrowserController.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DataBrowserController.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DataBrowserController.java
new file mode 100644
index 0000000..e980559
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/DataBrowserController.java
@@ -0,0 +1,404 @@
+package com.gemstone.gemfire.mgmt.DataBrowser.controller;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ClientConfiguration;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.ConnectionFailureException;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.DSConfiguration;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.EndPoint;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnection;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireConnectionListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.GemFireMemberListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.JMXOperationFailureException;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.SecurityAttributes;
+import com.gemstone.gemfire.mgmt.DataBrowser.connection.internal.JMXCallExecutor;
+import com.gemstone.gemfire.mgmt.DataBrowser.controller.internal.DataBrowserCustomClassLoader;
+import com.gemstone.gemfire.mgmt.DataBrowser.controller.internal.QueryExecutionHelper;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.IMemberEvent;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.GemFireMember;
+import com.gemstone.gemfire.mgmt.DataBrowser.model.member.MemberConfigurationPrms;
+import com.gemstone.gemfire.mgmt.DataBrowser.prefs.DataBrowserPreferences;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.QueryExecutionException;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.QueryResult;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQException;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQQuery;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQResult;
+import com.gemstone.gemfire.mgmt.DataBrowser.ui.CustomMsgDispatcher;
+import com.gemstone.gemfire.mgmt.DataBrowser.ui.CustomUIMessages;
+import com.gemstone.gemfire.mgmt.DataBrowser.utils.LogUtil;
+import com.gemstone.gemfire.security.AuthenticationFailedException;
+
+/**
+ * Abstract class ,Act as bridge between view and model. It delegates the view
+ * operation to model. e.g establishing connection with DS, Executing query,
+ * getting snapshot
+ *
+ * Also registers itself as member change listener with model
+ *
+ * @author mjha
+ */
+
+public abstract class DataBrowserController implements GemFireMemberListener, GemFireConnectionListener {
+ // TODO MGH - Any advantage if this was read from config?
+ private static final String CLIENT_AUTHINITIALIZE_INTERFACE = "com.gemstone.gemfire.security.AuthInitialize";
+
+ private QueryExecutionHelper queryHelper;
+ private GemFireConnection connection = null;
+ private ConfigurationValidator configValidator;
+ private SecurityAttributes secAttributes;
+ private boolean isSecurityAttributesChanged;
+ private DSConfiguration currentConfiguartion;
+ final static CustomMsgDispatcher msgDispatcher_ = new CustomMsgDispatcher();
+
+ private DataBrowserCustomClassLoader dataBrowserCustomClassLoader = null;
+
+ public DataBrowserController() {
+ configValidator = new ConfigurationValidator();
+ }
+
+ private void initializeApplicationClassLoader(){
+ dataBrowserCustomClassLoader = new DataBrowserCustomClassLoader(new URL[] { }, ClassLoader.getSystemClassLoader());
+ Thread.currentThread().setContextClassLoader(dataBrowserCustomClassLoader);
+
+ String[] applicationClasses = DataBrowserPreferences.getApplicationClasses();
+ try {
+ setApplicationClassJars(applicationClasses);
+ }
+ catch (IOException e) {
+ // TODO do we need to log??
+ }
+ }
+
+ // MGH - Adding a method to indicate whether a connection is present
+ // This controller currently does not multiplex connections.
+ // When that happens, we should add a new public API that
+ // takes a DSSnapShot (or some other way to identify a DS)
+ // and return to the caller a flag indicating whether there
+ // is an active connection to that DS.
+ public boolean hasConnection() {
+ return null != connection;
+ }
+ /**
+ * Disconnect from currently connected DS
+ *
+ * @return true if disconnection happens successfully
+ */
+ // TODO MGH - no need to have any return value here
+ public boolean disconnect() {
+ if (connection != null) {
+ connection.removeConnectionNotificationListener(this);
+ connection.removeGemFireMemberListener(this);
+ connection.close();
+// connection.removeGemFireMemberListener(this);
+ connection = null;
+ // set flags to stop query executions, if query is already in queue
+ queryHelper.shutDown();
+ queryHelper = null;
+ currentConfiguartion = null;
+ secAttributes = null;
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Establish connection with DS based on configuration
+ *
+ * @return true if disconnection happens successfully
+ */
+ public boolean connect(DSConfiguration configuration)
+ throws ConnectionFailureException {
+
+ if (connection != null)
+ return false;
+
+ long connectionTimeoutInterval = DataBrowserPreferences.getConnectionTimeoutInterval();
+ JMXCallExecutor exec = new JMXCallExecutor(connectionTimeoutInterval);
+ try {
+ connection = exec.makeConnection(configuration);
+ }
+ catch (JMXOperationFailureException e) {
+ throw new ConnectionFailureException(e);
+ }
+
+ currentConfiguartion = configuration;
+ // initialize query helper for new connection
+ queryHelper = QueryExecutionHelper.getInstance();
+ queryHelper.start(connection);
+
+ connection.addGemFireMemberListener(this);
+ connection.addConnectionNotificationListener(this);
+ secAttributes = configuration.getSecAttributes();
+
+ return true;
+ }
+
+ public QueryResult getLastQueryResults(){
+ if(connection == null)
+ return null;
+
+ return queryHelper.getLastQueryResults();
+ }
+
+ public void stopCq(String name){
+ if(this.connection != null) {
+ try {
+ this.connection.stopCQ(name);
+ }
+ catch (CQException e) {
+ LogUtil.error("Error occured while stopping CQ" + e);
+ }
+ }
+ }
+
+ public void closeCq(String name){
+ if(this.connection != null) {
+ try {
+ this.connection.closeCQ(name);
+ }
+ catch (CQException e) {
+ LogUtil.error("Error occured while stopping CQ" + e);
+ }
+ }
+ }
+
+
+ /**
+ * Execute the query on server specified in queryPrms
+ *
+ * @param queryPrms
+ * instance of {@link QueryConfigurationPrms} , required to perform
+ * query
+ * @throws MissingSecuritiesPropException
+ * @throws ClassNotFoundException
+ */
+ public void executeQuery(QueryConfigurationPrms queryPrms)
+ throws InvalidConfigurationException, QueryExecutionException, MissingSecuritiesPropException, AuthenticationFailedException {
+ if (connection == null)
+ return;
+
+ //Perform the cleanup before executing the query. This is required to fix BUG669.
+ if(!(queryPrms instanceof CQConfiguarationPrms)) {
+ queryHelper.setLastQueryResults(null);
+ msgDispatcher_.sendCustomMessage(CustomUIMessages.DISPOSE_QUERY_RESULTS, new ArrayList(), new ArrayList());
+ }
+
+ GemFireMember member = queryPrms.getMember();
+ boolean securityEnabled = isSecurityEnabled(member) ;
+ if(securityEnabled && secAttributes == null){
+ throw new MissingSecuritiesPropException(member.getId() +" running in secure mode" + "\n" + "Specify security");
+ }
+
+ if(dataBrowserCustomClassLoader == null)
+ initializeApplicationClassLoader();
+
+ if (securityEnabled && isSecurityAttributesChanged) {
+ try {
+ DSConfiguration config = currentConfiguartion;
+ String authImpl = validateAndCLientAutherticationProperties(secAttributes.getSecurityPluginPath());
+ secAttributes.setAuthImplString(authImpl);
+ config.setSecAttributes(secAttributes);
+ disconnect();
+ connect(config);
+ }
+ catch (ConnectionFailureException e) {
+ throw new QueryExecutionException(e);
+ }
+ catch (ClassNotFoundException e) {
+ throw new QueryExecutionException(e);
+ }
+ catch (IOException e) {
+ throw new QueryExecutionException(e);
+ }
+ try {
+ addJar(secAttributes.getSecurityPluginPath());
+ }
+ catch (IOException e) {
+ throw new QueryExecutionException(e);
+ }
+
+ isSecurityAttributesChanged = false;
+ }
+
+ boolean validated = configValidator.validateQueryPrms(queryPrms);
+ if (!validated) {
+ String message = configValidator.getMessage();
+ throw new InvalidConfigurationException(message);
+ }
+
+ if (queryHelper != null){
+ if (queryPrms.isCQ()) {
+ CQConfiguarationPrms cqPrms = (CQConfiguarationPrms)queryPrms;
+ String queryString = cqPrms.getQueryString();
+ ICQueryEventListener cQEventLstnr = cqPrms.getCQeventListner();
+ CQQuery cQuery = connection
+ .newCQ(cqPrms.getName(), queryString, member);
+ cQEventLstnr.setQuery(cQuery);
+ CQResult queryResult = cQuery.getQueryResult();
+ ICQueryEventListener qeventListner = cqPrms.getCQeventListner();
+ queryResult.addCQEventListener(qeventListner);
+ queryResult.addTypeListener(qeventListner);
+ cqPrms.setCQuery(cQuery);
+ }
+ queryHelper.submitQuery(queryPrms);
+ }
+ }
+
+ public void executeCQ(CQConfiguarationPrms cqPrms)
+ throws InvalidConfigurationException, QueryExecutionException,
+ MissingSecuritiesPropException {
+ executeQuery(cqPrms);
+ }
+
+ private String validateAndCLientAutherticationProperties(String securityPluginPath) throws ClassNotFoundException, IOException{
+ String authImpl = null;
+ if(securityPluginPath != null && securityPluginPath.length() != 0){
+ try {
+ Class<?> authClass = ClassLoader.getSystemClassLoader().loadClass(CLIENT_AUTHINITIALIZE_INTERFACE);
+ authImpl = DataBrowserCustomClassLoader.getAuthProp(securityPluginPath, authClass);
+ }
+ catch (ClassNotFoundException e) {
+ String errMsg = "Implementation of "+CLIENT_AUTHINITIALIZE_INTERFACE + " not found in classpath";
+ LogUtil.error(errMsg);
+ throw new ClassNotFoundException(errMsg);
+ }
+ catch (IOException e) {
+ LogUtil.error(e.getMessage());
+ throw e;
+ }
+ }
+ return authImpl;
+ }
+
+ private boolean isSecurityEnabled(GemFireMember member){
+ boolean enabled = false;
+ MemberConfigurationPrms[] config = member.getConfig();
+ for (int i = 0; i < config.length; i++) {
+ if(config[i].getName().equals("security-client-authenticator")){
+ Object value = config[i].getValue();
+ if(value != null){
+ if(value instanceof String){
+ String val= (String)value;
+ if(val.trim().length() != 0){
+ enabled = true;
+ return enabled;
+ }
+ }
+ }
+ }
+ }
+ return enabled ;
+ }
+
+ /**
+ * Api, to add jar of custom classes used as region entry in cache severs
+ * region, at runtime
+ *
+ * @param jars
+ * @throws IOException
+ */
+ public void setApplicationClassJars(String[] jars) throws IOException {
+ if(dataBrowserCustomClassLoader == null)
+ return;
+
+ for (int i = 0; i < jars.length; i++) {
+ addJar(jars[i]);
+ }
+
+ try {
+
+ DataBrowserCustomClassLoader.loadApplicationClass(dataBrowserCustomClassLoader, jars);
+ }
+ catch (ClassNotFoundException e) {
+ // TODO need to log???
+ }
+ }
+
+ private void addJar(String jar) throws IOException {
+ File file = new File(jar);
+ URL url = file.toURI().toURL();
+ dataBrowserCustomClassLoader.addURL(url);
+ }
+
+ /**
+ *
+ * @return instance {@link DSSnapShot} of Ds connected
+ */
+ public DSSnapShot getDSSnapShot() {
+ if (connection == null)
+ return null;
+ DSSnapShot snapShot = new DSSnapShot(connection);
+ return snapShot;
+ }
+
+ public DSConfiguration getCurrentConnection(){
+ return currentConfiguartion;
+ }
+
+ public void setSecurityAttributes(SecurityAttributes attr) {
+ secAttributes = attr;
+ isSecurityAttributesChanged = true;
+ }
+
+ public abstract void memberEventReceived(IMemberEvent memEvent);
+
+
+ public void setConnectionRefreshInterval(long interval) {
+ if(null != this.connection) {
+ this.connection.setRefreshInterval(interval);
+ }
+ }
+
+ /**
+ * Validator to validate connection configuration
+ *
+ * @author mjha
+ */
+ private static class ConfigurationValidator {
+
+ private String msg;
+
+ public boolean validateConnectionConfiguartion(ClientConfiguration configuration) {
+ List<EndPoint> cacheServers = configuration.getCacheServers();
+
+ if (cacheServers == null || cacheServers.size() == 0) {
+
+ msg = "Host/port List provided is not valid";
+
+ return false;
+ }
+
+ return true;
+ }
+
+ public boolean validateQueryPrms(QueryConfigurationPrms prms) {
+ String queryString = prms.getQueryString();
+ GemFireMember member = prms.getMember();
+ IQueryExecutionListener queryExecutionListener = prms
+ .getQueryExecutionListener();
+
+ if (queryString == null) {
+ msg = "no query string found to query";
+ return false;
+ }
+ else if (member == null) {
+ msg = "pass valid member to pass query";
+ return false;
+ }
+ else if (queryExecutionListener == null) {
+ msg = "call back object to process query is null";
+ return false;
+ }
+
+ return true;
+ }
+
+ public String getMessage() {
+ return msg;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/e2ccf3da/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/ICQueryEventListener.java
----------------------------------------------------------------------
diff --git a/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/ICQueryEventListener.java b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/ICQueryEventListener.java
new file mode 100644
index 0000000..871bf5f
--- /dev/null
+++ b/databrowser/src/com/gemstone/gemfire/mgmt/DataBrowser/controller/ICQueryEventListener.java
@@ -0,0 +1,17 @@
+package com.gemstone.gemfire.mgmt.DataBrowser.controller;
+
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQEventListener;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.CQQuery;
+import com.gemstone.gemfire.mgmt.DataBrowser.query.cq.TypeListener;
+
+/**
+ * This interface provides call back functionality for listening query execution
+ * completed result
+ *
+ * @author mjha
+ **/
+public interface ICQueryEventListener extends CQEventListener, TypeListener{
+
+ public void setQuery(CQQuery cQuery);
+
+}