You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by kl...@apache.org on 2017/05/05 20:54:44 UTC
[3/3] geode git commit: fixup
fixup
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/db79d223
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/db79d223
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/db79d223
Branch: refs/heads/feature/GEODE-2632-12
Commit: db79d2238d626ed74c2c00261c22c7a817d69318
Parents: bccdb56
Author: Kirk Lund <kl...@apache.org>
Authored: Fri May 5 13:54:25 2017 -0700
Committer: Kirk Lund <kl...@apache.org>
Committed: Fri May 5 13:54:25 2017 -0700
----------------------------------------------------------------------
.../geode/internal/cache/GemFireCacheImpl.java | 2 +
.../geode/internal/cache/InternalCache.java | 5 +
.../geode/internal/cache/LocalRegion.java | 18 +-
.../geode/internal/cache/PartitionedRegion.java | 2 +-
.../cache/tier/sockets/CacheClientNotifier.java | 357 +------------------
.../cache/tier/sockets/CacheClientUpdater.java | 74 +---
.../internal/cache/xmlcache/CacheCreation.java | 32 ++
.../internal/BaseManagementService.java | 20 +-
.../management/internal/FederatingManager.java | 76 +---
.../geode/management/internal/LocalManager.java | 47 +--
.../internal/beans/BeanUtilFuncs.java | 17 +-
.../internal/beans/QueryDataFunction.java | 25 +-
.../internal/cli/commands/ClientCommands.java | 22 +-
.../internal/cli/commands/DataCommands.java | 2 +-
.../internal/cli/commands/MemberCommands.java | 38 +-
.../internal/cli/commands/RegionCommands.java | 17 +-
.../internal/cli/commands/WanCommands.java | 175 +++------
.../cli/functions/DataCommandFunction.java | 54 +--
.../extension/mock/MockExtensionCommands.java | 19 +-
.../tier/sockets/command/CommitCommandTest.java | 4 +-
.../tier/sockets/command/ContainsKey66Test.java | 10 +-
.../tier/sockets/command/ContainsKeyTest.java | 7 +-
.../tier/sockets/command/CreateRegionTest.java | 6 +-
.../tier/sockets/command/Destroy65Test.java | 10 +-
.../tier/sockets/command/DestroyRegionTest.java | 8 +-
.../cache/tier/sockets/command/DestroyTest.java | 11 +-
.../cache/tier/sockets/command/Get70Test.java | 14 +-
.../tier/sockets/command/GetAll651Test.java | 13 +-
.../tier/sockets/command/GetAll70Test.java | 13 +-
.../cache/tier/sockets/command/GetAllTest.java | 6 +-
.../sockets/command/GetAllWithCallbackTest.java | 16 +-
...tClientPartitionAttributesCommand66Test.java | 7 +-
...GetClientPartitionAttributesCommandTest.java | 4 +-
.../tier/sockets/command/InvalidateTest.java | 14 +-
.../cache/tier/sockets/command/KeySetTest.java | 12 +-
.../cache/tier/sockets/command/Put61Test.java | 12 +-
.../cache/tier/sockets/command/Put65Test.java | 12 +-
.../cache/tier/sockets/command/PutTest.java | 10 +-
.../sockets/command/RegisterInterest61Test.java | 9 +-
.../command/RegisterInterestList61Test.java | 13 +-
.../command/RegisterInterestList66Test.java | 7 +-
.../command/RegisterInterestListTest.java | 13 +-
.../sockets/command/RegisterInterestTest.java | 21 +-
.../tier/sockets/command/RemoveAllTest.java | 15 +-
.../cache/tier/sockets/command/RequestTest.java | 5 +-
.../sockets/command/UnregisterInterestTest.java | 12 +-
.../internal/cli/CliUtilDUnitTest.java | 81 +----
.../AbstractCommandsSupportJUnitTest.java | 31 +-
.../commands/DiskStoreCommandsJUnitTest.java | 31 +-
.../commands/GemfireDataCommandsDUnitTest.java | 87 +++--
.../cli/commands/IndexCommandsJUnitTest.java | 18 +-
.../internal/cli/LuceneIndexCommands.java | 13 +-
.../cli/LuceneIndexCommandsJUnitTest.java | 60 ++--
.../ClusterConfigurationDUnitTest.java | 7 +-
54 files changed, 471 insertions(+), 1143 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
index fbfd2b7..d9368e9 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/GemFireCacheImpl.java
@@ -4406,6 +4406,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
/**
* Blocks until no register interests are in progress.
*/
+ @Override
public void waitForRegisterInterestsInProgress() {
// In *this* particular context, let the caller know that
// his cache has been cancelled. doWait below would do that as
@@ -5169,6 +5170,7 @@ public class GemFireCacheImpl implements InternalCache, InternalClientCache, Has
return PdxInstanceFactoryImpl.newCreator(className, true);
}
+ @Override
public PdxInstanceFactory createPdxInstanceFactory(String className, boolean expectDomainClass) {
return PdxInstanceFactoryImpl.newCreator(className, expectDomainClass);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
index c7ee361..5533ed1 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/InternalCache.java
@@ -62,6 +62,7 @@ import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.offheap.MemoryAllocator;
import org.apache.geode.management.internal.JmxManagerAdvisor;
import org.apache.geode.management.internal.RestAgent;
+import org.apache.geode.pdx.PdxInstanceFactory;
import org.apache.geode.pdx.internal.TypeRegistry;
/**
@@ -302,4 +303,8 @@ public interface InternalCache extends Cache, Extensible<Cache>, CacheTime {
CacheServer addCacheServer(boolean isGatewayReceiver);
void setReadSerialized(boolean value);
+
+ PdxInstanceFactory createPdxInstanceFactory(String className, boolean expectDomainClass);
+
+ void waitForRegisterInterestsInProgress();
}
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index 73db0a4..54ec6a4 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -678,16 +678,14 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
/**
* initialize the event tracker. Not all region implementations want or need one of these. Regions
- * that require one should reimplement this method and create one like so: {@code
- *
- *
-
- <pre>
+ * that require one should reimplement this method and create one like so:
+ *
+ * <pre>
+ * {@code
* this.eventTracker = new EventTracker(this.cache);
* this.eventTracker.start();
- </pre>
-
- }
+ * }
+ * </pre>
*/
void createEventTracker() {
// if LocalRegion is changed to have an event tracker, then the initialize()
@@ -696,7 +694,6 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
}
/**
- *
* Other region classes may track events using different mechanisms than EventTrackers
*/
EventTracker getEventTracker() {
@@ -1959,9 +1956,8 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
boolean result = entry != null;
if (result) {
ReferenceCountHelper.skipRefCountTracking();
- // no need to decompress since we only want to
+ // no need to decompress since we only want to know if we have an existing value
Object val = entry.getTransformedValue();
- // know if we have an existing value
if (val instanceof StoredObject) {
OffHeapHelper.release(val);
ReferenceCountHelper.unskipRefCountTracking();
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
index 4e061ea..79dd50b 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
@@ -6586,7 +6586,7 @@ public class PartitionedRegion extends LocalRegion
/**
* Returns the lockname used by Distributed Lock service to clean the
- * {@code allPartitionedRegions{@code
+ * {@code allPartitionedRegions}.
*
* @return String
*/
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
index 8a79fb6..e79bfbd 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientNotifier.java
@@ -50,7 +50,6 @@ import org.apache.geode.DataSerializer;
import org.apache.geode.Instantiator;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.StatisticsFactory;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.InterestRegistrationEvent;
@@ -135,15 +134,10 @@ public class CacheClientNotifier {
/**
* Factory method to construct a CacheClientNotifier <code>CacheClientNotifier</code> instance.
*
- * @param cache The GemFire <code>Cache</code>
- * @param acceptorStats
- * @param maximumMessageCount
- * @param messageTimeToLive
- * @param listener
- * @param overflowAttributesList
+ * @param cache The GemFire <code>InternalCache</code>
* @return A <code>CacheClientNotifier</code> instance
*/
- public static synchronized CacheClientNotifier getInstance(Cache cache,
+ public static synchronized CacheClientNotifier getInstance(InternalCache cache,
CacheServerStats acceptorStats, int maximumMessageCount, int messageTimeToLive,
ConnectionListener listener, List overflowAttributesList, boolean isGatewayReceiver) {
if (ccnSingleton == null) {
@@ -156,13 +150,6 @@ public class CacheClientNotifier {
// In this case, the HaContainer should be lazily created here
ccnSingleton.initHaContainer(overflowAttributesList);
}
- // else {
- // ccnSingleton.acceptorStats = acceptorStats;
- // ccnSingleton.maximumMessageCount = maximumMessageCount;
- // ccnSingleton.messageTimeToLive = messageTimeToLive;
- // ccnSingleton._connectionListener = listener;
- // ccnSingleton.setCache((GemFireCache)cache);
- // }
return ccnSingleton;
}
@@ -176,8 +163,6 @@ public class CacheClientNotifier {
* @param dos the <code>DataOutputStream</code> to use for writing the message
* @param type a byte representing the message type
* @param p_msg the message to be written; can be null
- * @param clientVersion
- *
*/
private void writeMessage(DataOutputStream dos, byte type, String p_msg, Version clientVersion)
throws IOException {
@@ -246,35 +231,12 @@ public class CacheClientNotifier {
* @param dos the <code>DataOutputStream</code> to use for writing the message
* @param type a byte representing the exception type
* @param ex the exception to be written; should not be null
- * @param clientVersion
- *
*/
private void writeException(DataOutputStream dos, byte type, Exception ex, Version clientVersion)
throws IOException {
-
writeMessage(dos, type, ex.toString(), clientVersion);
}
- // /**
- // * Factory method to return the singleton <code>CacheClientNotifier</code>
- // * instance.
- // * @return the singleton <code>CacheClientNotifier</code> instance
- // */
- // public static CacheClientNotifier getInstance()
- // {
- // return _instance;
- // }
-
- // /**
- // * Shuts down the singleton <code>CacheClientNotifier</code> instance.
- // */
- // public static void shutdownInstance()
- // {
- // if (_instance == null) return;
- // _instance.shutdown();
- // _instance = null;
- // }
-
/**
* Registers a new client updater that wants to receive updates with this server.
*
@@ -353,12 +315,10 @@ public class CacheClientNotifier {
DistributedSystem system = this.getCache().getDistributedSystem();
Properties sysProps = system.getProperties();
String authenticator = sysProps.getProperty(SECURITY_CLIENT_AUTHENTICATOR);
- // TODO;hitesh for conflation
+
if (clientVersion.compareTo(Version.GFE_603) >= 0) {
byte[] overrides = HandShake.extractOverrides(new byte[] {(byte) dis.read()});
-
clientConflation = overrides[0];
-
} else {
clientConflation = (byte) dis.read();
}
@@ -377,7 +337,6 @@ public class CacheClientNotifier {
proxy = registerClient(socket, proxyID, proxy, isPrimary, clientConflation, clientVersion,
acceptorId, notifyBySubscription);
- // TODO:hitesh
Properties credentials = HandShake.readCredentials(dis, dos, system);
if (credentials != null && proxy != null) {
if (securityLogWriter.fineEnabled()) {
@@ -443,11 +402,9 @@ public class CacheClientNotifier {
return;
}
-
this._statistics.endClientRegistration(startTime);
}
-
/**
* Registers a new client that wants to receive updates with this server.
*
@@ -676,7 +633,6 @@ public class CacheClientNotifier {
/**
* Makes Primary to this CacheClientProxy and start the dispatcher of the CacheClientProxy
*
- * @param proxyId
* @param isClientReady Whether the marker has already been processed. This value helps determine
* whether to start the dispatcher.
*/
@@ -693,9 +649,6 @@ public class CacheClientNotifier {
* Then, start or resume the dispatcher. Otherwise, let the clientReady message start the
* dispatcher. See CacheClientProxy.startOrResumeMessageDispatcher if
* (!proxy._messageDispatcher.isAlive()) {
- *
- * proxy._messageDispatcher._messageQueue.setPrimary(true); proxy._messageDispatcher.start();
- * }
*/
if (isClientReady || !proxy.isDurable()) {
if (logger.isDebugEnabled()) {
@@ -711,8 +664,6 @@ public class CacheClientNotifier {
/**
* Adds or updates entry in the dispatched message map when client sends an ack.
*
- * @param proxyId
- * @param eid
* @return success
*/
public boolean processDispatchedMessage(ClientProxyMembershipID proxyId, EventID eid) {
@@ -752,8 +703,6 @@ public class CacheClientNotifier {
* Unregisters an existing client from this server.
*
* @param memberId Uniquely identifies the client
- *
- *
*/
public void unregisterClient(ClientProxyMembershipID memberId, boolean normalShutdown) {
if (logger.isDebugEnabled()) {
@@ -779,13 +728,11 @@ public class CacheClientNotifier {
/**
* The client represented by the proxyId is ready to receive updates.
- *
- * @param proxyId
*/
public void readyForEvents(ClientProxyMembershipID proxyId) {
CacheClientProxy proxy = getClientProxy(proxyId);
if (proxy == null) {
- // @todo log a message
+ // TODO: log a message
} else {
// False signifies that a marker message has not already been processed.
// Generate and send one.
@@ -815,7 +762,6 @@ public class CacheClientNotifier {
CacheClientNotifier instance = ccnSingleton;
if (instance != null) {
instance.singletonNotifyClients(event, null);
-
}
}
@@ -827,7 +773,6 @@ public class CacheClientNotifier {
CacheClientNotifier instance = ccnSingleton;
if (instance != null) {
instance.singletonNotifyClients(event, cmsg);
-
}
}
@@ -837,10 +782,6 @@ public class CacheClientNotifier {
FilterInfo filterInfo = event.getLocalFilterInfo();
- // if (_logger.fineEnabled()) {
- // _logger.fine("Client dispatcher processing event " + event);
- // }
-
FilterProfile regionProfile = ((LocalRegion) event.getRegion()).getFilterProfile();
if (filterInfo != null) {
// if the routing was made using an old profile we need to recompute it
@@ -903,7 +844,7 @@ public class CacheClientNotifier {
}
if (!ids.isEmpty()) {
if (isTraceEnabled) {
- logger.trace("adding invalidation routing to message for {}" + ids);
+ logger.trace("adding invalidation routing to message for {}", ids);
}
clientMessage.addClientInterestList(ids, false);
filterClients.addAll(ids);
@@ -962,10 +903,8 @@ public class CacheClientNotifier {
if (filterInfo.filterProcessedLocally) {
removeDestroyTokensFromCqResultKeys(event, filterInfo);
}
-
}
-
private void removeDestroyTokensFromCqResultKeys(InternalCacheEvent event,
FilterInfo filterInfo) {
FilterProfile regionProfile = ((LocalRegion) event.getRegion()).getFilterProfile();
@@ -984,12 +923,9 @@ public class CacheClientNotifier {
}
}
-
/**
* delivers the given message to all proxies for routing. The message should already have client
* interest established, or override the isClientInterested method to implement its own routing
- *
- * @param clientMessage
*/
public static void routeClientMessage(Conflatable clientMessage) {
CacheClientNotifier instance = ccnSingleton;
@@ -1012,7 +948,7 @@ public class CacheClientNotifier {
}
}
- /*
+ /**
* this is for server side registration of client queue
*/
public static void routeSingleClientMessage(ClientUpdateMessage clientMessage,
@@ -1052,7 +988,6 @@ public class CacheClientNotifier {
if (deadProxies != null) {
closeDeadProxies(deadProxies, false);
}
-
}
/**
@@ -1144,7 +1079,6 @@ public class CacheClientNotifier {
* @param operation The operation that occurred (e.g. AFTER_CREATE)
* @param event The event containing the data to be updated
* @return a <code>ClientUpdateMessage</code>
- * @throws Exception
*/
private ClientUpdateMessageImpl initializeMessage(EnumListenerEvent operation, CacheEvent event)
throws Exception {
@@ -1217,87 +1151,6 @@ public class CacheClientNotifier {
|| operation == EnumListenerEvent.AFTER_REGION_INVALIDATE;
}
- // /**
- // * Queues the <code>ClientUpdateMessage</code> to be distributed
- // * to interested clients. This method is not being used currently.
- // * @param clientMessage The <code>ClientUpdateMessage</code> to be queued
- // */
- // protected void notifyClients(final ClientUpdateMessage clientMessage)
- // {
- // if (USE_SYNCHRONOUS_NOTIFICATION)
- // {
- // // Execute the method in the same thread as the caller
- // deliver(clientMessage);
- // }
- // else {
- // // Obtain an Executor and use it to execute the method in its own thread
- // try
- // {
- // getExecutor().execute(new Runnable()
- // {
- // public void run()
- // {
- // deliver(clientMessage);
- // }
- // }
- // );
- // } catch (InterruptedException e)
- // {
- // _logger.warning("CacheClientNotifier: notifyClients interrupted", e);
- // Thread.currentThread().interrupt();
- // }
- // }
- // }
-
- // /**
- // * Updates the information this <code>CacheClientNotifier</code> maintains
- // * for a given edge client. It is invoked when a edge client re-connects to
- // * the server.
- // *
- // * @param clientHost
- // * The host on which the client runs (i.e. the host the
- // * CacheClientNotifier uses to communicate with the
- // * CacheClientUpdater) This is used with the clientPort to uniquely
- // * identify the client
- // * @param clientPort
- // * The port through which the server communicates with the client
- // * (i.e. the port the CacheClientNotifier uses to communicate with
- // * the CacheClientUpdater) This is used with the clientHost to
- // * uniquely identify the client
- // * @param remotePort
- // * The port through which the client communicates with the server
- // * (i.e. the new port the ConnectionImpl uses to communicate with the
- // * ServerConnection)
- // * @param membershipID
- // * Uniquely idenifies the client
- // */
- // public void registerClientPort(String clientHost, int clientPort,
- // int remotePort, ClientProxyMembershipID membershipID)
- // {
- // if (_logger.fineEnabled())
- // _logger.fine("CacheClientNotifier: Registering client port: "
- // + clientHost + ":" + clientPort + " with remote port " + remotePort
- // + " and ID " + membershipID);
- // for (Iterator i = getClientProxies().iterator(); i.hasNext();) {
- // CacheClientProxy proxy = (CacheClientProxy)i.next();
- // if (_logger.finerEnabled())
- // _logger.finer("CacheClientNotifier: Potential client: " + proxy);
- // //if (proxy.representsCacheClientUpdater(clientHost, clientPort))
- // if (proxy.isMember(membershipID)) {
- // if (_logger.finerEnabled())
- // _logger
- // .finer("CacheClientNotifier: Updating remotePorts since host and port are a match");
- // proxy.addPort(remotePort);
- // }
- // else {
- // if (_logger.finerEnabled())
- // _logger.finer("CacheClientNotifier: Host and port "
- // + proxy.getRemoteHostAddress() + ":" + proxy.getRemotePort()
- // + " do not match " + clientHost + ":" + clientPort);
- // }
- // }
- // }
-
/**
* Registers client interest in the input region and key.
*
@@ -1348,23 +1201,9 @@ public class CacheClientNotifier {
}
}
- /*
- * protected void addFilterRegisteredClients(String regionName, ClientProxyMembershipID
- * membershipID) throws RegionNotFoundException { // Update Regions book keeping. LocalRegion
- * region = (LocalRegion)this._cache.getRegion(regionName); if (region == null) { //throw new
- * AssertionError("Could not find region named '" + regionName + "'"); // @todo: see bug 36805 //
- * fix for bug 37979 if (_logger.fineEnabled()) { _logger .fine("CacheClientNotifier: Client " +
- * membershipID + " :Throwing RegionDestroyedException as region: " + regionName +
- * " is not present."); } throw new RegionDestroyedException("registerInterest failed",
- * regionName); } else { region.getFilterProfile().addFilterRegisteredClients(this, membershipID);
- * } }
- */
-
/**
* Store region and delta relation
*
- * @param regionsWithEmptyDataPolicy
- * @param regionName
* @param regionDataPolicy (0==empty)
* @since GemFire 6.1
*/
@@ -1455,13 +1294,11 @@ public class CacheClientNotifier {
}
}
-
/**
* If the conflatable is an instance of HAEventWrapper, and if the corresponding entry is present
* in the haContainer, set the reference to the clientUpdateMessage to null and putInProgress flag
* to false. Also, if the ref count is zero, then remove the entry from the haContainer.
*
- * @param conflatable
* @since GemFire 5.7
*/
private void checkAndRemoveFromClientMsgsRegion(Conflatable conflatable) {
@@ -1482,9 +1319,6 @@ public class CacheClientNotifier {
}
}
}
- // else {
- // This is a replay-of-event case.
- // }
} else {
// This wrapper resides in haContainer.
wrapper.setClientUpdateMessage(null);
@@ -1523,7 +1357,6 @@ public class CacheClientNotifier {
return proxy;
}
-
/**
* Returns the <code>CacheClientProxy</code> associated to the durableClientId
*
@@ -1593,10 +1426,6 @@ public class CacheClientNotifier {
membershipID);
logger.debug("{}::getClientProxySameDS(), Number of proxies in the Cache Clinet Notifier: {}",
this, getClientProxies().size());
- /*
- * _logger.fine(this + "::getClientProxySameDS(), Proxies in the Cache Clinet Notifier: " +
- * getClientProxies());
- */
}
CacheClientProxy proxy = null;
for (Iterator i = getClientProxies().iterator(); i.hasNext();) {
@@ -1616,7 +1445,6 @@ public class CacheClientNotifier {
return proxy;
}
-
/**
* It will remove the clients connected to the passed acceptorId. If its the only server, shuts
* down this instance.
@@ -1702,7 +1530,6 @@ public class CacheClientNotifier {
}
}
this.timedOutDurableClientProxies.remove(proxy.getProxyID());
-
}
protected void addClientInitProxy(CacheClientProxy proxy) throws IOException {
@@ -1717,7 +1544,6 @@ public class CacheClientNotifier {
return this._initClientProxies.containsKey(proxy.getProxyID());
}
-
/**
* Returns (possibly stale) set of memberIds for all clients being actively notified by this
* server.
@@ -1779,7 +1605,6 @@ public class CacheClientNotifier {
* @since GemFire 5.6
*/
public boolean hasPrimaryForDurableClient(String durableId) {
-
for (Iterator iter = this._clientProxies.values().iterator(); iter.hasNext();) {
CacheClientProxy proxy = (CacheClientProxy) iter.next();
ClientProxyMembershipID proxyID = proxy.getProxyID();
@@ -1826,7 +1651,6 @@ public class CacheClientNotifier {
return false;
}
-
/**
* Removes an existing <code>CacheClientProxy</code> from the list of known client proxies
*
@@ -1845,7 +1669,6 @@ public class CacheClientNotifier {
chm.numOfClientsPerVersion.decrementAndGet(proxy.getVersion().ordinal());
}
}
-
}
void durableClientTimedOut(ClientProxyMembershipID client) {
@@ -1866,17 +1689,6 @@ public class CacheClientNotifier {
return Collections.unmodifiableCollection(this._clientProxies.values());
}
- // /**
- // * Returns the <code>Executor</code> that delivers messages to the
- // * <code>CacheClientProxy</code> instances.
- // * @return the <code>Executor</code> that delivers messages to the
- // * <code>CacheClientProxy</code> instances
- // */
- // protected Executor getExecutor()
- // {
- // return _executor;
- // }
-
private void closeAllClientCqs(CacheClientProxy proxy) {
CqService cqService = proxy.getCache().getCqService();
if (cqService != null) {
@@ -1899,7 +1711,6 @@ public class CacheClientNotifier {
/**
* Shuts down durable client proxy
- *
*/
public boolean closeDurableClientProxy(String durableClientId) throws CacheException {
CacheClientProxy ccp = getClientProxy(durableClientId);
@@ -1915,6 +1726,7 @@ public class CacheClientNotifier {
if (logger.isDebugEnabled()) {
logger.debug("Cannot close running durable client: {}", durableClientId);
}
+ // TODO: never throw an anonymous inner class
throw new CacheException("Cannot close a running durable client : " + durableClientId) {};
}
}
@@ -1958,7 +1770,6 @@ public class CacheClientNotifier {
} // for
}
-
/**
* Registers a new <code>InterestRegistrationListener</code> with the set of
* <code>InterestRegistrationListener</code>s.
@@ -2029,9 +1840,9 @@ public class CacheClientNotifier {
}
/**
- * Returns this <code>CacheClientNotifier</code>'s <code>Cache</code>.
+ * Returns this <code>CacheClientNotifier</code>'s <code>InternalCache</code>.
*
- * @return this <code>CacheClientNotifier</code>'s <code>Cache</code>
+ * @return this <code>CacheClientNotifier</code>'s <code>InternalCache</code>
*/
protected InternalCache getCache() { // TODO:SYNC: looks wrong
if (this._cache != null && this._cache.isClosed()) {
@@ -2066,19 +1877,12 @@ public class CacheClientNotifier {
protected void handleInterestEvent(InterestRegistrationEvent event) {
LocalRegion region = (LocalRegion) event.getRegion();
region.handleInterestEvent(event);
-
}
/**
- * Constructor.
- *
- * @param cache The GemFire <code>Cache</code>
- * @param acceptorStats
- * @param maximumMessageCount
- * @param messageTimeToLive
+ * @param cache The GemFire <code>InternalCache</code>
* @param listener a listener which should receive notifications abouts queues being added or
* removed.
- * @param overflowAttributesList
*/
private CacheClientNotifier(InternalCache cache, CacheServerStats acceptorStats,
int maximumMessageCount, int messageTimeToLive, ConnectionListener listener,
@@ -2109,9 +1913,6 @@ public class CacheClientNotifier {
}
this._statistics = new CacheClientNotifierStats(factory);
- // Initialize the executors
- // initializeExecutors(this._logger);
-
try {
this.logFrequency = Long.valueOf(System.getProperty(MAX_QUEUE_LOG_FREQUENCY));
if (this.logFrequency <= 0) {
@@ -2165,13 +1966,6 @@ public class CacheClientNotifier {
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.distributed.internal.DistributionMessage#process(org.apache.geode.
- * distributed.internal.DistributionManager)
- */
@Override
protected void process(DistributionManager dm) {
// Get the proxy for the proxy id
@@ -2197,11 +1991,6 @@ public class CacheClientNotifier {
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.internal.DataSerializableFixedID#getDSFID()
- */
public int getDSFID() {
return SERVER_INTEREST_REGISTRATION_MESSAGE;
}
@@ -2223,107 +2012,8 @@ public class CacheClientNotifier {
this.clientMessage = new ClientInterestMessageImpl();
InternalDataSerializer.invokeFromData(this.clientMessage, in);
}
-
}
-
- // * Initializes the <code>QueuedExecutor</code> and
- // <code>PooledExecutor</code>
- // * used to deliver messages to <code>CacheClientProxy</code> instances.
- // * @param logger The GemFire <code>LogWriterI18n</code>
- // */
- // private void initializeExecutors(LogWriterI18n logger)
- // {
- // // Create the thread groups
- // final ThreadGroup loggerGroup = LoggingThreadGroup.createThreadGroup("Cache
- // Client Notifier Logger Group", logger);
- // final ThreadGroup notifierGroup =
- // new ThreadGroup("Cache Client Notifier Group")
- // {
- // public void uncaughtException(Thread t, Throwable e)
- // {
- // Thread.dumpStack();
- // loggerGroup.uncaughtException(t, e);
- // //CacheClientNotifier.exceptionInThreads = true;
- // }
- // };
- //
- // // Originally set ThreadGroup to be a daemon, but it was causing the
- // following
- // // exception after five minutes of non-activity (the keep alive time of the
- // // threads in the PooledExecutor.
- //
- // // java.lang.IllegalThreadStateException
- // // at java.lang.ThreadGroup.add(Unknown Source)
- // // at java.lang.Thread.init(Unknown Source)
- // // at java.lang.Thread.<init>(Unknown Source)
- // // at
- // org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier$4.newThread(CacheClientNotifier.java:321)
- // // at
- // org.apache.edu.oswego.cs.dl.util.concurrent.PooledExecutor.addThread(PooledExecutor.java:512)
- // // at
- // org.apache.edu.oswego.cs.dl.util.concurrent.PooledExecutor.execute(PooledExecutor.java:888)
- // // at
- // org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier.notifyClients(CacheClientNotifier.java:95)
- // // at
- // org.apache.geode.internal.cache.tier.sockets.ServerConnection.run(ServerConnection.java:271)
- //
- // //notifierGroup.setDaemon(true);
- //
- // if (USE_QUEUED_EXECUTOR)
- // createQueuedExecutor(notifierGroup);
- // else
- // createPooledExecutor(notifierGroup);
- // }
-
- // /**
- // * Creates the <code>QueuedExecutor</code> used to deliver messages
- // * to <code>CacheClientProxy</code> instances
- // * @param notifierGroup The <code>ThreadGroup</code> to which the
- // * <code>QueuedExecutor</code>'s <code>Threads</code> belong
- // */
- // protected void createQueuedExecutor(final ThreadGroup notifierGroup)
- // {
- // QueuedExecutor queuedExecutor = new QueuedExecutor(new LinkedQueue());
- // queuedExecutor.setThreadFactory(new ThreadFactory()
- // {
- // public Thread newThread(Runnable command)
- // {
- // Thread thread = new Thread(notifierGroup, command, "Queued Cache Client
- // Notifier");
- // thread.setDaemon(true);
- // return thread;
- // }
- // });
- // _executor = queuedExecutor;
- // }
-
- // /**
- // * Creates the <code>PooledExecutor</code> used to deliver messages
- // * to <code>CacheClientProxy</code> instances
- // * @param notifierGroup The <code>ThreadGroup</code> to which the
- // * <code>PooledExecutor</code>'s <code>Threads</code> belong
- // */
- // protected void createPooledExecutor(final ThreadGroup notifierGroup)
- // {
- // PooledExecutor pooledExecutor = new PooledExecutor(new
- // BoundedLinkedQueue(4096), 50);
- // pooledExecutor.setMinimumPoolSize(10);
- // pooledExecutor.setKeepAliveTime(1000 * 60 * 5);
- // pooledExecutor.setThreadFactory(new ThreadFactory()
- // {
- // public Thread newThread(Runnable command)
- // {
- // Thread thread = new Thread(notifierGroup, command, "Pooled Cache Client
- // Notifier");
- // thread.setDaemon(true);
- // return thread;
- // }
- // });
- // pooledExecutor.createThreads(5);
- // _executor = pooledExecutor;
- // }
-
protected void deliverInterestChange(ClientProxyMembershipID proxyID,
ClientInterestMessageImpl message) {
DM dm = ((InternalDistributedSystem) this.getCache().getDistributedSystem())
@@ -2469,23 +2159,6 @@ public class CacheClientNotifier {
*/
protected static final int ALL_PORTS = -1;
- // /**
- // * Whether to synchonously deliver messages to proxies.
- // * This is currently hard-coded to true to ensure ordering.
- // */
- // protected static final boolean USE_SYNCHRONOUS_NOTIFICATION =
- // true;
- // Boolean.getBoolean("CacheClientNotifier.USE_SYNCHRONOUS_NOTIFICATION");
-
- // /**
- // * Whether to use the <code>QueuedExecutor</code> (or the
- // * <code>PooledExecutor</code>) to deliver messages to proxies.
- // * Currently, delivery is synchronous. No <code>Executor</code> is
- // * used.
- // */
- // protected static final boolean USE_QUEUED_EXECUTOR =
- // Boolean.getBoolean("CacheClientNotifier.USE_QUEUED_EXECUTOR");
-
/**
* The map of known <code>CacheClientProxy</code> instances. Maps ClientProxyMembershipID to
* CacheClientProxy. Note that the keys in this map are not updated when a durable client
@@ -2506,9 +2179,9 @@ public class CacheClientNotifier {
new HashSet<ClientProxyMembershipID>();
/**
- * The GemFire <code>Cache</code>. Note that since this is a singleton class you should not use a
- * direct reference to _cache in CacheClientNotifier code. Instead, you should always use
- * <code>getCache()</code>
+ * The GemFire <code>InternalCache</code>. Note that since this is a singleton class you should
+ * not use a direct reference to _cache in CacheClientNotifier code. Instead, you should always
+ * use <code>getCache()</code>
*/
private InternalCache _cache;
@@ -2541,10 +2214,6 @@ public class CacheClientNotifier {
*/
private volatile HAContainerWrapper haContainer;
- // /**
- // * The singleton <code>CacheClientNotifier</code> instance
- // */
- // protected static CacheClientNotifier _instance;
/**
* The size of the server-to-client communication socket buffers. This can be modified using the
* BridgeServer.SOCKET_BUFFER_SIZE system property.
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
index ca25b6e..728abf7 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
@@ -138,6 +138,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
* The buffer upon which we receive messages
*/
private final ByteBuffer commBuffer;
+
private boolean commBufferReleased;
private final CCUStats stats;
@@ -146,6 +147,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
* Cache for which we provide service
*/
private /* final */ InternalCache cache;
+
private /* final */ CachedRegionHelper cacheHelper;
/**
@@ -202,7 +204,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
* @return true if cache appears
*/
private boolean waitForCache() {
- InternalCache c;
+ InternalCache cache;
long tilt = System.currentTimeMillis() + MAX_CACHE_WAIT * 1000;
for (;;) {
if (quitting()) {
@@ -222,8 +224,8 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
new Object[] {this, MAX_CACHE_WAIT}));
return false;
}
- c = GemFireCacheImpl.getInstance();
- if (c != null && !c.isClosed()) {
+ cache = GemFireCacheImpl.getInstance();
+ if (cache != null && !cache.isClosed()) {
break;
}
boolean interrupted = Thread.interrupted();
@@ -237,8 +239,8 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
}
}
} // for
- this.cache = c;
- this.cacheHelper = new CachedRegionHelper(c);
+ this.cache = cache;
+ this.cacheHelper = new CachedRegionHelper(cache);
return true;
}
@@ -287,7 +289,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
OutputStream tmpOut = null;
InputStream tmpIn = null;
try {
- /** Size of the server-to-client communication socket buffers */
+ // Size of the server-to-client communication socket buffers
int socketBufferSize =
Integer.getInteger("BridgeServer.SOCKET_BUFFER_SIZE", 32768).intValue();
@@ -340,7 +342,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
// create a "server" memberId we currently don't know much about the
// server.
// Would be nice for it to send us its member id
- // @todo - change the serverId to use the endpoint's getMemberId() which
+ // TODO: change the serverId to use the endpoint's getMemberId() which
// returns a
// DistributedMember (once gfecq branch is merged to trunk).
MemberAttributes ma =
@@ -1159,21 +1161,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
// message
if (region.hasServerProxy()) {
return;
-
- // NOTE:
- // As explained in the method description, this code is added as part
- // of CQ bug fix. Cache server team needs to look into changes relating
- // to local region.
- //
- // Locally invalidate the region
- // region.basicBridgeClientInvalidate(callbackArgument,
- // proxy.getProcessedMarker());
-
- // if (logger.debugEnabled()) {
- // logger.debug(toString() + ": Cleared region: " + regionName
- // + " callbackArgument: " + callbackArgument);
- // }
-
}
} catch (Exception e) {
@@ -1212,12 +1199,12 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
// servers recursively
}
- // // CALLBACK TESTING PURPOSE ONLY ////
+ // CALLBACK TESTING PURPOSE ONLY
if (PoolImpl.IS_INSTANTIATOR_CALLBACK) {
ClientServerObserver bo = ClientServerObserverHolder.getInstance();
bo.afterReceivingFromServer(eventId);
}
- // /////////////////////////////////////
+
}
// TODO bug: can the following catch be more specific?
catch (Exception e) {
@@ -1233,7 +1220,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
final boolean isDebugEnabled = logger.isDebugEnabled();
try {
int noOfParts = msg.getNumberOfParts();
- // int numOfClasses = noOfParts - 3; // 1 for ds classname, 1 for ds id and 1 for eventId.
if (isDebugEnabled) {
logger.debug("{}: Received register dataserializer message of parts {}", getName(),
noOfParts);
@@ -1244,8 +1230,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
String dataSerializerClassName =
(String) CacheServerHelper.deserialize(msg.getPart(i).getSerializedForm());
int id = msg.getPart(i + 1).getInt();
- InternalDataSerializer.register(dataSerializerClassName, false, eventId,
- null/* context */, id);
+ InternalDataSerializer.register(dataSerializerClassName, false, eventId, null, id);
// distribute is false because we don't want to propagate this to
// servers recursively
@@ -1266,12 +1251,12 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
}
}
- // // CALLBACK TESTING PURPOSE ONLY ////
+ // CALLBACK TESTING PURPOSE ONLY
if (PoolImpl.IS_INSTANTIATOR_CALLBACK) {
ClientServerObserver bo = ClientServerObserverHolder.getInstance();
bo.afterReceivingFromServer(eventId);
}
- ///////////////////////////////////////
+
}
// TODO bug: can the following catch be more specific?
catch (Exception e) {
@@ -1284,12 +1269,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
/**
* Processes message to invoke CQ listeners.
- *
- * @param startMessagePart
- * @param numCqParts
- * @param messageType
- * @param key
- * @param value
*/
private int processCqs(Message m, int startMessagePart, int numCqParts, int messageType,
Object key, Object value) {
@@ -1299,7 +1278,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
private int processCqs(Message m, int startMessagePart, int numCqParts, int messageType,
Object key, Object value, byte[] delta, EventID eventId) {
- // String[] cqs = new String[numCqs/2];
HashMap cqs = new HashMap();
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -1467,7 +1445,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
}
}
-
private void handleTombstoneOperation(Message msg) {
String regionName = "unknown";
try { // not sure why this isn't done by the caller
@@ -1721,10 +1698,7 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
// originating from the client
// and by updating the last update stat, the ServerMonitor is less
// likely to send pings...
- // and the ClientHealthMonitor will cause a disconnect -- mthomas
- // 10/18/2006
-
- // this._endpoint.setLastUpdate();
+ // and the ClientHealthMonitor will cause a disconnect
} catch (InterruptedIOException e) {
// Per Sun's support web site, this exception seems to be peculiar
@@ -1839,13 +1813,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
return socket.getLocalPort();
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.distributed.internal.InternalDistributedSystem.DisconnectListener#onDisconnect
- * (org.apache.geode.distributed.internal.InternalDistributedSystem)
- */
public void onDisconnect(InternalDistributedSystem sys) {
stopUpdater();
}
@@ -1855,15 +1822,6 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
*/
private volatile boolean endPointDied = false;
- /**
- * Returns true if the end point represented by this updater is considered dead.
- *
- * @return true if {@link #endpoint} died.
- */
- public boolean isEndPointDead() {
- return this.endPointDied;
- }
-
private void verifySocketBufferSize(int requestedBufferSize, int actualBufferSize, String type) {
if (actualBufferSize < requestedBufferSize) {
logger.info(LocalizedMessage.create(
@@ -1944,11 +1902,9 @@ public class CacheClientUpdater extends Thread implements ClientUpdater, Disconn
public long startTime() {
return DistributionStats.getStatTime();
}
-
}
public boolean isProcessing() {
- // TODO Auto-generated method stub
return continueProcessing.get();
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java b/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
index 4cc75aa..7f623c7 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/CacheCreation.java
@@ -145,6 +145,7 @@ import org.apache.geode.internal.logging.LocalLogWriter;
import org.apache.geode.internal.logging.LogWriterFactory;
import org.apache.geode.internal.offheap.MemoryAllocator;
import org.apache.geode.management.internal.JmxManagerAdvisor;
+import org.apache.geode.management.internal.RestAgent;
import org.apache.geode.pdx.PdxInstance;
import org.apache.geode.pdx.PdxInstanceFactory;
import org.apache.geode.pdx.PdxSerializer;
@@ -1022,6 +1023,22 @@ public class CacheCreation implements InternalCache {
return bridge;
}
+ @Override
+ public void setReadSerialized(final boolean value) {
+ throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+ }
+
+ @Override
+ public PdxInstanceFactory createPdxInstanceFactory(final String className,
+ final boolean expectDomainClass) {
+ throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+ }
+
+ @Override
+ public void waitForRegisterInterestsInProgress() {
+ throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+ }
+
void addDeclarableProperties(final Declarable declarable, final Properties properties) {
this.declarablePropertiesMap.put(declarable, properties);
}
@@ -1163,10 +1180,25 @@ public class CacheCreation implements InternalCache {
}
@Override
+ public RestAgent getRestAgent() {
+ throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+ }
+
+ @Override
public Properties getDeclarableProperties(final Declarable declarable) {
throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
}
+ @Override
+ public void setRESTServiceRunning(final boolean isRESTServiceRunning) {
+ throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+ }
+
+ @Override
+ public void close(final String reason, final boolean keepAlive, final boolean keepDS) {
+ throw new UnsupportedOperationException(LocalizedStrings.SHOULDNT_INVOKE.toLocalizedString());
+ }
+
public void addGatewayReceiver(GatewayReceiver receiver) {
this.gatewayReceivers.add(receiver);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/BaseManagementService.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/BaseManagementService.java b/geode-core/src/main/java/org/apache/geode/management/internal/BaseManagementService.java
index 34bb6c0..bc317bc 100755
--- a/geode-core/src/main/java/org/apache/geode/management/internal/BaseManagementService.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/BaseManagementService.java
@@ -21,9 +21,9 @@ import java.util.Map;
import org.apache.logging.log4j.Logger;
-import org.apache.geode.cache.Cache;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.management.ManagementService;
@@ -50,32 +50,28 @@ public abstract class BaseManagementService extends ManagementService {
protected BaseManagementService() {}
// Static block to initialize the ConnectListener on the System
-
static {
initInternalDistributedSystem();
-
}
/**
* This method will close the service. Any operation on the service instance will throw exception
*/
-
protected abstract void close();
/**
* This method will close the service. Any operation on the service instance will throw exception
*/
-
protected abstract boolean isClosed();
/**
* Returns a ManagementService to use for the specified Cache.
- *
+ *
* @param cache defines the scope of resources to be managed
*/
- public static ManagementService getManagementService(Cache cache) {
+ public static ManagementService getManagementService(InternalCache cache) {
synchronized (instances) {
- BaseManagementService service = (BaseManagementService) instances.get(cache);
+ BaseManagementService service = instances.get(cache);
if (service == null) {
service = SystemManagementService.newSystemManagementService(cache);
instances.put(cache, service);
@@ -85,14 +81,13 @@ public abstract class BaseManagementService extends ManagementService {
}
}
- public static ManagementService getExistingManagementService(Cache cache) {
+ public static ManagementService getExistingManagementService(InternalCache cache) {
synchronized (instances) {
BaseManagementService service = (BaseManagementService) instances.get(cache);
return service;
}
}
-
/**
* Initialises the distributed system listener
*/
@@ -125,8 +120,6 @@ public abstract class BaseManagementService extends ManagementService {
/**
* Add an Distributed System and adds a Discon Listener
- *
- * @param sys
*/
private static void addInternalDistributedSystem(InternalDistributedSystem sys) {
synchronized (instances) {
@@ -147,8 +140,6 @@ public abstract class BaseManagementService extends ManagementService {
/**
* Remove a Distributed System from the system lists. If list is empty it closes down all the
* services if not closed
- *
- * @param sys
*/
private static void removeInternalDistributedSystem(InternalDistributedSystem sys) {
synchronized (instances) {
@@ -172,7 +163,6 @@ public abstract class BaseManagementService extends ManagementService {
}
instances.clear();
}
-
}
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java b/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
index da92f69..b6ffa41 100755
--- a/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/FederatingManager.java
@@ -25,6 +25,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
import javax.management.Notification;
import javax.management.ObjectName;
@@ -32,7 +33,6 @@ import javax.management.ObjectName;
import org.apache.logging.log4j.Logger;
import org.apache.geode.cache.AttributesFactory;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EvictionAction;
import org.apache.geode.cache.EvictionAttributes;
@@ -40,11 +40,13 @@ import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.Scope;
+import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.HasCachePerfStats;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.management.ManagementException;
@@ -60,48 +62,38 @@ import org.apache.geode.management.ManagementException;
public class FederatingManager extends Manager {
public static final Logger logger = LogService.getLogger();
-
/**
- *
* This Executor uses a pool of thread to execute the member addition /removal tasks, This will
* utilize the processing powers available. Going with unbounded queue because tasks wont be
* unbounded in practical situation as number of members will be a finite set at any given point
* of time
*/
-
private ExecutorService pooledMembershipExecutor;
-
/**
* Proxy factory is used to create , remove proxies
*/
protected MBeanProxyFactory proxyFactory;
-
/**
* Remote Filter chain for local MBean filters
*/
-
private RemoteFilterChain remoteFilterChain;
private MBeanJMXAdapter jmxAdapter;
private MemberMessenger messenger;
-
private SystemManagementService service;
-
/**
- * Public Constructor
- *
* @param jmxAdapter JMX Adpater
* @param repo Management resource repo
* @param system Internal Distributed System
* @param service SystemManagement Service
*/
public FederatingManager(MBeanJMXAdapter jmxAdapter, ManagementResourceRepo repo,
- InternalDistributedSystem system, SystemManagementService service, Cache cache) {
+ InternalDistributedSystem system, SystemManagementService service, InternalCache cache) {
super(repo, system, cache);
this.remoteFilterChain = new RemoteFilterChain();
this.service = service;
@@ -110,14 +102,12 @@ public class FederatingManager extends Manager {
this.messenger = new MemberMessenger(jmxAdapter, repo, system);
}
-
/**
* This method will be invoked whenever a member wants to be a managing node. The exception
* Management exception has to be handled by the caller.
*/
@Override
public synchronized void startManager() {
-
try {
if (logger.isDebugEnabled()) {
logger.debug("Starting the Federating Manager.... ");
@@ -138,7 +128,6 @@ public class FederatingManager extends Manager {
running = false;
throw new ManagementException(e);
}
-
}
public synchronized void stopManager() {
@@ -151,18 +140,13 @@ public class FederatingManager extends Manager {
logger.debug("Stopping the Federating Manager.... ");
}
stopManagingActivity();
-
}
/**
* This method will be invoked whenever a member stops being a managing node. The exception
* Management exception has to be handled by the caller. *
- *
- * @throws ManagementException
- *
*/
private void stopManagingActivity() {
-
try {
this.pooledMembershipExecutor.shutdownNow();
Iterator<DistributedMember> it = repo.getMonitoringRegionMap().keySet().iterator();
@@ -173,10 +157,7 @@ public class FederatingManager extends Manager {
} catch (Exception e) {
throw new ManagementException(e);
- } finally {
- // For future use
}
-
}
@Override
@@ -190,8 +171,6 @@ public class FederatingManager extends Manager {
*
* This method will delegate task to another thread and exit, so that it wont block the membership
* listener
- *
- * @param member
*/
public void addMember(DistributedMember member) {
GIITask giiTask = new GIITask(member);
@@ -203,33 +182,22 @@ public class FederatingManager extends Manager {
});
}
-
/**
* This method will be invoked from MembershipListener which is registered when the member becomes
* a Management node.
*
* This method will delegate task to another thread and exit, so that it wont block the membership
* listener
- *
- * @param member
*/
public void removeMember(DistributedMember member, boolean crashed) {
RemoveMemberTask removeTask = new RemoveMemberTask(member, crashed);
executeTask(removeTask);
}
- private void submitTask(Callable<DistributedMember> task) {
- try {
- pooledMembershipExecutor.submit(task);
- } catch (java.util.concurrent.RejectedExecutionException ex) {
- // Ignore, we are getting shutdown
- }
- }
-
private void executeTask(Runnable task) {
try {
pooledMembershipExecutor.execute(task);
- } catch (java.util.concurrent.RejectedExecutionException ex) {
+ } catch (RejectedExecutionException ex) {
// Ignore, we are getting shutdown
}
}
@@ -278,22 +246,16 @@ public class FederatingManager extends Manager {
*
* this method will delegate task to another thread and exit, so that it wont block the membership
* listener
- *
- * @param member
- * @param reason TODO
*/
public void suspectMember(DistributedMember member, InternalDistributedMember whoSuspected,
String reason) {
service.memberSuspect((InternalDistributedMember) member, whoSuspected, reason);
-
}
/**
* This method will be invoked when a node transitions from managed node to managing node This
* method will block for all GIIs to be completed But each GII is given a specific time frame.
* After that the task will be marked as cancelled.
- *
- * @throws InterruptedException
*/
public void startManagingActivity() throws Exception {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -304,7 +266,6 @@ public class FederatingManager extends Manager {
Iterator<DistributedMember> it = members.iterator();
DistributedMember member;
-
final List<Callable<DistributedMember>> giiTaskList = new ArrayList<>();
List<Future<DistributedMember>> futureTaskList;
@@ -354,9 +315,7 @@ public class FederatingManager extends Manager {
logger.debug("InterruptedException while creating Monitoring resource with error : {}",
mgEx.getMessage(), mgEx);
}
-
}
-
}
} catch (InterruptedException e) {
if (isDebugEnabled) {
@@ -370,20 +329,14 @@ public class FederatingManager extends Manager {
logger.debug("Management Resource creation completed");
}
}
-
}
-
-
/**
* Actual task of doing the GII
*
* It will perform the GII request which might originate from TranstionListener or Membership
* Listener.
- *
- *
- *
- *
+ *
* Managing Node side resources are created per member which is visible to this node
*
* 1)Management Region : its a Replicated NO_ACK region 2)Notification Region : its a Replicated
@@ -394,10 +347,7 @@ public class FederatingManager extends Manager {
*
* This task can be cancelled from the calling thread if a timeout happens. In that case we have
* to handle the thread interrupt
- *
- *
*/
-
private class GIITask implements Callable<DistributedMember> {
private DistributedMember member;
@@ -438,7 +388,6 @@ public class FederatingManager extends Manager {
ManagementCacheListener mgmtCacheListener = new ManagementCacheListener(proxyFactory);
monitorAttrFactory.addCacheListener(mgmtCacheListener);
-
RegionAttributes<String, Object> monitoringRegionAttrs = monitorAttrFactory.create();
// Notification region for member is created
@@ -472,8 +421,7 @@ public class FederatingManager extends Manager {
monitoringRegionAttrs, internalArgs);
proxyMonitoringRegionCreated = true;
-
- } catch (org.apache.geode.cache.TimeoutException e) {
+ } catch (TimeoutException e) {
if (logger.isDebugEnabled()) {
logger.debug("Error During Internal Region creation {}", e.getMessage(), e);
}
@@ -504,7 +452,7 @@ public class FederatingManager extends Manager {
cache.createVMRegion(ManagementConstants.NOTIFICATION_REGION + "_" + appender,
notifRegionAttrs, internalArgs);
proxyNotifRegionCreated = true;
- } catch (org.apache.geode.cache.TimeoutException e) {
+ } catch (TimeoutException e) {
if (logger.isDebugEnabled()) {
logger.debug("Error During Internal Region creation {}", e.getMessage(), e);
}
@@ -530,7 +478,6 @@ public class FederatingManager extends Manager {
// region is not created
proxyMonitoringRegion.localDestroyRegion();
}
-
}
if (logger.isDebugEnabled()) {
@@ -560,25 +507,20 @@ public class FederatingManager extends Manager {
throw new ManagementException(e);
}
-
}
} catch (Exception e) {
throw new ManagementException(e);
}
-
// Before completing task intimate all listening ProxyListener which might send notifications.
service.memberJoined((InternalDistributedMember) member);
// Send manager info to the added member
messenger.sendManagerInfo(member);
-
return member;
-
}
-
}
/**
@@ -622,10 +564,8 @@ public class FederatingManager extends Manager {
return proxyFactory.findAllProxies(member);
}
-
public MemberMessenger getMessenger() {
return messenger;
}
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/LocalManager.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/LocalManager.java b/geode-core/src/main/java/org/apache/geode/management/internal/LocalManager.java
index 3adbce0..b5b97ca 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/LocalManager.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/LocalManager.java
@@ -17,7 +17,6 @@ package org.apache.geode.management.internal;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -36,14 +35,15 @@ import org.apache.geode.CancelException;
import org.apache.geode.GemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.cache.AttributesFactory;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionExistsException;
import org.apache.geode.cache.Scope;
+import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.CachePerfStats;
import org.apache.geode.internal.cache.HasCachePerfStats;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalRegionArguments;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.LoggingThreadGroup;
@@ -55,11 +55,7 @@ import org.apache.geode.management.ManagementException;
* a) Handles proxy creation when Management node comes up b) Handles proxy creation when a member
* joins c) Remove proxies when a member leaves or node stops being management node. d) Takes care
* to create resources like hidden regions for MBean and notification federation.
- *
- *
- *
*/
-
public class LocalManager extends Manager {
private static final Logger logger = LogService.getLogger();
@@ -74,27 +70,22 @@ public class LocalManager extends Manager {
*/
protected ScheduledExecutorService singleThreadFederationScheduler;
-
/**
* This map holds all the components which are eligible for federation. Although filters might
* prevent any of the component from getting federated.
*/
-
private Map<ObjectName, FederationComponent> federatedComponentMap;
-
private Object lock = new Object();
private SystemManagementService service;
/**
- * Public constructor
- *
* @param repo management resource repo
* @param system internal distributed system
*/
public LocalManager(ManagementResourceRepo repo, InternalDistributedSystem system,
- SystemManagementService service, Cache cache) {
+ SystemManagementService service, InternalCache cache) {
super(repo, system, cache);
this.service = service;
this.federatedComponentMap = new ConcurrentHashMap<ObjectName, FederationComponent>();
@@ -105,7 +96,6 @@ public class LocalManager extends Manager {
*
* Management Region : its a Replicated NO_ACK region Notification Region : its a Replicated Proxy
* NO_ACK region
- *
*/
private void startLocalManagement(Map<ObjectName, FederationComponent> federatedComponentMap) {
@@ -137,9 +127,9 @@ public class LocalManager extends Manager {
logger.debug("Creating Management Region :");
}
- /**
+ /*
* Sharing the same Internal Argument for both notification region and monitoring region
- **/
+ */
InternalRegionArguments internalArgs = new InternalRegionArguments();
internalArgs.setIsUsedForMetaRegion(true);
@@ -184,7 +174,7 @@ public class LocalManager extends Manager {
monitoringRegionAttrs, internalArgs));
monitoringRegionCreated = true;
- } catch (org.apache.geode.cache.TimeoutException e) {
+ } catch (TimeoutException e) {
throw new ManagementException(e);
} catch (RegionExistsException e) {
throw new ManagementException(e);
@@ -199,7 +189,7 @@ public class LocalManager extends Manager {
cache.createVMRegion(ManagementConstants.NOTIFICATION_REGION + "_" + appender,
notifRegionAttrs, internalArgs));
notifRegionCreated = true;
- } catch (org.apache.geode.cache.TimeoutException e) {
+ } catch (TimeoutException e) {
throw new ManagementException(e);
} catch (RegionExistsException e) {
throw new ManagementException(e);
@@ -212,7 +202,6 @@ public class LocalManager extends Manager {
repo.getLocalMonitoringRegion().localDestroyRegion();
}
-
}
managementTask = new ManagementTask(federatedComponentMap);
@@ -230,11 +219,8 @@ public class LocalManager extends Manager {
logger.debug("Notification Region created with Name : {}",
repo.getLocalNotificationRegion().getName());
}
-
}
-
}
-
}
public void markForFederation(ObjectName objName, FederationComponent fedComp) {
@@ -251,23 +237,18 @@ public class LocalManager extends Manager {
// To delete an entry from the region
repo.getLocalMonitoringRegion().remove(objName.toString());
}
-
}
-
}
/**
* This method will shutdown various tasks running for management
*/
private void shutdownTasks() {
-
// No need of pooledGIIExecutor as this node wont do GII again
// so better to release resources
if (this.singleThreadFederationScheduler != null) {
- List<Runnable> l = this.singleThreadFederationScheduler.shutdownNow();
-
+ this.singleThreadFederationScheduler.shutdownNow();
}
-
}
/**
@@ -294,7 +275,6 @@ public class LocalManager extends Manager {
logger.debug("Unable to clean MBean: {} due to {}", objName, e.getMessage(), e);
}
}
-
}
repo.destroyLocalMonitoringRegion();
}
@@ -330,11 +310,7 @@ public class LocalManager extends Manager {
* guaranteed to execute sequentially, and no more than one task will be active at any given time.
* Unlike the otherwise equivalent <tt>newScheduledThreadPool(1)</tt> the returned executor is
* guaranteed not to be reconfigurable to use additional threads.
- *
- *
- *
*/
-
private class ManagementTask implements Runnable {
private Map<String, FederationComponent> replicaMap;
@@ -346,12 +322,10 @@ public class LocalManager extends Manager {
@Override
public void run() {
-
if (logger.isTraceEnabled()) {
logger.trace("Federation started at managed node : ");
}
-
try {
synchronized (lock) {
replicaMap.clear();
@@ -380,7 +354,6 @@ public class LocalManager extends Manager {
replicaMap.put(key, fedCompInstance);
}
}
-
}
}
@@ -434,13 +407,11 @@ public class LocalManager extends Manager {
@Override
public void stopManager() {
- // Shutting down the GII executor as this node wont require
- // it anymore
+ // Shutting down the GII executor as this node wont require it anymore
shutdownTasks();
// Clean up management Resources
cleanUpResources();
running = false;
-
}
public void stopCacheOps() {
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/beans/BeanUtilFuncs.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/beans/BeanUtilFuncs.java b/geode-core/src/main/java/org/apache/geode/management/internal/beans/BeanUtilFuncs.java
index 6861a7d..499ef01 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/beans/BeanUtilFuncs.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/beans/BeanUtilFuncs.java
@@ -26,10 +26,10 @@ import java.util.Set;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionConfig;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.management.GemFireProperties;
import org.apache.geode.management.internal.cli.CliUtil;
@@ -41,9 +41,7 @@ public class BeanUtilFuncs {
/**
* returns the tail of the log file
*
- * @param logFile
* @return tail of the log file
- * @throws IOException
*/
public static String tailSystemLog(File logFile, final int numLines) throws IOException {
if (logFile == null || logFile.equals(new File(""))) {
@@ -107,11 +105,8 @@ public class BeanUtilFuncs {
return returnStr.toString();
}
-
/**
- * @param sc
* @return tail of log
- * @throws IOException
*/
public static String tailSystemLog(DistributionConfig sc, final int numLines) throws IOException {
File logFile = sc.getLogFile();
@@ -128,7 +123,7 @@ public class BeanUtilFuncs {
DistributedMember memberFound = null;
if (memberNameOrId != null) {
- Cache cache = CacheFactory.getAnyInstance();
+ InternalCache cache = (InternalCache) CacheFactory.getAnyInstance();
Set<DistributedMember> memberSet = CliUtil.getAllMembers(cache);
for (DistributedMember member : memberSet) {
if (memberNameOrId.equals(member.getId()) || memberNameOrId.equals(member.getName())) {
@@ -141,12 +136,8 @@ public class BeanUtilFuncs {
}
public static GemFireProperties initGemfireProperties(DistributionConfig config) {
-
- // **TODO **/
String memberGroups = "";
- // **TODO **/
String configFile = null;
- // ** TODO **//
String includeFile = null;
GemFireProperties gemFirePropertyData = new GemFireProperties();
@@ -329,17 +320,14 @@ public class BeanUtilFuncs {
gemFirePropertyData.setSSLDefaultAlias(config.getSSLDefaultAlias());
return gemFirePropertyData;
-
}
-
/**
* Compresses a given String. It is encoded using ISO-8859-1, So any decompression of the
* compressed string should also use ISO-8859-1
*
* @param str String to be compressed.
* @return compressed bytes
- * @throws IOException
*/
public static byte[] compress(String str) throws IOException {
if (str == null || str.length() == 0) {
@@ -356,7 +344,6 @@ public class BeanUtilFuncs {
/**
* @param bytes bytes to be decompressed
* @return a decompressed String
- * @throws IOException
*/
public static String decompress(byte[] bytes) throws IOException {
if (bytes == null || bytes.length == 0) {
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/beans/QueryDataFunction.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/beans/QueryDataFunction.java b/geode-core/src/main/java/org/apache/geode/management/internal/beans/QueryDataFunction.java
index efd7436..705ba4d 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/beans/QueryDataFunction.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/beans/QueryDataFunction.java
@@ -30,7 +30,6 @@ import java.util.regex.Pattern;
import org.apache.logging.log4j.Logger;
import org.apache.geode.SystemFailure;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.Region;
@@ -51,6 +50,7 @@ import org.apache.geode.cache.query.internal.QCompiler;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.internal.InternalEntity;
import org.apache.geode.internal.cache.BucketRegion;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalDataSet;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.PartitionedRegionHelper;
@@ -127,7 +127,7 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
private QueryDataFunctionResult selectWithType(final FunctionContext context, String queryString,
final boolean showMember, final String regionName, final int limit,
final int queryResultSetLimit, final int queryCollectionsDepth) throws Exception {
- Cache cache = CacheFactory.getAnyInstance();
+ InternalCache cache = getCache();
Function loclQueryFunc = new LocalQueryFunction("LocalQueryFunction", regionName, showMember)
.setOptimizeForWrite(true);
queryString = applyLimitClause(queryString, limit, queryResultSetLimit);
@@ -293,12 +293,11 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
List<String> decompressedList = new ArrayList<String>();
while (it.hasNext()) {
-
String decompressedStr = null;
decompressedStr = BeanUtilFuncs.decompress(it.next().compressedBytes);
decompressedList.add(decompressedStr);
-
}
+
if (zipResult) {
return BeanUtilFuncs.compress(wrapResult(decompressedList.toString()));
} else {
@@ -359,7 +358,7 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
}
}
- Cache cache = CacheFactory.getAnyInstance();
+ InternalCache cache = (InternalCache) CacheFactory.getAnyInstance();
try {
SystemManagementService service =
@@ -386,7 +385,6 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
}
}
}
-
}
} else {
return new JsonisedErroMessage(ManagementStrings.QUERY__MSG__INVALID_QUERY
@@ -449,6 +447,10 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
}
}
+ private InternalCache getCache() {
+ return (InternalCache) CacheFactory.getAnyInstance();
+ }
+
private static class JsonisedErroMessage {
private static String message = "message";
@@ -467,7 +469,6 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
public String toString() {
return gFJsonObject.toString();
}
-
}
/**
@@ -479,7 +480,7 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
*
* @return a set of regions involved in the query
*/
- private static Set<String> compileQuery(final Cache cache, final String query)
+ private static Set<String> compileQuery(final InternalCache cache, final String query)
throws QueryInvalidException {
QCompiler compiler = new QCompiler();
Set<String> regionsInQuery = null;
@@ -498,7 +499,7 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
/**
* Function to gather data locally. This function is required to execute query with region context
*/
- private class LocalQueryFunction extends FunctionAdapter {
+ private static class LocalQueryFunction extends FunctionAdapter {
private static final long serialVersionUID = 1L;
@@ -537,7 +538,7 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
@Override
public void execute(final FunctionContext context) {
- Cache cache = CacheFactory.getAnyInstance();
+ InternalCache cache = getCache();
QueryService queryService = cache.getQueryService();
String qstr = (String) context.getArguments();
Region r = cache.getRegion(regionName);
@@ -557,6 +558,10 @@ public class QueryDataFunction extends FunctionAdapter implements InternalEntity
}
}
+ private InternalCache getCache() {
+ return (InternalCache) CacheFactory.getAnyInstance();
+ }
+
@Override
public String getId() {
return this.id;
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ClientCommands.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ClientCommands.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ClientCommands.java
index 3d155e5..18940fc 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ClientCommands.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/ClientCommands.java
@@ -12,13 +12,12 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.management.internal.cli.commands;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.distributed.DistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.management.CacheServerMXBean;
import org.apache.geode.management.ClientHealthStatus;
import org.apache.geode.management.ManagementService;
@@ -53,10 +52,8 @@ import java.util.Set;
import javax.management.ObjectName;
/**
- *
* @since GemFire 8.0
*/
-
public class ClientCommands implements CommandMarker {
private Gfsh getGfsh() {
@@ -77,7 +74,7 @@ public class ClientCommands implements CommandMarker {
String headerText = "ClientList";
resultTable = resultTable.setHeader(headerText);
- Cache cache = CacheFactory.getAnyInstance();
+ InternalCache cache = getCache();
ManagementService service = ManagementService.getExistingManagementService(cache);
ObjectName[] cacheServers = service.getDistributedSystemMXBean().listCacheServerObjectNames();
@@ -151,6 +148,9 @@ public class ClientCommands implements CommandMarker {
return result;
}
+ private InternalCache getCache() {
+ return (InternalCache) CacheFactory.getAnyInstance();
+ }
@CliCommand(value = CliStrings.DESCRIBE_CLIENT, help = CliStrings.DESCRIBE_CLIENT__HELP)
@CliMetaData(relatedTopic = {CliStrings.TOPIC_CLIENT})
@@ -174,7 +174,7 @@ public class ClientCommands implements CommandMarker {
try {
CompositeResultData compositeResultData = ResultBuilder.createCompositeResultData();
SectionResultData sectionResult = compositeResultData.addSection("InfoSection");
- Cache cache = CacheFactory.getAnyInstance();
+ InternalCache cache = getCache();
ManagementService service = ManagementService.getExistingManagementService(cache);
ObjectName[] cacheServers = service.getDistributedSystemMXBean().listCacheServerObjectNames();
@@ -204,7 +204,6 @@ public class ClientCommands implements CommandMarker {
}
}
}
-
}
if (clientHealthStatus == null) {
@@ -225,12 +224,7 @@ public class ClientCommands implements CommandMarker {
for (int i = 0; i < resultList.size(); i++) {
try {
Object object = resultList.get(i);
- if (object instanceof Exception) {
- LogWrapper.getInstance().warning(
- "Exception in Describe Client " + ((Throwable) object).getMessage(),
- ((Throwable) object));
- continue;
- } else if (object instanceof Throwable) {
+ if (object instanceof Throwable) {
LogWrapper.getInstance().warning(
"Exception in Describe Client " + ((Throwable) object).getMessage(),
((Throwable) object));
@@ -259,7 +253,6 @@ public class ClientCommands implements CommandMarker {
secondaryServers.add(objectResult.secondaryServer);
}
}
-
}
} catch (Exception e) {
LogWrapper.getInstance().info(CliStrings.DESCRIBE_CLIENT_ERROR_FETCHING_STATS_0 + " :: "
@@ -354,7 +347,6 @@ public class ClientCommands implements CommandMarker {
}
}
}
-
}
@CliAvailabilityIndicator({CliStrings.LIST_CLIENTS, CliStrings.DESCRIBE_CLIENT})
http://git-wip-us.apache.org/repos/asf/geode/blob/db79d223/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/DataCommands.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/DataCommands.java b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/DataCommands.java
index 46ee2b0..36197cd 100644
--- a/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/DataCommands.java
+++ b/geode-core/src/main/java/org/apache/geode/management/internal/cli/commands/DataCommands.java
@@ -1132,7 +1132,7 @@ public class DataCommands implements CommandMarker {
}
Object[] arguments = new Object[] {query, stepName, interactive};
- CLIStep exec = new DataCommandFunction.SelectExecStep(arguments);
+ CLIStep exec = new DataCommandFunction.SelectExecStep(arguments, getCache());
CLIStep display = new DataCommandFunction.SelectDisplayStep(arguments);
CLIStep move = new DataCommandFunction.SelectMoveStep(arguments);
CLIStep quit = new DataCommandFunction.SelectQuitStep(arguments);