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 2017/04/25 18:29:37 UTC
[01/11] geode git commit: GEODE-2796: ClassPathLoader does not blow
up with null TCCL
Repository: geode
Updated Branches:
refs/heads/feature/GEODE-2801 232a61e67 -> ab47b200e
GEODE-2796: ClassPathLoader does not blow up with null TCCL
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/48d662aa
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/48d662aa
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/48d662aa
Branch: refs/heads/feature/GEODE-2801
Commit: 48d662aab7ae4110f36a05aadefe9c12b48c1293
Parents: aaef124
Author: Jared Stewart <js...@pivotal.io>
Authored: Tue Apr 18 14:08:19 2017 -0700
Committer: Jared Stewart <js...@pivotal.io>
Committed: Wed Apr 19 20:08:04 2017 -0700
----------------------------------------------------------------------
.../apache/geode/internal/ClassPathLoader.java | 10 +++++++--
.../ClassPathLoaderIntegrationTest.java | 22 +++++++++++++++++---
2 files changed, 27 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/48d662aa/geode-core/src/main/java/org/apache/geode/internal/ClassPathLoader.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/ClassPathLoader.java b/geode-core/src/main/java/org/apache/geode/internal/ClassPathLoader.java
index 41cce05..39e9d62 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/ClassPathLoader.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/ClassPathLoader.java
@@ -32,6 +32,7 @@ import java.net.URLClassLoader;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
/**
* The delegating <tt>ClassLoader</tt> used by GemFire to load classes and other resources. This
@@ -303,10 +304,15 @@ public final class ClassPathLoader {
ArrayList<ClassLoader> classLoaders = new ArrayList<>();
if (!excludeTCCL) {
- classLoaders.add(Thread.currentThread().getContextClassLoader());
+ ClassLoader tccl = Thread.currentThread().getContextClassLoader();
+ if (tccl != null) {
+ classLoaders.add(tccl);
+ }
}
- classLoaders.add(classLoaderForDeployedJars);
+ if (classLoaderForDeployedJars != null) {
+ classLoaders.add(classLoaderForDeployedJars);
+ }
return classLoaders;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/48d662aa/geode-core/src/test/java/org/apache/geode/internal/ClassPathLoaderIntegrationTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/ClassPathLoaderIntegrationTest.java b/geode-core/src/test/java/org/apache/geode/internal/ClassPathLoaderIntegrationTest.java
index c783318..2a3a7dd 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/ClassPathLoaderIntegrationTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/ClassPathLoaderIntegrationTest.java
@@ -91,6 +91,24 @@ public class ClassPathLoaderIntegrationTest {
ClassPathLoader.setLatestToDefault(temporaryFolder.getRoot());
}
+ @Test
+ public void testClassLoaderWithNullTccl() throws IOException, ClassNotFoundException {
+ // GEODE-2796
+ Thread.currentThread().setContextClassLoader(null);
+ String jarName = "JarDeployerIntegrationTest.jar";
+
+ String classAResource = "integration/parent/ClassA.class";
+
+ String classAName = "integration.parent.ClassA";
+
+ byte[] firstJarBytes = createJarWithClass("ClassA");
+
+ // First deploy of the JAR file
+ ClassPathLoader.getLatest().getJarDeployer().deploy(jarName, firstJarBytes).getFile();
+
+ assertThatClassCanBeLoaded(classAName);
+ assertThatResourceCanBeLoaded(classAResource);
+ }
@Test
public void testDeployFileAndChange() throws IOException, ClassNotFoundException {
@@ -258,13 +276,11 @@ public class ClassPathLoaderIntegrationTest {
List<String> result = (List<String>) execution.execute("MyFunction").getResult();
assertThat(result.get(0)).isEqualTo("Version1");
-
ClassPathLoader.getLatest().getJarDeployer().deploy("MyJar.jar",
FileUtils.readFileToByteArray(jarVersion2));
result = (List<String>) execution.execute("MyFunction").getResult();
assertThat(result.get(0)).isEqualTo("Version2");
-
serverStarterRule.after();
}
@@ -451,7 +467,7 @@ public class ClassPathLoaderIntegrationTest {
/**
* Currently unused but potentially useful for some future test. This causes this loader to only
* generate a class that the parent could not find.
- *
+ *
* @param parent the parent class loader to check with first
*/
@SuppressWarnings("unused")
[09/11] geode git commit: GEODE-2808 - Fixing lock ordering issues in
DeltaSession
Posted by ds...@apache.org.
GEODE-2808 - Fixing lock ordering issues in DeltaSession
Region expiration of sessions and explicit expiration of sessions had
lock ordering issues. Fixing the code so that expiration goes through
the region entry lock first, before getting the lock on StandardSession.
Adding a workaround for the fact that liferay calls removeAttribute
from within session expiration by ignoreing remoteAttribute calls during
expiration.
This closes #472
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/45dc6744
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/45dc6744
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/45dc6744
Branch: refs/heads/feature/GEODE-2801
Commit: 45dc6744154a08986e833852ef743af5d8bf19ba
Parents: 47d8c82
Author: Dan Smith <up...@apache.org>
Authored: Fri Apr 21 11:36:24 2017 -0700
Committer: Dan Smith <up...@apache.org>
Committed: Fri Apr 21 16:23:24 2017 -0700
----------------------------------------------------------------------
.../modules/session/catalina/DeltaSession7.java | 14 +++++++-
.../modules/session/catalina/DeltaSession8.java | 14 +++++++-
.../session/TestSessionsTomcat8Base.java | 34 ++++++++++++++++++++
.../modules/session/catalina/DeltaSession.java | 14 +++++++-
.../geode/modules/session/CommandServlet.java | 4 +++
.../geode/modules/session/QueryCommand.java | 2 ++
.../geode/modules/session/TestSessionsBase.java | 34 ++++++++++++++++++++
7 files changed, 113 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules-tomcat7/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession7.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules-tomcat7/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession7.java b/extensions/geode-modules-tomcat7/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession7.java
index 204ff5e..d7f30bd 100644
--- a/extensions/geode-modules-tomcat7/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession7.java
+++ b/extensions/geode-modules-tomcat7/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession7.java
@@ -263,6 +263,9 @@ public class DeltaSession7 extends StandardSession
public void removeAttribute(String name, boolean notify) {
checkBackingCacheAvailable();
+ if (expired) {
+ return;
+ }
synchronized (this.changeLock) {
// Remove the attribute locally
super.removeAttribute(name, true);
@@ -322,7 +325,7 @@ public class DeltaSession7 extends StandardSession
setExpired(true);
// Do expire processing
- expire();
+ super.expire(true);
// Update statistics
if (manager != null) {
@@ -330,6 +333,15 @@ public class DeltaSession7 extends StandardSession
}
}
+ @Override
+ public void expire(boolean notify) {
+ if (notify) {
+ getOperatingRegion().destroy(this.getId(), this);
+ } else {
+ super.expire(false);
+ }
+ }
+
public void setMaxInactiveInterval(int interval) {
super.setMaxInactiveInterval(interval);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules-tomcat8/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession8.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules-tomcat8/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession8.java b/extensions/geode-modules-tomcat8/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession8.java
index b5e7d0c..f69382a 100644
--- a/extensions/geode-modules-tomcat8/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession8.java
+++ b/extensions/geode-modules-tomcat8/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession8.java
@@ -258,6 +258,9 @@ public class DeltaSession8 extends StandardSession
public void removeAttribute(String name, boolean notify) {
checkBackingCacheAvailable();
+ if (expired) {
+ return;
+ }
synchronized (this.changeLock) {
// Remove the attribute locally
super.removeAttribute(name, true);
@@ -317,7 +320,7 @@ public class DeltaSession8 extends StandardSession
setExpired(true);
// Do expire processing
- expire();
+ super.expire(true);
// Update statistics
if (manager != null) {
@@ -325,6 +328,15 @@ public class DeltaSession8 extends StandardSession
}
}
+ @Override
+ public void expire(boolean notify) {
+ if (notify) {
+ getOperatingRegion().destroy(this.getId(), this);
+ } else {
+ super.expire(false);
+ }
+ }
+
public void setMaxInactiveInterval(int interval) {
super.setMaxInactiveInterval(interval);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules-tomcat8/src/test/java/org/apache/geode/modules/session/TestSessionsTomcat8Base.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules-tomcat8/src/test/java/org/apache/geode/modules/session/TestSessionsTomcat8Base.java b/extensions/geode-modules-tomcat8/src/test/java/org/apache/geode/modules/session/TestSessionsTomcat8Base.java
index 15b3874..1dc1d8b 100644
--- a/extensions/geode-modules-tomcat8/src/test/java/org/apache/geode/modules/session/TestSessionsTomcat8Base.java
+++ b/extensions/geode-modules-tomcat8/src/test/java/org/apache/geode/modules/session/TestSessionsTomcat8Base.java
@@ -233,6 +233,40 @@ public abstract class TestSessionsTomcat8Base extends JUnit4DistributedTestCase
}
/**
+ * Test expiration of a session by the tomcat container, rather than gemfire expiration
+ */
+ @Test
+ public void testSessionExpirationByContainer() throws Exception {
+
+ String key = "value_testSessionExpiration1";
+ String value = "Foo";
+
+ WebConversation wc = new WebConversation();
+ WebRequest req = new GetMethodWebRequest(String.format("http://localhost:%d/test", port));
+
+ // Set an attribute
+ req.setParameter("cmd", QueryCommand.SET.name());
+ req.setParameter("param", key);
+ req.setParameter("value", value);
+ WebResponse response = wc.getResponse(req);
+
+ // Set the session timeout of this one session.
+ req.setParameter("cmd", QueryCommand.SET_MAX_INACTIVE.name());
+ req.setParameter("value", "1");
+ response = wc.getResponse(req);
+
+ // Wait until the session should expire
+ Thread.sleep(2000);
+
+ // Do a request, which should cause the session to be expired
+ req.setParameter("cmd", QueryCommand.GET.name());
+ req.setParameter("param", key);
+ response = wc.getResponse(req);
+
+ assertEquals("", response.getText());
+ }
+
+ /**
* Test that removing a session attribute also removes it from the region
*/
@Test
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession.java b/extensions/geode-modules/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession.java
index bc421a5..ac612da 100644
--- a/extensions/geode-modules/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession.java
+++ b/extensions/geode-modules/src/main/java/org/apache/geode/modules/session/catalina/DeltaSession.java
@@ -266,6 +266,9 @@ public class DeltaSession extends StandardSession
public void removeAttribute(String name, boolean notify) {
checkBackingCacheAvailable();
+ if (expired) {
+ return;
+ }
synchronized (this.changeLock) {
// Remove the attribute locally
super.removeAttribute(name, true);
@@ -325,7 +328,7 @@ public class DeltaSession extends StandardSession
setExpired(true);
// Do expire processing
- expire();
+ super.expire(true);
// Update statistics
if (manager != null) {
@@ -333,6 +336,15 @@ public class DeltaSession extends StandardSession
}
}
+ @Override
+ public void expire(boolean notify) {
+ if (notify) {
+ getOperatingRegion().destroy(this.getId(), this);
+ } else {
+ super.expire(false);
+ }
+ }
+
public void setMaxInactiveInterval(int interval) {
super.setMaxInactiveInterval(interval);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/CommandServlet.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/CommandServlet.java b/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/CommandServlet.java
index 3fede62..a04194b 100644
--- a/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/CommandServlet.java
+++ b/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/CommandServlet.java
@@ -60,6 +60,10 @@ public class CommandServlet extends HttpServlet {
session = request.getSession();
session.setAttribute(param, value);
break;
+ case SET_MAX_INACTIVE:
+ session = request.getSession();
+ session.setMaxInactiveInterval(Integer.valueOf(value));
+ break;
case GET:
session = request.getSession();
String val = (String) session.getAttribute(param);
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/QueryCommand.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/QueryCommand.java b/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/QueryCommand.java
index 2b89e68..622866e 100644
--- a/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/QueryCommand.java
+++ b/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/QueryCommand.java
@@ -21,6 +21,8 @@ public enum QueryCommand {
SET,
+ SET_MAX_INACTIVE,
+
GET,
INVALIDATE,
http://git-wip-us.apache.org/repos/asf/geode/blob/45dc6744/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/TestSessionsBase.java
----------------------------------------------------------------------
diff --git a/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/TestSessionsBase.java b/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/TestSessionsBase.java
index d7674dd..a6bec6c 100644
--- a/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/TestSessionsBase.java
+++ b/extensions/geode-modules/src/test/java/org/apache/geode/modules/session/TestSessionsBase.java
@@ -267,6 +267,40 @@ public abstract class TestSessionsBase {
}
/**
+ * Test expiration of a session by the tomcat container, rather than gemfire expiration
+ */
+ @Test
+ public void testSessionExpirationByContainer() throws Exception {
+
+ String key = "value_testSessionExpiration1";
+ String value = "Foo";
+
+ WebConversation wc = new WebConversation();
+ WebRequest req = new GetMethodWebRequest(String.format("http://localhost:%d/test", port));
+
+ // Set an attribute
+ req.setParameter("cmd", QueryCommand.SET.name());
+ req.setParameter("param", key);
+ req.setParameter("value", value);
+ WebResponse response = wc.getResponse(req);
+
+ // Set the session timeout of this one session.
+ req.setParameter("cmd", QueryCommand.SET_MAX_INACTIVE.name());
+ req.setParameter("value", "1");
+ response = wc.getResponse(req);
+
+ // Wait until the session should expire
+ Thread.sleep(2000);
+
+ // Do a request, which should cause the session to be expired
+ req.setParameter("cmd", QueryCommand.GET.name());
+ req.setParameter("param", key);
+ response = wc.getResponse(req);
+
+ assertEquals("", response.getText());
+ }
+
+ /**
* Test that removing a session attribute also removes it from the region
*/
@Test
[08/11] geode git commit: GEODE-2632: refactor code to reduce
GemFireCacheImpl dependencies
Posted by ds...@apache.org.
GEODE-2632: refactor code to reduce GemFireCacheImpl dependencies
* extract fetching GemFireCacheImpl to Provider interface/class
* use InternalCache instead of casting to Impl
* delete useless javadocs and comments
* reduce scope of constants, vars and methods
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/47d8c820
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/47d8c820
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/47d8c820
Branch: refs/heads/feature/GEODE-2801
Commit: 47d8c82036a9863c9fa7c9142c170c9f8552abb4
Parents: 60ec931
Author: Kirk Lund <kl...@apache.org>
Authored: Wed Apr 19 09:56:24 2017 -0700
Committer: Kirk Lund <kl...@apache.org>
Committed: Fri Apr 21 14:47:50 2017 -0700
----------------------------------------------------------------------
.../web/controllers/AbstractBaseController.java | 194 +++++++++----------
.../web/controllers/BaseControllerAdvice.java | 28 ++-
.../web/controllers/CommonCrudController.java | 29 +--
.../controllers/FunctionAccessController.java | 39 ++--
.../web/controllers/PdxBasedCrudController.java | 28 ++-
.../web/controllers/QueryAccessController.java | 49 ++---
.../web/controllers/support/CacheProvider.java | 22 +++
.../controllers/support/CacheProviderImpl.java | 29 +++
8 files changed, 219 insertions(+), 199 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/AbstractBaseController.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/AbstractBaseController.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/AbstractBaseController.java
index 68080a3..d8eb572 100644
--- a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/AbstractBaseController.java
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/AbstractBaseController.java
@@ -12,15 +12,44 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.rest.internal.web.controllers;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URLDecoder;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.annotation.PostConstruct;
+
import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.logging.log4j.Logger;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.json.JSONTokener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
+import org.springframework.web.servlet.support.ServletUriComponentsBuilder;
+
import org.apache.geode.SerializationException;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.LowMemoryException;
@@ -33,11 +62,12 @@ import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.LeaseExpiredException;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.pdx.JSONFormatter;
import org.apache.geode.pdx.JSONFormatterException;
import org.apache.geode.pdx.PdxInstance;
+import org.apache.geode.rest.internal.web.controllers.support.CacheProvider;
import org.apache.geode.rest.internal.web.controllers.support.JSONTypes;
import org.apache.geode.rest.internal.web.controllers.support.UpdateOp;
import org.apache.geode.rest.internal.web.exception.DataTypeNotSupportedException;
@@ -51,79 +81,49 @@ import org.apache.geode.rest.internal.web.util.IdentifiableUtils;
import org.apache.geode.rest.internal.web.util.JSONUtils;
import org.apache.geode.rest.internal.web.util.NumberUtils;
import org.apache.geode.rest.internal.web.util.ValidationUtils;
-import org.apache.logging.log4j.Logger;
-import org.json.JSONArray;
-import org.json.JSONException;
-import org.json.JSONObject;
-import org.json.JSONTokener;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
-import org.springframework.util.Assert;
-import org.springframework.util.ClassUtils;
-import org.springframework.util.CollectionUtils;
-import org.springframework.util.StringUtils;
-import org.springframework.web.servlet.support.ServletUriComponentsBuilder;
-
-import java.io.IOException;
-import java.io.UnsupportedEncodingException;
-import java.net.URI;
-import java.net.URLDecoder;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicLong;
-import javax.annotation.PostConstruct;
-
/**
* AbstractBaseController class contains common functionalities required for other controllers.
*
* @since GemFire 8.0
*/
-
@SuppressWarnings("unused")
public abstract class AbstractBaseController {
- protected static final String NEW_META_DATA_PROPERTY = "@new";
- protected static final String OLD_META_DATA_PROPERTY = "@old";
- protected static final String TYPE_META_DATA_PROPERTY = "@type";
- protected static final String UTF_8 = "UTF-8";
- protected static final String DEFAULT_ENCODING = UTF_8;
+ private static final String NEW_META_DATA_PROPERTY = "@new";
+ private static final String OLD_META_DATA_PROPERTY = "@old";
+ private static final String TYPE_META_DATA_PROPERTY = "@type";
+ private static final String UTF_8 = "UTF-8";
+ private static final String DEFAULT_ENCODING = UTF_8;
private static final Logger logger = LogService.getLogger();
private static final AtomicLong ID_SEQUENCE = new AtomicLong(0l);
- // private Cache cache = GemFireCacheImpl.getExisting(null);
@Autowired
- protected RestSecurityService securityService;
+ private RestSecurityService securityService;
@Autowired
private ObjectMapper objectMapper;
+ @Autowired
+ private CacheProvider cacheProvider;
@PostConstruct
private void init() {
JSONUtils.setObjectMapper(objectMapper);
}
- protected Cache getCache() {
- Cache cache = GemFireCacheImpl.getExisting();
+ protected InternalCache getCache() {
+ InternalCache cache = cacheProvider.getInternalCache();
Assert.state(cache != null, "The Gemfire Cache reference was not properly initialized");
return cache;
}
- protected URI toUri(final String... pathSegments) {
+ URI toUri(final String... pathSegments) {
return ServletUriComponentsBuilder.fromCurrentContextPath().path(getRestApiVersion())
.pathSegment(pathSegments).build().toUri();
}
protected abstract String getRestApiVersion();
- protected String validateQuery(String queryInUrl, String queryInBody) {
+ String validateQuery(String queryInUrl, String queryInBody) {
if (!(StringUtils.hasText(queryInUrl) || StringUtils.hasText(queryInBody))) {
throw new GemfireRestException("could not process null value specified in query String");
@@ -131,7 +131,7 @@ public abstract class AbstractBaseController {
return (StringUtils.hasText(queryInUrl) ? decode(queryInUrl) : queryInBody);
}
- protected String decode(final String value) {
+ String decode(final String value) {
if (value == null) {
throw new GemfireRestException("could not process null value specified in query String");
}
@@ -175,7 +175,7 @@ public abstract class AbstractBaseController {
}
@SuppressWarnings("unchecked")
- protected <T> T casValue(String regionNamePath, String key, String jsonData) {
+ private <T> T casValue(String regionNamePath, String key, String jsonData) {
JSONObject jsonObject;
try {
jsonObject = new JSONObject(jsonData);
@@ -189,7 +189,7 @@ public abstract class AbstractBaseController {
}
}
- public ResponseEntity<String> processQueryResponse(Query query, Object args[], Object queryResult)
+ ResponseEntity<String> processQueryResponse(Query query, Object args[], Object queryResult)
throws JSONException {
if (queryResult instanceof Collection<?>) {
Collection processedResults = new ArrayList(((Collection) queryResult).size());
@@ -207,7 +207,7 @@ public abstract class AbstractBaseController {
}
}
- protected Collection<PdxInstance> convertJsonArrayIntoPdxCollection(final String jsonArray) {
+ Collection<PdxInstance> convertJsonArrayIntoPdxCollection(final String jsonArray) {
JSONArray jsonArr = null;
try {
jsonArr = new JSONArray(jsonArray);
@@ -229,18 +229,7 @@ public abstract class AbstractBaseController {
}
}
-
- /*
- * protected PdxInstance convertJsonIntoPdxCollection(final String jsonArray) { JSONArray jsonArr
- * = null;
- *
- * PdxInstance pi = convert(jsonArray);
- * System.out.println("Successfully converted into PdxInstance..!!"); return pi;
- *
- * }
- */
-
- protected Object casValue(final String regionNamePath, final Object key, final Object oldValue,
+ private Object casValue(final String regionNamePath, final Object key, final Object oldValue,
final Object newValue) {
final Region<Object, Object> region = getRegion(regionNamePath);
try {
@@ -281,7 +270,7 @@ public abstract class AbstractBaseController {
}
}
- protected void replaceValue(final String regionNamePath, final Object key,
+ private void replaceValue(final String regionNamePath, final Object key,
final PdxInstance value) {
try {
if (getRegion(regionNamePath).replace(key, value) == null) {
@@ -367,7 +356,7 @@ public abstract class AbstractBaseController {
}
}
- protected void putValue(final String regionNamePath, final Object key, final Object value) {
+ private void putValue(final String regionNamePath, final Object key, final Object value) {
try {
getRegion(regionNamePath).put(key, value);
} catch (NullPointerException npe) {
@@ -397,23 +386,23 @@ public abstract class AbstractBaseController {
}
}
- protected void deleteQueryId(final String regionNamePath, final String key) {
+ private void deleteQueryId(final String regionNamePath, final String key) {
getQueryStore(regionNamePath).remove(key);
}
- protected void deleteNamedQuery(final String regionNamePath, final String key) {
+ void deleteNamedQuery(final String regionNamePath, final String key) {
// Check whether query ID exist in region or not
checkForQueryIdExist(regionNamePath, key);
deleteQueryId(regionNamePath, key);
}
- protected void checkForQueryIdExist(String region, String key) {
+ void checkForQueryIdExist(String region, String key) {
if (!getQueryStore(region).containsKey(key)) {
throw new ResourceNotFoundException(String.format("Named query (%1$s) does not exist!", key));
}
}
- protected Region<String, String> getQueryStore(final String namePath) {
+ Region<String, String> getQueryStore(final String namePath) {
return ValidationUtils.returnValueThrowOnNull(getCache().<String, String>getRegion(namePath),
new GemfireRestException(String.format("Query store does not exist!", namePath)));
}
@@ -436,8 +425,7 @@ public abstract class AbstractBaseController {
}
}
- protected void updateNamedQuery(final String regionNamePath, final String key,
- final String value) {
+ void updateNamedQuery(final String regionNamePath, final String key, final String value) {
try {
getQueryStore(regionNamePath).put(key, value);
} catch (NullPointerException npe) {
@@ -457,8 +445,7 @@ public abstract class AbstractBaseController {
}
@SuppressWarnings("unchecked")
- protected <T> T createNamedQuery(final String regionNamePath, final String key,
- final String value) {
+ <T> T createNamedQuery(final String regionNamePath, final String key, final String value) {
try {
return (T) getQueryStore(regionNamePath).putIfAbsent(key, value);
} catch (UnsupportedOperationException use) {
@@ -482,7 +469,7 @@ public abstract class AbstractBaseController {
}
}
- protected void putPdxValues(final String regionNamePath, final Map<Object, PdxInstance> map) {
+ private void putPdxValues(final String regionNamePath, final Map<Object, PdxInstance> map) {
try {
getRegion(regionNamePath).putAll(map);
} catch (LowMemoryException lme) {
@@ -490,7 +477,7 @@ public abstract class AbstractBaseController {
}
}
- protected void putValues(final String regionNamePath, final Map<Object, Object> values) {
+ private void putValues(final String regionNamePath, final Map<Object, Object> values) {
getRegion(regionNamePath).putAll(values);
}
@@ -510,7 +497,7 @@ public abstract class AbstractBaseController {
}
@SuppressWarnings("unchecked")
- protected <T> T postValue(final String regionNamePath, final Object key, final Object value) {
+ <T> T postValue(final String regionNamePath, final Object key, final Object value) {
try {
return (T) getRegion(regionNamePath).putIfAbsent(key, value);
} catch (UnsupportedOperationException use) {
@@ -561,11 +548,11 @@ public abstract class AbstractBaseController {
return actualValue;
}
- protected String generateKey(final String existingKey) {
+ String generateKey(final String existingKey) {
return generateKey(existingKey, null);
}
- protected String generateKey(final String existingKey, final Object domainObject) {
+ private String generateKey(final String existingKey, final Object domainObject) {
Object domainObjectId = IdentifiableUtils.getId(domainObject);
String newKey;
@@ -597,7 +584,7 @@ public abstract class AbstractBaseController {
return newKey;
}
- protected String decode(final String value, final String encoding) {
+ private String decode(final String value, final String encoding) {
try {
return URLDecoder.decode(value, encoding);
} catch (UnsupportedEncodingException e) {
@@ -612,19 +599,17 @@ public abstract class AbstractBaseController {
String.format("The Region identified by name (%1$s) could not be found!", namePath)));
}
- protected void checkForKeyExist(String region, String key) {
+ private void checkForKeyExist(String region, String key) {
if (!getRegion(region).containsKey(key)) {
throw new ResourceNotFoundException(
String.format("Key (%1$s) does not exist for region (%2$s) in cache!", key, region));
}
}
- protected List<String> checkForMultipleKeysExist(String region, String... keys) {
+ List<String> checkForMultipleKeysExist(String region, String... keys) {
List<String> unknownKeys = new ArrayList<String>();
for (int index = 0; index < keys.length; index++) {
if (!getRegion(region).containsKey(keys[index])) {
- // throw new ResourceNotFoundException(String.format("Key [(%1$s)] does not exist for region
- // [(%2$s)] in cache!", key, region));
unknownKeys.add(keys[index]);
}
}
@@ -664,11 +649,11 @@ public abstract class AbstractBaseController {
return entries.values();
}
- protected void deleteValue(final String regionNamePath, final Object key) {
+ private void deleteValue(final String regionNamePath, final Object key) {
getRegion(regionNamePath).remove(key);
}
- protected void deleteValues(final String regionNamePath, final Object... keys) {
+ void deleteValues(final String regionNamePath, final Object... keys) {
// Check whether all keys exist in cache or not
for (final Object key : keys) {
checkForKeyExist(regionNamePath, key.toString());
@@ -679,7 +664,7 @@ public abstract class AbstractBaseController {
}
}
- protected void deleteValues(String regionNamePath) {
+ void deleteValues(String regionNamePath) {
try {
getRegion(regionNamePath).clear();
} catch (UnsupportedOperationException ue) {
@@ -694,7 +679,7 @@ public abstract class AbstractBaseController {
}
@SuppressWarnings("unchecked")
- protected <T> T introspectAndConvert(final T value) {
+ private <T> T introspectAndConvert(final T value) {
if (value instanceof Map) {
final Map rawDataBinding = (Map) value;
@@ -736,15 +721,15 @@ public abstract class AbstractBaseController {
return value;
}
- protected String convertErrorAsJson(String errorMessage) {
+ String convertErrorAsJson(String errorMessage) {
return ("{" + "\"message\"" + ":" + "\"" + errorMessage + "\"" + "}");
}
- protected String convertErrorAsJson(Throwable t) {
+ String convertErrorAsJson(Throwable t) {
return String.format("{\"message\" : \"%1$s\"}", t.getMessage());
}
- protected Map<?, ?> convertJsonToMap(final String jsonString) {
+ private Map<?, ?> convertJsonToMap(final String jsonString) {
Map<String, String> map = new HashMap<String, String>();
// convert JSON string to Map
@@ -762,11 +747,11 @@ public abstract class AbstractBaseController {
return map;
}
- protected Object jsonToObject(final String jsonString) {
+ private Object jsonToObject(final String jsonString) {
return introspectAndConvert(convertJsonToMap(jsonString));
}
- protected Object[] jsonToObjectArray(final String arguments) {
+ Object[] jsonToObjectArray(final String arguments) {
final JSONTypes jsonType = validateJsonAndFindType(arguments);
if (JSONTypes.JSON_ARRAY.equals(jsonType)) {
try {
@@ -787,8 +772,8 @@ public abstract class AbstractBaseController {
}
}
- public ResponseEntity<String> updateSingleKey(final String region, final String key,
- final String json, final String opValue) {
+ ResponseEntity<String> updateSingleKey(final String region, final String key, final String json,
+ final String opValue) {
final JSONTypes jsonType = validateJsonAndFindType(json);
@@ -822,7 +807,7 @@ public abstract class AbstractBaseController {
}
- public ResponseEntity<String> updateMultipleKeys(final String region, final String[] keys,
+ ResponseEntity<String> updateMultipleKeys(final String region, final String[] keys,
final String json) {
JSONArray jsonArr = null;
@@ -862,7 +847,7 @@ public abstract class AbstractBaseController {
return new ResponseEntity<String>(headers, HttpStatus.OK);
}
- public JSONTypes validateJsonAndFindType(String json) {
+ JSONTypes validateJsonAndFindType(String json) {
try {
Object jsonObj = new JSONTokener(json).nextValue();
@@ -927,17 +912,16 @@ public abstract class AbstractBaseController {
}
protected Set<DistributedMember> getMembers(final String... memberIdNames) {
-
ValidationUtils.returnValueThrowOnNull(memberIdNames,
new GemfireRestException("No member found to run function"));
- final Set<DistributedMember> targetedMembers =
- new HashSet<DistributedMember>(ArrayUtils.length(memberIdNames));
+ final Set<DistributedMember> targetedMembers = new HashSet<>(ArrayUtils.length(memberIdNames));
final List<String> memberIdNameList = Arrays.asList(memberIdNames);
- GemFireCacheImpl c = (GemFireCacheImpl) getCache();
- Set<DistributedMember> distMembers = c.getDistributedSystem().getAllOtherMembers();
+
+ InternalCache cache = getCache();
+ Set<DistributedMember> distMembers = cache.getDistributedSystem().getAllOtherMembers();
// Add the local node to list
- distMembers.add(c.getDistributedSystem().getDistributedMember());
+ distMembers.add(cache.getDistributedSystem().getDistributedMember());
for (DistributedMember member : distMembers) {
if (memberIdNameList.contains(member.getId())
|| memberIdNameList.contains(member.getName())) {
@@ -947,10 +931,10 @@ public abstract class AbstractBaseController {
return targetedMembers;
}
- protected Set<DistributedMember> getAllMembersInDS() {
- GemFireCacheImpl c = (GemFireCacheImpl) getCache();
- Set<DistributedMember> distMembers = c.getDistributedSystem().getAllOtherMembers();
- final Set<DistributedMember> targetedMembers = new HashSet<DistributedMember>();
+ Set<DistributedMember> getAllMembersInDS() {
+ InternalCache cache = getCache();
+ Set<DistributedMember> distMembers = cache.getDistributedSystem().getAllOtherMembers();
+ final Set<DistributedMember> targetedMembers = new HashSet<>();
// find valid data nodes, i.e non locator, non-admin, non-loner nodes
for (DistributedMember member : distMembers) {
@@ -960,7 +944,7 @@ public abstract class AbstractBaseController {
}
}
// Add the local node to list
- targetedMembers.add(c.getDistributedSystem().getDistributedMember());
+ targetedMembers.add(cache.getDistributedSystem().getDistributedMember());
return targetedMembers;
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/BaseControllerAdvice.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/BaseControllerAdvice.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/BaseControllerAdvice.java
index 8939542..751e6a0 100644
--- a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/BaseControllerAdvice.java
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/BaseControllerAdvice.java
@@ -12,16 +12,11 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.rest.internal.web.controllers;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.rest.internal.web.exception.DataTypeNotSupportedException;
-import org.apache.geode.rest.internal.web.exception.GemfireRestException;
-import org.apache.geode.rest.internal.web.exception.MalformedJsonException;
-import org.apache.geode.rest.internal.web.exception.RegionNotFoundException;
-import org.apache.geode.rest.internal.web.exception.ResourceNotFoundException;
-import org.apache.geode.security.NotAuthorizedException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
import org.apache.logging.log4j.Logger;
import org.springframework.http.HttpStatus;
import org.springframework.security.access.AccessDeniedException;
@@ -31,24 +26,26 @@ import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
-import java.io.PrintWriter;
-import java.io.StringWriter;
-
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.rest.internal.web.exception.DataTypeNotSupportedException;
+import org.apache.geode.rest.internal.web.exception.GemfireRestException;
+import org.apache.geode.rest.internal.web.exception.MalformedJsonException;
+import org.apache.geode.rest.internal.web.exception.RegionNotFoundException;
+import org.apache.geode.rest.internal.web.exception.ResourceNotFoundException;
+import org.apache.geode.security.NotAuthorizedException;
/**
* The CrudControllerAdvice class handles exception thrown while serving the REST request
- * <p/>
- *
+ *
* @since GemFire 8.0
*/
-
@ControllerAdvice
@SuppressWarnings("unused")
public class BaseControllerAdvice extends AbstractBaseController {
private static final Logger logger = LogService.getLogger();
- protected static final String REST_API_VERSION = "/v1";
+ private static final String REST_API_VERSION = "/v1";
@Override
protected String getRestApiVersion() {
@@ -186,4 +183,3 @@ public class BaseControllerAdvice extends AbstractBaseController {
}
}
-
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/CommonCrudController.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/CommonCrudController.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/CommonCrudController.java
index 62ce860..0449a45 100644
--- a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/CommonCrudController.java
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/CommonCrudController.java
@@ -14,9 +14,24 @@
*/
package org.apache.geode.rest.internal.web.controllers;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
+import org.apache.logging.log4j.Logger;
+import org.json.JSONException;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+
import org.apache.geode.cache.LowMemoryException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.execute.Execution;
@@ -29,20 +44,6 @@ import org.apache.geode.rest.internal.web.controllers.support.RestServersResultC
import org.apache.geode.rest.internal.web.exception.GemfireRestException;
import org.apache.geode.rest.internal.web.util.ArrayUtils;
import org.apache.geode.rest.internal.web.util.JSONUtils;
-import org.apache.logging.log4j.Logger;
-import org.json.JSONException;
-import org.springframework.http.HttpHeaders;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.MediaType;
-import org.springframework.http.ResponseEntity;
-import org.springframework.security.access.prepost.PreAuthorize;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestMethod;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Set;
/**
* The CommonCrudController serves REST Requests related to listing regions, listing keys in region,
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/FunctionAccessController.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/FunctionAccessController.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/FunctionAccessController.java
index 80996c3..473b8b9 100644
--- a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/FunctionAccessController.java
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/FunctionAccessController.java
@@ -12,24 +12,17 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.rest.internal.web.controllers;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
-import org.apache.geode.cache.LowMemoryException;
-import org.apache.geode.cache.execute.Execution;
-import org.apache.geode.cache.execute.Function;
-import org.apache.geode.cache.execute.FunctionException;
-import org.apache.geode.cache.execute.FunctionService;
-import org.apache.geode.cache.execute.ResultCollector;
-import org.apache.geode.internal.cache.execute.NoResult;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.rest.internal.web.exception.GemfireRestException;
-import org.apache.geode.rest.internal.web.util.ArrayUtils;
-import org.apache.geode.rest.internal.web.util.JSONUtils;
import org.apache.logging.log4j.Logger;
import org.json.JSONException;
import org.springframework.http.HttpHeaders;
@@ -47,10 +40,17 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import org.apache.geode.cache.LowMemoryException;
+import org.apache.geode.cache.execute.Execution;
+import org.apache.geode.cache.execute.Function;
+import org.apache.geode.cache.execute.FunctionException;
+import org.apache.geode.cache.execute.FunctionService;
+import org.apache.geode.cache.execute.ResultCollector;
+import org.apache.geode.internal.cache.execute.NoResult;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.rest.internal.web.exception.GemfireRestException;
+import org.apache.geode.rest.internal.web.util.ArrayUtils;
+import org.apache.geode.rest.internal.web.util.JSONUtils;
/**
* The FunctionsController class serving REST Requests related to the function execution
@@ -58,20 +58,18 @@ import java.util.Set;
* @see org.springframework.stereotype.Controller
* @since GemFire 8.0
*/
-
@Controller("functionController")
@Api(value = "functions", description = "Rest api for gemfire function execution")
@RequestMapping(FunctionAccessController.REST_API_VERSION + "/functions")
@SuppressWarnings("unused")
public class FunctionAccessController extends AbstractBaseController {
// Constant String value indicating the version of the REST API.
- protected static final String REST_API_VERSION = "/v1";
+ static final String REST_API_VERSION = "/v1";
private static final Logger logger = LogService.getLogger();
/**
* Gets the version of the REST API implemented by this @Controller.
- * <p>
- *
+ *
* @return a String indicating the REST API version.
*/
@Override
@@ -256,4 +254,3 @@ public class FunctionAccessController extends AbstractBaseController {
}
}
}
-
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/PdxBasedCrudController.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/PdxBasedCrudController.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/PdxBasedCrudController.java
index 3b08c5f..a6f10b1 100644
--- a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/PdxBasedCrudController.java
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/PdxBasedCrudController.java
@@ -14,16 +14,14 @@
*/
package org.apache.geode.rest.internal.web.controllers;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.rest.internal.web.controllers.support.JSONTypes;
-import org.apache.geode.rest.internal.web.controllers.support.RegionData;
-import org.apache.geode.rest.internal.web.controllers.support.RegionEntryData;
-import org.apache.geode.rest.internal.web.exception.ResourceNotFoundException;
-import org.apache.geode.rest.internal.web.util.ArrayUtils;
import org.apache.logging.log4j.Logger;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
@@ -38,19 +36,20 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.rest.internal.web.controllers.support.JSONTypes;
+import org.apache.geode.rest.internal.web.controllers.support.RegionData;
+import org.apache.geode.rest.internal.web.controllers.support.RegionEntryData;
+import org.apache.geode.rest.internal.web.exception.ResourceNotFoundException;
+import org.apache.geode.rest.internal.web.util.ArrayUtils;
/**
* The PdxBasedCrudController class serving REST Requests related to the REST CRUD operation on
* region
- * <p/>
- *
+ *
* @see org.springframework.stereotype.Controller
* @since GemFire 8.0
*/
-
@Controller("pdxCrudController")
@Api(value = "region", description = "region CRUD operations")
@RequestMapping(PdxBasedCrudController.REST_API_VERSION)
@@ -59,9 +58,9 @@ public class PdxBasedCrudController extends CommonCrudController {
private static final Logger logger = LogService.getLogger();
- protected static final String REST_API_VERSION = "/v1";
+ static final String REST_API_VERSION = "/v1";
- protected static final String DEFAULT_GETALL_RESULT_LIMIT = "50";
+ private static final String DEFAULT_GETALL_RESULT_LIMIT = "50";
@Override
protected String getRestApiVersion() {
@@ -76,7 +75,6 @@ public class PdxBasedCrudController extends CommonCrudController {
* @param json JSON document that is stored against the key
* @return JSON document
*/
-
@RequestMapping(method = RequestMethod.POST, value = "/{region}",
consumes = MediaType.APPLICATION_JSON_VALUE, produces = {MediaType.APPLICATION_JSON_VALUE})
@ApiOperation(value = "create entry", notes = "Create (put-if-absent) data in region",
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/QueryAccessController.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/QueryAccessController.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/QueryAccessController.java
index b00a7aa..8007491 100644
--- a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/QueryAccessController.java
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/QueryAccessController.java
@@ -12,28 +12,14 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.rest.internal.web.controllers;
+import java.util.concurrent.ConcurrentHashMap;
+
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
-import org.apache.geode.cache.Region;
-import org.apache.geode.cache.query.FunctionDomainException;
-import org.apache.geode.cache.query.NameResolutionException;
-import org.apache.geode.cache.query.Query;
-import org.apache.geode.cache.query.QueryExecutionLowMemoryException;
-import org.apache.geode.cache.query.QueryExecutionTimeoutException;
-import org.apache.geode.cache.query.QueryInvalidException;
-import org.apache.geode.cache.query.QueryInvocationTargetException;
-import org.apache.geode.cache.query.TypeMismatchException;
-import org.apache.geode.cache.query.internal.DefaultQuery;
-import org.apache.geode.internal.logging.LogService;
-import org.apache.geode.rest.internal.web.exception.GemfireRestException;
-import org.apache.geode.rest.internal.web.exception.ResourceNotFoundException;
-import org.apache.geode.rest.internal.web.util.JSONUtils;
-import org.apache.geode.rest.internal.web.util.ValidationUtils;
import org.apache.logging.log4j.Logger;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
@@ -49,17 +35,28 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
-import java.util.concurrent.ConcurrentHashMap;
-
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.query.FunctionDomainException;
+import org.apache.geode.cache.query.NameResolutionException;
+import org.apache.geode.cache.query.Query;
+import org.apache.geode.cache.query.QueryExecutionLowMemoryException;
+import org.apache.geode.cache.query.QueryExecutionTimeoutException;
+import org.apache.geode.cache.query.QueryInvalidException;
+import org.apache.geode.cache.query.QueryInvocationTargetException;
+import org.apache.geode.cache.query.TypeMismatchException;
+import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.internal.logging.LogService;
+import org.apache.geode.rest.internal.web.exception.GemfireRestException;
+import org.apache.geode.rest.internal.web.exception.ResourceNotFoundException;
+import org.apache.geode.rest.internal.web.util.JSONUtils;
+import org.apache.geode.rest.internal.web.util.ValidationUtils;
/**
* The QueryingController class serves all HTTP REST requests related to the gemfire querying
- * <p/>
- *
+ *
* @see org.springframework.stereotype.Controller
* @since GemFire 8.0
*/
-
@Controller("queryController")
@Api(value = "queries", description = "Rest api for geode query execution",
produces = MediaType.APPLICATION_JSON_VALUE)
@@ -69,17 +66,16 @@ public class QueryAccessController extends AbstractBaseController {
private static final Logger logger = LogService.getLogger();
- protected static final String PARAMETERIZED_QUERIES_REGION = "__ParameterizedQueries__";
+ private static final String PARAMETERIZED_QUERIES_REGION = "__ParameterizedQueries__";
private final ConcurrentHashMap<String, DefaultQuery> compiledQueries = new ConcurrentHashMap<>();
// Constant String value indicating the version of the REST API.
- protected static final String REST_API_VERSION = "/v1";
+ static final String REST_API_VERSION = "/v1";
/**
* Gets the version of the REST API implemented by this @Controller.
- * <p/>
- *
+ *
* @return a String indicating the REST API version.
*/
@Override
@@ -324,7 +320,6 @@ public class QueryAccessController extends AbstractBaseController {
logger.debug("Updating a named, parametrized Query ({}) with ID ({})...", oqlStatement,
queryId);
-
// update the OQL statement with 'queryId' as the Key into the hidden, Parametrized Queries
// Region...
checkForQueryIdExist(PARAMETERIZED_QUERIES_REGION, queryId);
@@ -334,7 +329,6 @@ public class QueryAccessController extends AbstractBaseController {
return new ResponseEntity<>(HttpStatus.OK);
}
- // delete named, parametrized query
/**
* Delete named, parametrized Query
*
@@ -360,4 +354,3 @@ public class QueryAccessController extends AbstractBaseController {
}
}
-
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProvider.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProvider.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProvider.java
new file mode 100644
index 0000000..19d2755
--- /dev/null
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProvider.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.rest.internal.web.controllers.support;
+
+import org.apache.geode.internal.cache.InternalCache;
+
+public interface CacheProvider {
+
+ InternalCache getInternalCache();
+}
http://git-wip-us.apache.org/repos/asf/geode/blob/47d8c820/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProviderImpl.java
----------------------------------------------------------------------
diff --git a/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProviderImpl.java b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProviderImpl.java
new file mode 100644
index 0000000..8c00923
--- /dev/null
+++ b/geode-web-api/src/main/java/org/apache/geode/rest/internal/web/controllers/support/CacheProviderImpl.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package org.apache.geode.rest.internal.web.controllers.support;
+
+import org.springframework.stereotype.Component;
+
+import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
+
+@Component("cacheProvider")
+public class CacheProviderImpl implements CacheProvider {
+
+ @Override
+ public InternalCache getInternalCache() {
+ return GemFireCacheImpl.getExisting();
+ }
+}
[07/11] geode git commit: GEODE-2799: Handle different types of
KeyInfo set when creating the KeySet Iterator.
Posted by ds...@apache.org.
GEODE-2799: Handle different types of KeyInfo set when creating the KeySet Iterator.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/60ec931f
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/60ec931f
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/60ec931f
Branch: refs/heads/feature/GEODE-2801
Commit: 60ec931f53ead9a18950fb4b8e441ff3e9993820
Parents: 363e50d
Author: eshu <es...@pivotal.io>
Authored: Fri Apr 21 13:55:49 2017 -0700
Committer: eshu <es...@pivotal.io>
Committed: Fri Apr 21 13:55:49 2017 -0700
----------------------------------------------------------------------
.../geode/internal/cache/LocalRegion.java | 2 +-
.../internal/cache/LocalRegionDataView.java | 14 ++++-
.../apache/geode/internal/cache/TXState.java | 1 +
.../cache/TXStateProxyImplJUnitTest.java | 60 ++++++++++++++++++++
4 files changed, 73 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/60ec931f/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 5d5044b..45035d7 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
@@ -11177,7 +11177,7 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
return getRegionMap().keySet();
}
- public final InternalDataView getSharedDataView() {
+ public InternalDataView getSharedDataView() {
return this.sharedDataView;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/60ec931f/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
index b4aa20b..8db979b 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
@@ -200,11 +200,19 @@ public class LocalRegionDataView implements InternalDataView {
*/
public Object getKeyForIterator(final KeyInfo keyInfo, final LocalRegion currRgn,
boolean rememberReads, boolean allowTombstones) {
- final AbstractRegionEntry re = (AbstractRegionEntry) keyInfo.getKey();
+ final Object key = keyInfo.getKey();
+ if (key == null) {
+ return null;
+ }
// fix for 42182, before returning a key verify that its value
// is not a removed token
- if (re != null && (!re.isDestroyedOrRemoved() || (allowTombstones && re.isTombstone()))) {
- return re.getKey();
+ if (key instanceof RegionEntry) {
+ RegionEntry re = (RegionEntry) key;
+ if (!re.isDestroyedOrRemoved() || (allowTombstones && re.isTombstone())) {
+ return re.getKey();
+ }
+ } else if (getEntry(keyInfo, currRgn, allowTombstones) != null) {
+ return key;
}
return null;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/60ec931f/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
index 234baee..0a9d80e 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXState.java
@@ -1705,6 +1705,7 @@ public class TXState implements TXStateInterface {
*/
public Object getKeyForIterator(KeyInfo curr, LocalRegion currRgn, boolean rememberReads,
boolean allowTombstones) {
+ assert !(curr.getKey() instanceof RegionEntry);
if (!readEntryAndCheckIfDestroyed(curr, currRgn, rememberReads)) {
return curr.getKey();
} else {
http://git-wip-us.apache.org/repos/asf/geode/blob/60ec931f/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplJUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplJUnitTest.java
new file mode 100644
index 0000000..9ce76c5
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/TXStateProxyImplJUnitTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
+ * agreements. See the NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.geode.internal.cache;
+
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.LocalRegion.NonTXEntry;
+import org.apache.geode.internal.cache.region.entry.RegionEntryFactoryBuilder;
+import org.apache.geode.test.junit.categories.UnitTest;
+
+@Category(UnitTest.class)
+public class TXStateProxyImplJUnitTest {
+ @Test
+ public void testGetKeyForIterator() {
+ RegionEntryFactory factory = new RegionEntryFactoryBuilder().getRegionEntryFactoryOrNull(false,
+ false, false, false, false);
+ LocalRegion region = mock(LocalRegion.class);
+ String key = "testkey";
+ RegionEntry re = factory.createEntry(region, key, null);
+ TXId txId = new TXId(mock(InternalDistributedMember.class), 1);
+ TXStateProxyImpl tx = new TXStateProxyImpl(mock(TXManagerImpl.class), txId, false);
+ LocalRegionDataView view = mock(LocalRegionDataView.class);
+ boolean allowTombstones = false;
+ boolean rememberReads = true;
+
+ KeyInfo stringKeyInfo = new KeyInfo(key, null, null);
+ KeyInfo regionEntryKeyInfo = new KeyInfo(re, null, null);
+
+ when(region.getSharedDataView()).thenReturn(view);
+ when(view.getEntry(stringKeyInfo, region, allowTombstones)).thenReturn(mock(NonTXEntry.class));
+ when(view.getKeyForIterator(stringKeyInfo, region, rememberReads, allowTombstones))
+ .thenCallRealMethod();
+ when(view.getKeyForIterator(regionEntryKeyInfo, region, rememberReads, allowTombstones))
+ .thenCallRealMethod();
+
+ Object key1 = tx.getKeyForIterator(regionEntryKeyInfo, region, rememberReads, allowTombstones);
+ assertTrue(key1.equals(key));
+ Object key2 = tx.getKeyForIterator(stringKeyInfo, region, rememberReads, allowTombstones);
+ assertTrue(key2.equals(key));
+ }
+
+}
[05/11] geode git commit: GEODE-2632: refactor code to use
InternalCache instead of GemFireCacheImpl
Posted by ds...@apache.org.
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
index 8435c4c..5dc7bb0 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceVsdStats.java
@@ -25,9 +25,9 @@ import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.internal.NanoTimer;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.statistics.StatisticsTypeFactoryImpl;
import org.apache.geode.internal.cache.FilterProfile;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.logging.LogService;
/**
@@ -44,35 +44,34 @@ public class CqServiceVsdStats {
private static final StatisticsType _type;
/** Name of the created CQs statistic */
- protected static final String CQS_CREATED = "numCqsCreated";
+ private static final String CQS_CREATED = "numCqsCreated";
/** Name of the active CQs statistic */
- protected static final String CQS_ACTIVE = "numCqsActive";
+ private static final String CQS_ACTIVE = "numCqsActive";
/** Name of the stopped CQs statistic */
- protected static final String CQS_STOPPED = "numCqsStopped";
+ private static final String CQS_STOPPED = "numCqsStopped";
/** Name of the closed CQs statistic */
- protected static final String CQS_CLOSED = "numCqsClosed";
+ private static final String CQS_CLOSED = "numCqsClosed";
/** Name of the client's CQs statistic */
- protected static final String CQS_ON_CLIENT = "numCqsOnClient";
+ private static final String CQS_ON_CLIENT = "numCqsOnClient";
/** Number of clients with CQs statistic */
- protected static final String CLIENTS_WITH_CQS = "numClientsWithCqs";
-
+ private static final String CLIENTS_WITH_CQS = "numClientsWithCqs";
/** CQ query execution time. */
- protected static final String CQ_QUERY_EXECUTION_TIME = "cqQueryExecutionTime";
+ private static final String CQ_QUERY_EXECUTION_TIME = "cqQueryExecutionTime";
/** CQ query execution in progress */
- protected static final String CQ_QUERY_EXECUTION_IN_PROGRESS = "cqQueryExecutionInProgress";
+ private static final String CQ_QUERY_EXECUTION_IN_PROGRESS = "cqQueryExecutionInProgress";
/** Completed CQ query executions */
- protected static final String CQ_QUERY_EXECUTIONS_COMPLETED = "cqQueryExecutionsCompleted";
+ private static final String CQ_QUERY_EXECUTIONS_COMPLETED = "cqQueryExecutionsCompleted";
/** Unique CQs, number of different CQ queries */
- protected static final String UNIQUE_CQ_QUERY = "numUniqueCqQuery";
+ private static final String UNIQUE_CQ_QUERY = "numUniqueCqQuery";
/** Id of the CQs created statistic */
private static final int _numCqsCreatedId;
@@ -104,7 +103,7 @@ public class CqServiceVsdStats {
/** Id for unique CQs, difference in CQ queries */
private static final int _numUniqueCqQuery;
- /**
+ /*
* Static initializer to create and initialize the <code>StatisticsType</code>
*/
static {
@@ -140,7 +139,6 @@ public class CqServiceVsdStats {
_cqQueryExecutionsCompletedId = _type.nameToId(CQ_QUERY_EXECUTIONS_COMPLETED);
_cqQueryExecutionInProgressId = _type.nameToId(CQ_QUERY_EXECUTION_IN_PROGRESS);
_numUniqueCqQuery = _type.nameToId(UNIQUE_CQ_QUERY);
-
}
/** The <code>Statistics</code> instance to which most behavior is delegated */
@@ -152,12 +150,10 @@ public class CqServiceVsdStats {
* @param factory The <code>StatisticsFactory</code> which creates the <code>Statistics</code>
* instance
*/
- public CqServiceVsdStats(StatisticsFactory factory) {
+ CqServiceVsdStats(StatisticsFactory factory) {
this._stats = factory.createAtomicStatistics(_type, "CqServiceStats");
}
- // /////////////////// Instance Methods /////////////////////
-
/**
* Closes the <code>HARegionQueueStats</code>.
*/
@@ -170,14 +166,14 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsCreated" stat
*/
- public long getNumCqsCreated() {
+ long getNumCqsCreated() {
return this._stats.getLong(_numCqsCreatedId);
}
/**
* Increments the "numCqsCreated" stat by 1.
*/
- public void incCqsCreated() {
+ void incCqsCreated() {
this._stats.incLong(_numCqsCreatedId, 1);
}
@@ -186,21 +182,21 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsActive" stat
*/
- public long getNumCqsActive() {
+ long getNumCqsActive() {
return this._stats.getLong(_numCqsActiveId);
}
/**
* Increments the "numCqsActive" stat by 1.
*/
- public void incCqsActive() {
+ void incCqsActive() {
this._stats.incLong(_numCqsActiveId, 1);
}
/**
* Decrements the "numCqsActive" stat by 1.
*/
- public void decCqsActive() {
+ void decCqsActive() {
this._stats.incLong(_numCqsActiveId, -1);
}
@@ -209,21 +205,21 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsStopped" stat
*/
- public long getNumCqsStopped() {
+ long getNumCqsStopped() {
return this._stats.getLong(_numCqsStoppedId);
}
/**
* Increments the "numCqsStopped" stat by 1.
*/
- public void incCqsStopped() {
+ void incCqsStopped() {
this._stats.incLong(_numCqsStoppedId, 1);
}
/**
* Decrements the "numCqsStopped" stat by 1.
*/
- public void decCqsStopped() {
+ void decCqsStopped() {
this._stats.incLong(_numCqsStoppedId, -1);
}
@@ -232,14 +228,14 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsClosed" stat
*/
- public long getNumCqsClosed() {
+ long getNumCqsClosed() {
return this._stats.getLong(_numCqsClosedId);
}
/**
* Increments the "numCqsClosed" stat by 1.
*/
- public void incCqsClosed() {
+ void incCqsClosed() {
this._stats.incLong(_numCqsClosedId, 1);
}
@@ -248,21 +244,21 @@ public class CqServiceVsdStats {
*
* @return the current value of the "numCqsOnClient" stat
*/
- public long getNumCqsOnClient() {
+ long getNumCqsOnClient() {
return this._stats.getLong(_numCqsOnClientId);
}
/**
* Increments the "numCqsOnClient" stat by 1.
*/
- public void incCqsOnClient() {
+ void incCqsOnClient() {
this._stats.incLong(_numCqsOnClientId, 1);
}
/**
* Decrements the "numCqsOnClient" stat by 1.
*/
- public void decCqsOnClient() {
+ void decCqsOnClient() {
this._stats.incLong(_numCqsOnClientId, -1);
}
@@ -278,21 +274,21 @@ public class CqServiceVsdStats {
/**
* Increments the "numClientsWithCqs" stat by 1.
*/
- public void incClientsWithCqs() {
+ void incClientsWithCqs() {
this._stats.incLong(_numClientsWithCqsId, 1);
}
/**
* Decrements the "numCqsOnClient" stat by 1.
*/
- public void decClientsWithCqs() {
+ void decClientsWithCqs() {
this._stats.incLong(_numClientsWithCqsId, -1);
}
/**
* Start the CQ Query Execution time.
*/
- public long startCqQueryExecution() {
+ long startCqQueryExecution() {
this._stats.incInt(_cqQueryExecutionInProgressId, 1);
return NanoTimer.getTime();
}
@@ -302,7 +298,7 @@ public class CqServiceVsdStats {
*
* @param start long time value.
*/
- public void endCqQueryExecution(long start) {
+ void endCqQueryExecution(long start) {
long ts = NanoTimer.getTime();
this._stats.incLong(_cqQueryExecutionTimeId, ts - start);
this._stats.incInt(_cqQueryExecutionInProgressId, -1);
@@ -321,14 +317,14 @@ public class CqServiceVsdStats {
/**
* Increments number of Unique queries.
*/
- public void incUniqueCqQuery() {
+ void incUniqueCqQuery() {
this._stats.incInt(_numUniqueCqQuery, 1);
}
/**
* Decrements number of unique Queries.
*/
- public void decUniqueCqQuery() {
+ void decUniqueCqQuery() {
this._stats.incInt(_numUniqueCqQuery, -1);
}
@@ -338,11 +334,8 @@ public class CqServiceVsdStats {
* tests.
* <p>
* Returns the number of CQs (active + suspended) on the given region.
- *
- * @param regionName
*/
- public long numCqsOnRegion(String regionName) {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ public long numCqsOnRegion(final InternalCache cache, String regionName) {
if (cache == null) {
return 0;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
index ec6e984..c484105 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ServerCQImpl.java
@@ -21,24 +21,18 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.logging.log4j.Logger;
-import org.apache.geode.CancelException;
import org.apache.geode.DataSerializable;
import org.apache.geode.DataSerializer;
-import org.apache.geode.SystemFailure;
-import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EvictionAction;
-import org.apache.geode.cache.client.internal.UserAttributes;
import org.apache.geode.cache.query.CqAttributes;
import org.apache.geode.cache.query.CqAttributesMutator;
import org.apache.geode.cache.query.CqClosedException;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.CqExistsException;
-import org.apache.geode.cache.query.CqListener;
import org.apache.geode.cache.query.CqResults;
import org.apache.geode.cache.query.Query;
import org.apache.geode.cache.query.QueryException;
@@ -49,6 +43,7 @@ import org.apache.geode.cache.query.internal.CompiledRegion;
import org.apache.geode.cache.query.internal.CompiledSelect;
import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.Token;
@@ -58,7 +53,6 @@ import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-import org.apache.geode.i18n.StringId;
public class ServerCQImpl extends CqQueryImpl implements DataSerializable, ServerCQ {
private static final Logger logger = LogService.getLogger();
@@ -84,7 +78,7 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
public volatile boolean cqResultKeysInitialized = false;
/** Boolean flag to see if the CQ is on Partitioned Region */
- public volatile boolean isPR = false;
+ volatile boolean isPR = false;
private ClientProxyMembershipID clientProxyId = null;
@@ -92,7 +86,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
private String serverCqName;
-
/** identifier assigned to this query for FilterRoutingInfos */
private Long filterID;
@@ -106,21 +99,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// For deserialization
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#getFilterID()
- */
@Override
public Long getFilterID() {
return this.filterID;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#setFilterID(java.lang.Long)
- */
@Override
public void setFilterID(Long filterID) {
this.filterID = filterID;
@@ -142,19 +125,12 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
CacheClientProxy clientProxy = null;
this.clientProxyId = p_clientProxyId;
- // servConnection = serverSideConnection;
if (p_ccn != null) {
this.ccn = p_ccn;
clientProxy = p_ccn.getClientProxy(p_clientProxyId, true);
}
- /*
- * try { initCq(); } catch (CqExistsException cqe) { // Should not happen. throw new
- * CqException(LocalizedStrings.CqQueryImpl_UNABLE_TO_CREATE_CQ_0_ERROR__1.toLocalizedString(new
- * Object[] { cqName, cqe.getMessage()})); }
- */
-
validateCq();
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -228,13 +204,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
throw new CqException(errMsg);
}
- // checkAndSetCqOnRegion();
-
// Can be null by the time we are here
if (clientProxy != null) {
clientProxy.incCqCount();
if (clientProxy.hasOneCq()) {
- cqService.stats.incClientsWithCqs();
+ cqService.stats().incClientsWithCqs();
}
if (isDebugEnabled) {
logger.debug("Added CQ to the base region: {} With key as: {}", cqBaseRegion.getFullPath(),
@@ -307,7 +281,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
* query.
*
* @return String modified query.
- * @throws CqException
*/
private Query constructServerSideQuery() throws QueryException {
GemFireCacheImpl cache = (GemFireCacheImpl) cqService.getCache();
@@ -328,7 +301,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
* Returns if the passed key is part of the CQs result set. This method needs to be called once
* the CQ result key caching is completed (cqResultsCacheInitialized is true).
*
- * @param key
* @return true if key is in the Results Cache.
*/
public boolean isPartOfCqResult(Object key) {
@@ -352,27 +324,18 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#addToCqResultKeys(java.lang.Object)
- */
@Override
public void addToCqResultKeys(Object key) {
if (!CqServiceProvider.MAINTAIN_KEYS) {
return;
}
- // this.logger.fine("Adding key to Results Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.cqResultKeys != null) {
synchronized (this.cqResultKeys) {
this.cqResultKeys.put(key, TOKEN);
if (!this.cqResultKeysInitialized) {
// This key could be coming after add, destroy.
// Remove this from destroy queue.
- // this.logger.fine("Removing key from Destroy Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.destroysWhileCqResultsInProgress != null) {
this.destroysWhileCqResultsInProgress.remove(key);
}
@@ -381,21 +344,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqQuery2#removeFromCqResultKeys(java.lang.Object,
- * boolean)
- */
@Override
public void removeFromCqResultKeys(Object key, boolean isTokenMode) {
if (!CqServiceProvider.MAINTAIN_KEYS) {
return;
}
- // this.logger.fine("Removing key from Results Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.cqResultKeys != null) {
synchronized (this.cqResultKeys) {
if (isTokenMode && this.cqResultKeys.get(key) != Token.DESTROYED) {
@@ -403,8 +356,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
this.cqResultKeys.remove(key);
if (!this.cqResultKeysInitialized) {
- // this.logger.fine("Adding key to Destroy Cache For CQ :" +
- // this.cqName + " key :" + key);
if (this.destroysWhileCqResultsInProgress != null) {
this.destroysWhileCqResultsInProgress.add(key);
}
@@ -415,10 +366,8 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
/**
* Marks the key as destroyed in the CQ Results key cache.
- *
- * @param key
*/
- public void markAsDestroyedInCqResultKeys(Object key) {
+ void markAsDestroyedInCqResultKeys(Object key) {
if (!CqServiceProvider.MAINTAIN_KEYS) {
return;
}
@@ -439,12 +388,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#setCqResultsCacheInitialized()
- */
@Override
public void setCqResultsCacheInitialized() {
if (CqServiceProvider.MAINTAIN_KEYS) {
@@ -466,13 +409,6 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqQuery2#isOldValueRequiredForQueryProcessing(
- * java.lang.Object)
- */
@Override
public boolean isOldValueRequiredForQueryProcessing(Object key) {
if (this.cqResultKeysInitialized && this.isPartOfCqResult(key)) {
@@ -484,18 +420,11 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
/**
* Closes the Query. On Client side, sends the cq close request to server. On Server side, takes
* care of repository cleanup.
- *
- * @throws CqException
*/
public void close() throws CqClosedException, CqException {
close(true);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#close(boolean)
- */
@Override
public void close(boolean sendRequestToServer) throws CqClosedException, CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -523,9 +452,9 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// Stat update.
if (stateBeforeClosing == CqStateImpl.RUNNING) {
- cqService.stats.decCqsActive();
+ cqService.stats().decCqsActive();
} else if (stateBeforeClosing == CqStateImpl.STOPPED) {
- cqService.stats.decCqsStopped();
+ cqService.stats().decCqsStopped();
}
// Clean-up the CQ Results Cache.
@@ -537,8 +466,8 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// Set the state to close, and update stats
this.cqState.setState(CqStateImpl.CLOSED);
- cqService.stats.incCqsClosed();
- cqService.stats.decCqsOnClient();
+ cqService.stats().incCqsClosed();
+ cqService.stats().decCqsOnClient();
if (this.stats != null)
this.stats.close();
}
@@ -564,9 +493,8 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
/**
* Clears the resource used by CQ.
- *
- * @throws CqException
*/
+ @Override
protected void cleanup() throws CqException {
// CqBaseRegion
try {
@@ -575,7 +503,7 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
CacheClientProxy clientProxy = ccn.getClientProxy(clientProxyId);
clientProxy.decCqCount();
if (clientProxy.hasNoCq()) {
- cqService.stats.decClientsWithCqs();
+ cqService.stats().decClientsWithCqs();
}
}
} catch (Exception ex) {
@@ -587,16 +515,9 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
}
/**
- * @param serverCqName The serverCqName to set.
- */
- public void setServerCqName(String serverCqName) {
-
- this.serverCqName = serverCqName;
- }
-
- /**
* Stop or pause executing the query.
*/
+ @Override
public void stop() throws CqClosedException, CqException {
boolean isStopped = false;
synchronized (this.cqState) {
@@ -613,18 +534,16 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
// Change state and stats on the client side
this.cqState.setState(CqStateImpl.STOPPED);
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.decCqsActive();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().decCqsActive();
if (logger.isDebugEnabled()) {
logger.debug("Successfully stopped the CQ. {}", cqName);
}
}
}
- /* DataSerializableFixedID methods ---------------------------------------- */
-
+ @Override
public void fromData(DataInput in) throws IOException, ClassNotFoundException {
- // this.cqName = DataSerializer.readString(in);
synchronized (cqState) {
this.cqState.setState(DataSerializer.readInteger(in));
}
@@ -633,23 +552,14 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
this.filterID = in.readLong();
}
- /*
- * public int getDSFID() { return CQ_QUERY; }
- */
-
+ @Override
public void toData(DataOutput out) throws IOException {
- // DataSerializer.writeString(this.cqName, out);
DataSerializer.writeInteger(this.cqState.getState(), out);
DataSerializer.writeBoolean(this.isDurable, out);
DataSerializer.writeString(this.queryString, out);
out.writeLong(this.filterID);
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#isPR()
- */
@Override
public boolean isPR() {
return isPR;
@@ -676,5 +586,4 @@ public class ServerCQImpl extends CqQueryImpl implements DataSerializable, Serve
throw new IllegalStateException("Execute cannot be called on a CQ on the server");
}
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
index bcf9806..9bddbc7 100644
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ.java
@@ -27,7 +27,6 @@ import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
import org.apache.geode.distributed.internal.DistributionStats;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
@@ -85,8 +84,7 @@ public class ExecuteCQ extends BaseCQCommand {
ServerCQ cqQuery = null;
try {
- qService =
- (DefaultQueryService) ((GemFireCacheImpl) crHelper.getCache()).getLocalQueryService();
+ qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
// Authorization check
AuthorizeRequest authzRequest = servConn.getAuthzRequest();
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
index f333b4b..de61445 100755
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/ExecuteCQ61.java
@@ -28,7 +28,6 @@ import org.apache.geode.cache.query.internal.cq.CqServiceImpl;
import org.apache.geode.cache.query.internal.cq.CqServiceProvider;
import org.apache.geode.cache.query.internal.cq.ServerCQImpl;
import org.apache.geode.distributed.internal.DistributionStats;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
@@ -111,8 +110,7 @@ public class ExecuteCQ61 extends BaseCQCommand {
ServerCQImpl cqQuery = null;
try {
- qService =
- (DefaultQueryService) ((GemFireCacheImpl) crHelper.getCache()).getLocalQueryService();
+ qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
// Authorization check
AuthorizeRequest authzRequest = servConn.getAuthzRequest();
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
index eac9ed3..a2d201d 100755
--- a/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
+++ b/geode-cq/src/main/java/org/apache/geode/internal/cache/tier/sockets/command/GetDurableCQs.java
@@ -22,7 +22,6 @@ import java.util.List;
import org.apache.geode.cache.query.CqException;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.query.internal.cq.CqService;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
@@ -64,8 +63,7 @@ public class GetDurableCQs extends BaseCQCommand {
CqService cqServiceForExec = null;
try {
- qService =
- (DefaultQueryService) ((GemFireCacheImpl) crHelper.getCache()).getLocalQueryService();
+ qService = (DefaultQueryService) crHelper.getCache().getLocalQueryService();
this.securityService.authorizeClusterRead();
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
index 7ace0e8..f4cd706 100644
--- a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
+++ b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsDUnitTest.java
@@ -14,20 +14,15 @@
*/
package org.apache.geode.cache.query.cq.dunit;
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
import static org.junit.Assert.*;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
-
import java.util.Collection;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.query.CqException;
-import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqStatistics;
import org.apache.geode.cache.query.QueryService;
@@ -40,7 +35,7 @@ import org.apache.geode.cache.query.internal.cq.CqServiceImpl;
import org.apache.geode.cache.query.internal.cq.CqServiceVsdStats;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.cache30.CacheSerializableRunnable;
-import org.apache.geode.cache30.CacheTestCase;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
@@ -48,27 +43,26 @@ import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
/**
- * This class tests the ContiunousQuery mechanism in GemFire. This includes the test with different
+ * This class tests the ContinuousQuery mechanism in GemFire. This includes the test with different
* data activities.
- *
*/
@Category(DistributedTest.class)
public class CqStatsDUnitTest extends JUnit4CacheTestCase {
+ // TODO: delete this use of CqQueryDUnitTest
private CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest();
- public CqStatsDUnitTest() {
- super();
- }
-
@Override
public final void postSetUp() throws Exception {
// avoid IllegalStateException from HandShake by connecting all vms to
// system before creating pool
getSystem();
Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
+ @Override
public void run() {
getSystem();
}
@@ -81,6 +75,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
public void validateCQStats(VM vm, final String cqName, final int creates, final int updates,
final int deletes, final int totalEvents, final int cqListenerInvocations) {
vm.invoke(new CacheSerializableRunnable("Validate CQs") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Stats. ### " + cqName);
// Get CQ Service.
@@ -161,6 +156,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
final int stopped, final int closed, final int cqsOnClient, final int cqsOnRegion,
final int clientsWithCqs) {
vm.invoke(new CacheSerializableRunnable("Validate CQ Service Stats") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Service Stats. ### ");
// Get CQ Service.
@@ -176,7 +172,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
CqServiceVsdStats cqServiceVsdStats = null;
try {
cqServiceVsdStats =
- ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats;
+ ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats();
} catch (CqException e) {
// TODO Auto-generated catch block
e.printStackTrace();
@@ -185,12 +181,14 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
fail("Failed to get CQ Service Stats");
}
- getCache().getLogger().info("#### CQ Service stats: " + " CQs created: "
- + cqServiceStats.numCqsCreated() + " CQs active: " + cqServiceStats.numCqsActive()
- + " CQs stopped: " + cqServiceStats.numCqsStopped() + " CQs closed: "
- + cqServiceStats.numCqsClosed() + " CQs on Client: " + cqServiceStats.numCqsOnClient()
- + " CQs on region /root/regionA : " + cqServiceVsdStats.numCqsOnRegion("/root/regionA")
- + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
+ getCache().getLogger()
+ .info("#### CQ Service stats: " + " CQs created: " + cqServiceStats.numCqsCreated()
+ + " CQs active: " + cqServiceStats.numCqsActive() + " CQs stopped: "
+ + cqServiceStats.numCqsStopped() + " CQs closed: " + cqServiceStats.numCqsClosed()
+ + " CQs on Client: " + cqServiceStats.numCqsOnClient()
+ + " CQs on region /root/regionA : "
+ + cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA")
+ + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
// Check for created count.
@@ -223,7 +221,7 @@ public class CqStatsDUnitTest extends JUnit4CacheTestCase {
// Check for CQs on region.
if (cqsOnRegion != CqQueryDUnitTest.noTest) {
assertEquals("Number of CQs on region /root/regionA mismatch", cqsOnRegion,
- cqServiceVsdStats.numCqsOnRegion("/root/regionA"));
+ cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA"));
}
// Check for clients with CQs count.
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
index d6068f1..c03bb8b 100644
--- a/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
+++ b/geode-cq/src/test/java/org/apache/geode/cache/query/cq/dunit/CqStatsUsingPoolDUnitTest.java
@@ -14,21 +14,16 @@
*/
package org.apache.geode.cache.query.cq.dunit;
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
import static org.junit.Assert.*;
-import org.apache.geode.distributed.*;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
+import java.util.Collection;
+import java.util.Properties;
-import java.util.*;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
import org.apache.geode.cache.CacheException;
import org.apache.geode.cache.query.CqException;
-import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqStatistics;
import org.apache.geode.cache.query.QueryService;
@@ -41,7 +36,8 @@ import org.apache.geode.cache.query.internal.cq.CqServiceImpl;
import org.apache.geode.cache.query.internal.cq.CqServiceVsdStats;
import org.apache.geode.cache.query.internal.cq.InternalCqQuery;
import org.apache.geode.cache30.CacheSerializableRunnable;
-import org.apache.geode.cache30.CacheTestCase;
+import org.apache.geode.distributed.ConfigurationProperties;
+import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
@@ -49,21 +45,19 @@ import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.Wait;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
/**
- * This class tests the ContiunousQuery mechanism in GemFire. This includes the test with different
+ * This class tests the ContinuousQuery mechanism in GemFire. This includes the test with different
* data activities.
- *
*/
@Category(DistributedTest.class)
public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
+ // TODO: delete this use of CqQueryUsingPoolDUnitTest
private CqQueryUsingPoolDUnitTest cqDUnitTest = new CqQueryUsingPoolDUnitTest();
- public CqStatsUsingPoolDUnitTest() {
- super();
- }
-
@Override
public Properties getDistributedSystemProperties() {
Properties result = super.getDistributedSystemProperties();
@@ -77,6 +71,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
// system before creating pool
getSystem();
Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
+ @Override
public void run() {
getSystem();
}
@@ -89,6 +84,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
private void validateCQStats(VM vm, final String cqName, final int creates, final int updates,
final int deletes, final int totalEvents, final int cqListenerInvocations) {
vm.invoke(new CacheSerializableRunnable("Validate CQs") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Stats. ### " + cqName);
// Get CQ Service.
@@ -169,6 +165,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
final int stopped, final int closed, final int cqsOnClient, final int cqsOnRegion,
final int clientsWithCqs) {
vm.invoke(new CacheSerializableRunnable("Validate CQ Service Stats") {
+ @Override
public void run2() throws CacheException {
LogWriterUtils.getLogWriter().info("### Validating CQ Service Stats. ### ");
// Get CQ Service.
@@ -184,7 +181,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
CqServiceVsdStats cqServiceVsdStats = null;
try {
cqServiceVsdStats =
- ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats;
+ ((CqServiceImpl) ((DefaultQueryService) qService).getCqService()).stats();
} catch (CqException e) {
// TODO Auto-generated catch block
e.printStackTrace();
@@ -193,12 +190,14 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
fail("Failed to get CQ Service Stats");
}
- getCache().getLogger().info("#### CQ Service stats: " + " CQs created: "
- + cqServiceStats.numCqsCreated() + " CQs active: " + cqServiceStats.numCqsActive()
- + " CQs stopped: " + cqServiceStats.numCqsStopped() + " CQs closed: "
- + cqServiceStats.numCqsClosed() + " CQs on Client: " + cqServiceStats.numCqsOnClient()
- + " CQs on region /root/regionA : " + cqServiceVsdStats.numCqsOnRegion("/root/regionA")
- + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
+ getCache().getLogger()
+ .info("#### CQ Service stats: " + " CQs created: " + cqServiceStats.numCqsCreated()
+ + " CQs active: " + cqServiceStats.numCqsActive() + " CQs stopped: "
+ + cqServiceStats.numCqsStopped() + " CQs closed: " + cqServiceStats.numCqsClosed()
+ + " CQs on Client: " + cqServiceStats.numCqsOnClient()
+ + " CQs on region /root/regionA : "
+ + cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA")
+ + " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
// Check for created count.
@@ -231,7 +230,7 @@ public class CqStatsUsingPoolDUnitTest extends JUnit4CacheTestCase {
// Check for CQs on region.
if (cqsOnRegion != CqQueryUsingPoolDUnitTest.noTest) {
assertEquals("Number of CQs on region /root/regionA mismatch", cqsOnRegion,
- cqServiceVsdStats.numCqsOnRegion("/root/regionA"));
+ cqServiceVsdStats.numCqsOnRegion(GemFireCacheImpl.getInstance(), "/root/regionA"));
}
// Check for clients with CQs count.
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
index 66c4c0a..5dd0d24 100644
--- a/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
+++ b/geode-lucene/src/main/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollector.java
@@ -15,10 +15,8 @@
package org.apache.geode.cache.lucene.internal.distributed;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.logging.log4j.Logger;
@@ -27,7 +25,7 @@ import org.apache.geode.cache.execute.FunctionException;
import org.apache.geode.cache.execute.ResultCollector;
import org.apache.geode.cache.lucene.LuceneQuery;
import org.apache.geode.distributed.DistributedMember;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.logging.LogService;
/**
@@ -44,16 +42,15 @@ import org.apache.geode.internal.logging.LogService;
*/
public class TopEntriesFunctionCollector
implements ResultCollector<TopEntriesCollector, TopEntries> {
- // Use this instance to perform reduce operation
- final CollectorManager<TopEntriesCollector> manager;
+ private static final Logger logger = LogService.getLogger();
- final String id;
+ // Use this instance to perform reduce operation
+ private final CollectorManager<TopEntriesCollector> manager;
- // Instance of gemfire cache to check status and other utility methods
- final private GemFireCacheImpl cache;
- private static final Logger logger = LogService.getLogger();
+ private final String id;
private final Collection<TopEntriesCollector> subResults = new ArrayList<>();
+
private TopEntriesCollector mergedResults;
public TopEntriesFunctionCollector() {
@@ -65,8 +62,7 @@ public class TopEntriesFunctionCollector
}
public TopEntriesFunctionCollector(LuceneFunctionContext<TopEntriesCollector> context,
- GemFireCacheImpl cache) {
- this.cache = cache;
+ InternalCache cache) {
id = cache == null ? String.valueOf(this.hashCode()) : cache.getName();
int limit = context == null ? 0 : context.getLimit();
@@ -115,4 +111,8 @@ public class TopEntriesFunctionCollector
subResults.add(resultOfSingleExecution);
}
}
+
+ String id() {
+ return this.id;
+ }
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
index 5313ced..6690850 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/LuceneQueryFunctionJUnitTest.java
@@ -12,7 +12,6 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.cache.lucene.internal.distributed;
import static org.apache.geode.cache.lucene.test.LuceneTestUtilities.DEFAULT_FIELD;
@@ -41,6 +40,7 @@ import org.apache.geode.cache.lucene.internal.StringQueryProvider;
import org.apache.geode.cache.lucene.internal.repository.IndexRepository;
import org.apache.geode.cache.lucene.internal.repository.IndexResultCollector;
import org.apache.geode.cache.lucene.internal.repository.RepositoryManager;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.execute.InternalFunctionInvocationTargetException;
import org.apache.geode.internal.cache.execute.InternalRegionFunctionContext;
@@ -58,30 +58,30 @@ import org.mockito.Mockito;
@Category(UnitTest.class)
public class LuceneQueryFunctionJUnitTest {
- String regionPath = "/region";
- String indexName = "index";
- final EntryScore<String> r1_1 = new EntryScore<String>("key-1-1", .5f);
- final EntryScore<String> r1_2 = new EntryScore<String>("key-1-2", .4f);
- final EntryScore<String> r1_3 = new EntryScore<String>("key-1-3", .3f);
- final EntryScore<String> r2_1 = new EntryScore<String>("key-2-1", .45f);
- final EntryScore<String> r2_2 = new EntryScore<String>("key-2-2", .35f);
-
- InternalRegionFunctionContext mockContext;
- ResultSender<TopEntriesCollector> mockResultSender;
- Region<Object, Object> mockRegion;
-
- RepositoryManager mockRepoManager;
- IndexRepository mockRepository1;
- IndexRepository mockRepository2;
- IndexResultCollector mockCollector;
- InternalLuceneService mockService;
- LuceneIndexImpl mockIndex;
- LuceneIndexStats mockStats;
-
- ArrayList<IndexRepository> repos;
- LuceneFunctionContext<IndexResultCollector> searchArgs;
- LuceneQueryProvider queryProvider;
- Query query;
+ private String regionPath = "/region";
+
+ private final EntryScore<String> r1_1 = new EntryScore<>("key-1-1", .5f);
+ private final EntryScore<String> r1_2 = new EntryScore<>("key-1-2", .4f);
+ private final EntryScore<String> r1_3 = new EntryScore<>("key-1-3", .3f);
+ private final EntryScore<String> r2_1 = new EntryScore<>("key-2-1", .45f);
+ private final EntryScore<String> r2_2 = new EntryScore<>("key-2-2", .35f);
+
+ private InternalRegionFunctionContext mockContext;
+ private ResultSender<TopEntriesCollector> mockResultSender;
+ private Region<Object, Object> mockRegion;
+
+ private RepositoryManager mockRepoManager;
+ private IndexRepository mockRepository1;
+ private IndexRepository mockRepository2;
+ private IndexResultCollector mockCollector;
+ private InternalLuceneService mockService;
+ private LuceneIndexImpl mockIndex;
+ private LuceneIndexStats mockStats;
+
+ private ArrayList<IndexRepository> repos;
+ private LuceneFunctionContext<IndexResultCollector> searchArgs;
+ private LuceneQueryProvider queryProvider;
+ private Query query;
private InternalCache mockCache;
@@ -120,7 +120,7 @@ public class LuceneQueryFunctionJUnitTest {
List<EntryScore> hits = result.getEntries().getHits();
assertEquals(5, hits.size());
- TopEntriesJUnitTest.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2, r2_2,
+ LuceneTestUtilities.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2, r2_2,
r1_3);
}
@@ -161,7 +161,7 @@ public class LuceneQueryFunctionJUnitTest {
List<EntryScore> hits = result.getEntries().getHits();
assertEquals(3, hits.size());
- TopEntriesJUnitTest.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2);
+ LuceneTestUtilities.verifyResultOrder(result.getEntries().getHits(), r1_1, r2_1, r1_2);
}
@Test
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
index 3bfebdf..5767390 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesCollectorJUnitTest.java
@@ -26,21 +26,22 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.CopyHelper;
import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
import org.apache.geode.cache.lucene.internal.distributed.TopEntriesCollectorManager.ListScanner;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
public class TopEntriesCollectorJUnitTest {
- private EntryScore<String> r1_1 = new EntryScore<String>("1-1", .9f);
- private EntryScore<String> r1_2 = new EntryScore<String>("1-2", .7f);
- private EntryScore<String> r1_3 = new EntryScore<String>("1-3", .5f);
+ private EntryScore<String> r1_1 = new EntryScore<>("1-1", .9f);
+ private EntryScore<String> r1_2 = new EntryScore<>("1-2", .7f);
+ private EntryScore<String> r1_3 = new EntryScore<>("1-3", .5f);
- private EntryScore<String> r2_1 = new EntryScore<String>("2-1", .85f);
- private EntryScore<String> r2_2 = new EntryScore<String>("2-2", .65f);
+ private EntryScore<String> r2_1 = new EntryScore<>("2-1", .85f);
+ private EntryScore<String> r2_2 = new EntryScore<>("2-2", .65f);
- private EntryScore<String> r3_1 = new EntryScore<String>("3-1", .8f);
- private EntryScore<String> r3_2 = new EntryScore<String>("3-2", .6f);
- private EntryScore<String> r3_3 = new EntryScore<String>("3-3", .4f);
+ private EntryScore<String> r3_1 = new EntryScore<>("3-1", .8f);
+ private EntryScore<String> r3_2 = new EntryScore<>("3-2", .6f);
+ private EntryScore<String> r3_3 = new EntryScore<>("3-3", .4f);
private TopEntriesCollectorManager manager;
@@ -72,7 +73,7 @@ public class TopEntriesCollectorJUnitTest {
TopEntriesCollector hits = manager.reduce(collectors);
assertEquals(8, hits.getEntries().getHits().size());
- TopEntriesJUnitTest.verifyResultOrder(hits.getEntries().getHits(), r1_1, r2_1, r3_1, r1_2, r2_2,
+ LuceneTestUtilities.verifyResultOrder(hits.getEntries().getHits(), r1_1, r2_1, r3_1, r1_2, r2_2,
r3_2, r1_3, r3_3);
// input collector should not change
@@ -116,7 +117,7 @@ public class TopEntriesCollectorJUnitTest {
c1.collect(r1_3.getKey(), r1_3.getScore());
assertEquals(3, c1.getEntries().getHits().size());
- TopEntriesJUnitTest.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
+ LuceneTestUtilities.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
ListScanner scanner = new ListScanner(c1.getEntries().getHits());
assertTrue(scanner.hasNext());
@@ -131,6 +132,6 @@ public class TopEntriesCollectorJUnitTest {
assertFalse(scanner.hasNext());
assertEquals(3, c1.getEntries().getHits().size());
- TopEntriesJUnitTest.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
+ LuceneTestUtilities.verifyResultOrder(c1.getEntries().getHits(), r1_1, r1_2, r1_3);
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
index bf08877..5fd9e2d 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesFunctionCollectorJUnitTest.java
@@ -12,7 +12,6 @@
* or implied. See the License for the specific language governing permissions and limitations under
* the License.
*/
-
package org.apache.geode.cache.lucene.internal.distributed;
import static org.junit.Assert.*;
@@ -20,9 +19,7 @@ import static org.mockito.Matchers.any;
import static org.mockito.Mockito.*;
import java.util.Collection;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
import org.junit.Assert;
import org.junit.Before;
@@ -31,25 +28,26 @@ import org.junit.experimental.categories.Category;
import org.mockito.ArgumentMatcher;
import org.mockito.Mockito;
-import org.apache.geode.CancelCriterion;
-import org.apache.geode.cache.execute.FunctionException;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
public class TopEntriesFunctionCollectorJUnitTest {
- EntryScore<String> r1_1;
- EntryScore<String> r1_2;
- EntryScore<String> r2_1;
- EntryScore<String> r2_2;
- TopEntriesCollector result1, result2;
+
+ private EntryScore<String> r1_1;
+ private EntryScore<String> r1_2;
+ private EntryScore<String> r2_1;
+ private EntryScore<String> r2_2;
+ private TopEntriesCollector result1;
+ private TopEntriesCollector result2;
@Before
public void initializeCommonObjects() {
- r1_1 = new EntryScore<String>("3", .9f);
- r1_2 = new EntryScore<String>("1", .8f);
- r2_1 = new EntryScore<String>("2", 0.85f);
- r2_2 = new EntryScore<String>("4", 0.1f);
+ r1_1 = new EntryScore<>("3", .9f);
+ r1_2 = new EntryScore<>("1", .8f);
+ r2_1 = new EntryScore<>("2", 0.85f);
+ r2_2 = new EntryScore<>("4", 0.1f);
result1 = new TopEntriesCollector(null);
result1.collect(r1_1);
@@ -73,13 +71,9 @@ public class TopEntriesFunctionCollectorJUnitTest {
collector.addResult(null, result1);
collector.addResult(null, result2);
- final CountDownLatch insideThread = new CountDownLatch(1);
- final CountDownLatch resultReceived = new CountDownLatch(1);
-
- final AtomicReference<TopEntries> result = new AtomicReference<>();
TopEntries merged = collector.getResult(1, TimeUnit.SECONDS);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
@@ -95,7 +89,7 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(3, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2);
}
@Test
@@ -108,7 +102,7 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
@@ -121,12 +115,12 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(4, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
@@ -167,7 +161,7 @@ public class TopEntriesFunctionCollectorJUnitTest {
TopEntries merged = collector.getResult();
Assert.assertNotNull(merged);
assertEquals(2, merged.size());
- TopEntriesJUnitTest.verifyResultOrder(merged.getHits(), r2_1, r2_2);
+ LuceneTestUtilities.verifyResultOrder(merged.getHits(), r2_1, r2_2);
}
@Test(expected = RuntimeException.class)
@@ -184,10 +178,10 @@ public class TopEntriesFunctionCollectorJUnitTest {
@Test
public void testCollectorName() {
- GemFireCacheImpl mockCache = mock(GemFireCacheImpl.class);
+ InternalCache mockCache = mock(InternalCache.class);
Mockito.doReturn("server").when(mockCache).getName();
TopEntriesFunctionCollector function = new TopEntriesFunctionCollector(null, mockCache);
- assertEquals("server", function.id);
+ assertEquals("server", function.id());
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
index fcfebbc..e21ac7f 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/internal/distributed/TopEntriesJUnitTest.java
@@ -16,9 +16,6 @@ package org.apache.geode.cache.lucene.internal.distributed;
import static org.junit.Assert.*;
-import java.util.Collection;
-import java.util.Iterator;
-
import org.jmock.Mockery;
import org.jmock.lib.concurrent.Synchroniser;
import org.jmock.lib.legacy.ClassImposteriser;
@@ -30,6 +27,7 @@ import org.junit.experimental.categories.Category;
import org.apache.geode.CopyHelper;
import org.apache.geode.cache.lucene.LuceneQueryFactory;
import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
+import org.apache.geode.cache.lucene.test.LuceneTestUtilities;
import org.apache.geode.test.junit.categories.UnitTest;
@Category(UnitTest.class)
@@ -44,34 +42,34 @@ public class TopEntriesJUnitTest {
@Test
public void testPopulateTopEntries() {
- TopEntries<String> hits = new TopEntries<String>();
+ TopEntries<String> hits = new TopEntries<>();
hits.addHit(r1_1);
hits.addHit(r2_1);
hits.addHit(r1_2);
hits.addHit(r2_2);
assertEquals(4, hits.size());
- verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2, r2_2);
+ LuceneTestUtilities.verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2, r2_2);
}
@Test
public void putSameScoreEntries() {
- TopEntries<String> hits = new TopEntries<String>();
- EntryScore<String> r1 = new EntryScore<String>("1", .8f);
- EntryScore<String> r2 = new EntryScore<String>("2", .8f);
+ TopEntries<String> hits = new TopEntries<>();
+ EntryScore<String> r1 = new EntryScore<>("1", .8f);
+ EntryScore<String> r2 = new EntryScore<>("2", .8f);
hits.addHit(r1);
hits.addHit(r2);
assertEquals(2, hits.size());
- verifyResultOrder(hits.getHits(), r1, r2);
+ LuceneTestUtilities.verifyResultOrder(hits.getHits(), r1, r2);
}
@Test
public void testInitialization() {
- TopEntries<String> hits = new TopEntries<String>();
+ TopEntries<String> hits = new TopEntries<>();
assertEquals(LuceneQueryFactory.DEFAULT_LIMIT, hits.getLimit());
- hits = new TopEntries<String>(123);
+ hits = new TopEntries<>(123);
assertEquals(123, hits.getLimit());
}
@@ -82,47 +80,33 @@ public class TopEntriesJUnitTest {
@Test
public void enforceLimit() throws Exception {
- TopEntries<String> hits = new TopEntries<String>(3);
+ TopEntries<String> hits = new TopEntries<>(3);
hits.addHit(r1_1);
hits.addHit(r2_1);
hits.addHit(r1_2);
hits.addHit(r2_2);
assertEquals(3, hits.size());
- verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2);
+ LuceneTestUtilities.verifyResultOrder(hits.getHits(), r1_1, r2_1, r1_2);
}
@Test
public void testSerialization() {
LuceneServiceImpl.registerDataSerializables();
- TopEntries<String> hits = new TopEntries<String>(3);
+ TopEntries<String> hits = new TopEntries<>(3);
TopEntries<String> copy = CopyHelper.deepCopy(hits);
assertEquals(3, copy.getLimit());
assertEquals(0, copy.getHits().size());
- hits = new TopEntries<String>(3);
+ hits = new TopEntries<>(3);
hits.addHit(r1_1);
hits.addHit(r2_1);
hits.addHit(r1_2);
copy = CopyHelper.deepCopy(hits);
assertEquals(3, copy.size());
- verifyResultOrder(copy.getHits(), r1_1, r2_1, r1_2);
- }
-
- // TODO: extract to lucene test util class
- public static void verifyResultOrder(Collection<EntryScore<String>> list,
- EntryScore<String>... expectedEntries) {
- Iterator<EntryScore<String>> iter = list.iterator();
- for (EntryScore expectedEntry : expectedEntries) {
- if (!iter.hasNext()) {
- fail();
- }
- EntryScore toVerify = iter.next();
- assertEquals(expectedEntry.getKey(), toVerify.getKey());
- assertEquals(expectedEntry.getScore(), toVerify.getScore(), .0f);
- }
+ LuceneTestUtilities.verifyResultOrder(copy.getHits(), r1_1, r2_1, r1_2);
}
@Before
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
----------------------------------------------------------------------
diff --git a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
index 5563112..17f4dea 100644
--- a/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
+++ b/geode-lucene/src/test/java/org/apache/geode/cache/lucene/test/LuceneTestUtilities.java
@@ -18,8 +18,10 @@ import static org.junit.Assert.*;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -44,6 +46,7 @@ import org.apache.geode.cache.lucene.LuceneService;
import org.apache.geode.cache.lucene.LuceneServiceProvider;
import org.apache.geode.cache.lucene.internal.LuceneIndexForPartitionedRegion;
import org.apache.geode.cache.lucene.internal.LuceneServiceImpl;
+import org.apache.geode.cache.lucene.internal.distributed.EntryScore;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
import org.apache.geode.test.dunit.VM;
@@ -82,6 +85,19 @@ public class LuceneTestUtilities {
public static String Quarter3 = "Q3";
public static String Quarter4 = "Q4";
+ public static void verifyResultOrder(Collection<EntryScore<String>> list,
+ EntryScore<String>... expectedEntries) {
+ Iterator<EntryScore<String>> iter = list.iterator();
+ for (EntryScore expectedEntry : expectedEntries) {
+ if (!iter.hasNext()) {
+ fail();
+ }
+ EntryScore toVerify = iter.next();
+ assertEquals(expectedEntry.getKey(), toVerify.getKey());
+ assertEquals(expectedEntry.getScore(), toVerify.getScore(), .0f);
+ }
+ }
+
public static class IntRangeQueryProvider implements LuceneQueryProvider {
String fieldName;
int lowerValue;
[02/11] geode git commit: GEODE-728: Using the correct parameter in
withArgs
Posted by ds...@apache.org.
GEODE-728: Using the correct parameter in withArgs
ServerRegionFunctionExecutor.withArgs was not using it's argument, it
was just passing the (null) field named args to setArguments.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/5891ed7c
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/5891ed7c
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/5891ed7c
Branch: refs/heads/feature/GEODE-2801
Commit: 5891ed7c4306e761c1f12edf85401ab140429d04
Parents: 48d662a
Author: Dan Smith <up...@apache.org>
Authored: Thu Apr 20 10:45:55 2017 -0700
Committer: Dan Smith <up...@apache.org>
Committed: Thu Apr 20 10:47:23 2017 -0700
----------------------------------------------------------------------
.../geode/internal/cache/execute/ServerRegionFunctionExecutor.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/5891ed7c/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerRegionFunctionExecutor.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerRegionFunctionExecutor.java b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerRegionFunctionExecutor.java
index 6543daf..3a20dc3 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerRegionFunctionExecutor.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/execute/ServerRegionFunctionExecutor.java
@@ -318,7 +318,7 @@ public class ServerRegionFunctionExecutor extends AbstractExecution {
return new ServerRegionFunctionExecutor(this, args);
}
- public Execution withArgs(Object params) {
+ public Execution withArgs(Object args) {
return setArguments(args);
}
[04/11] geode git commit: GEODE-2806: if the batch is dispatched,
even the bucket is no longer primary,
the batch should still be deleted as planned.
Posted by ds...@apache.org.
GEODE-2806: if the batch is dispatched, even the bucket is no longer primary, the batch should still be deleted as planned.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/0862174c
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/0862174c
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/0862174c
Branch: refs/heads/feature/GEODE-2801
Commit: 0862174c30cad1536f2c105b783653bd0d4344e8
Parents: 50686b0
Author: zhouxh <gz...@pivotal.io>
Authored: Fri Apr 21 09:57:05 2017 -0700
Committer: zhouxh <gz...@pivotal.io>
Committed: Fri Apr 21 09:58:19 2017 -0700
----------------------------------------------------------------------
.../parallel/ParallelGatewaySenderQueue.java | 58 ++++++++++----------
1 file changed, 28 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/0862174c/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
index cf4c5a9..9696b90 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
@@ -1104,38 +1104,36 @@ public class ParallelGatewaySenderQueue implements RegionQueue {
private void destroyEventFromQueue(PartitionedRegion prQ, int bucketId, Object key) {
boolean isPrimary = prQ.getRegionAdvisor().getBucketAdvisor(bucketId).isPrimary();
- if (isPrimary) {
- BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
- // TODO : Kishor : Make sure we dont need to initalize a bucket
- // before destroying a key from it
- try {
- if (brq != null) {
- brq.destroyKey(key);
- }
- stats.decQueueSize();
- } catch (EntryNotFoundException e) {
- if (!this.sender.isBatchConflationEnabled() && logger.isDebugEnabled()) {
- logger.debug(
- "ParallelGatewaySenderQueue#remove: Got EntryNotFoundException while removing key {} for {} for bucket = {} for GatewaySender {}",
- key, this, bucketId, this.sender);
- }
- } catch (ForceReattemptException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Bucket :{} moved to other member", bucketId);
- }
- } catch (PrimaryBucketException e) {
- if (logger.isDebugEnabled()) {
- logger.debug("Primary bucket :{} moved to other member", bucketId);
- }
- } catch (RegionDestroyedException e) {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}",
- key, bucketId, prQ.getFullPath());
- }
+ BucketRegionQueue brq = getBucketRegionQueueByBucketId(prQ, bucketId);
+ // TODO : Kishor : Make sure we dont need to initalize a bucket
+ // before destroying a key from it
+ try {
+ if (brq != null) {
+ brq.destroyKey(key);
+ }
+ stats.decQueueSize();
+ } catch (EntryNotFoundException e) {
+ if (!this.sender.isBatchConflationEnabled() && logger.isDebugEnabled()) {
+ logger.debug(
+ "ParallelGatewaySenderQueue#remove: Got EntryNotFoundException while removing key {} for {} for bucket = {} for GatewaySender {}",
+ key, this, bucketId, this.sender);
+ }
+ } catch (ForceReattemptException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Bucket :{} moved to other member", bucketId);
+ }
+ } catch (PrimaryBucketException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Primary bucket :{} moved to other member", bucketId);
+ }
+ } catch (RegionDestroyedException e) {
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "Caught RegionDestroyedException attempting to remove key {} from bucket {} in {}", key,
+ bucketId, prQ.getFullPath());
}
- addRemovedEvent(prQ, bucketId, key);
}
+ addRemovedEvent(prQ, bucketId, key);
}
public void resetLastPeeked() {
[03/11] geode git commit: GEODE-576 & GEODE-516:
GemFireDeadlockDetectorDUnitTest failures
Posted by ds...@apache.org.
GEODE-576 & GEODE-516: GemFireDeadlockDetectorDUnitTest failures
Replaced pauses with Awaitility. Modified asyncs to use the DUnit
blackboard to synchronize their actions for repeatable behavior.
Cleaned up static locks to allow their reuse in other tests or in
repeating the same test.
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/50686b0b
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/50686b0b
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/50686b0b
Branch: refs/heads/feature/GEODE-2801
Commit: 50686b0b44024d2bcbb4bea8a36ce3a40ac158c2
Parents: 5891ed7
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Apr 21 09:05:07 2017 -0700
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Fri Apr 21 09:06:14 2017 -0700
----------------------------------------------------------------------
.../GemFireDeadlockDetectorDUnitTest.java | 116 +++++++++++--------
.../geode/test/dunit/DUnitBlackboard.java | 13 +++
.../test/dunit/internal/InternalBlackboard.java | 5 +
.../dunit/internal/InternalBlackboardImpl.java | 5 +
4 files changed, 91 insertions(+), 48 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java b/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
index e0bbde0..4a03c2d 100644
--- a/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/distributed/internal/deadlock/GemFireDeadlockDetectorDUnitTest.java
@@ -14,34 +14,14 @@
*/
package org.apache.geode.distributed.internal.deadlock;
-import org.apache.geode.test.dunit.ThreadUtils;
-import org.junit.experimental.categories.Category;
-import org.junit.Test;
-
-import static org.junit.Assert.*;
-
-import org.awaitility.Awaitility;
-import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
-import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
-import org.apache.geode.test.junit.categories.DistributedTest;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.junit.experimental.categories.Category;
-
-import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.execute.Function;
import org.apache.geode.cache.execute.FunctionContext;
import org.apache.geode.cache.execute.FunctionService;
import org.apache.geode.cache.execute.ResultCollector;
-import org.apache.geode.cache30.CacheTestCase;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.LockServiceDestroyedException;
@@ -54,7 +34,20 @@ import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableCallable;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.VM;
-import org.apache.geode.test.junit.categories.FlakyTest;
+import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
+import org.apache.geode.test.junit.categories.DistributedTest;
+import org.awaitility.Awaitility;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
@Category(DistributedTest.class)
public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
@@ -108,7 +101,8 @@ public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
private static final Lock lock = new ReentrantLock();
- @Category(FlakyTest.class) // GEODE-516 & GEODE-576: async actions, thread sleeps, time sensitive
+ // @Category(FlakyTest.class) // GEODE-516 & GEODE-576: async actions, thread sleeps, time
+ // sensitive
@Test
public void testDistributedDeadlockWithFunction() throws Throwable {
Host host = Host.getHost(0);
@@ -117,41 +111,62 @@ public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
getSystem();
InternalDistributedMember member1 = createCache(vm0);
final InternalDistributedMember member2 = createCache(vm1);
+ getBlackboard().initBlackboard();
// Have two threads lock locks on different members in different orders.
+ String gateOnMember1 = "gateOnMember1";
+ String gateOnMember2 = "gateOnMember2";
// This thread locks the lock member1 first, then member2.
- AsyncInvocation async1 = lockTheLocks(vm0, member2);
- // This thread locks the lock member2 first, then member1.
- AsyncInvocation async2 = lockTheLocks(vm1, member1);
+ AsyncInvocation async1 = lockTheLocks(vm0, member2, gateOnMember1, gateOnMember2);
- Thread.sleep(5000);
- GemFireDeadlockDetector detect = new GemFireDeadlockDetector();
- LinkedList<Dependency> deadlock = detect.find().findCycle();
- LogWriterUtils.getLogWriter().info("Deadlock=" + DeadlockDetector.prettyFormat(deadlock));
- assertEquals(8, deadlock.size());
- stopStuckThreads();
- async1.getResult(30000);
- async2.getResult(30000);
+ // This thread locks the lock member2 first, then member1.
+ AsyncInvocation async2 = lockTheLocks(vm1, member1, gateOnMember2, gateOnMember1);
+ try {
+ final LinkedList<Dependency> deadlockHolder[] = new LinkedList[1];
+ Awaitility.await("waiting for deadlock").atMost(20, TimeUnit.SECONDS).until(() -> {
+ GemFireDeadlockDetector detect = new GemFireDeadlockDetector();
+ LinkedList<Dependency> deadlock = detect.find().findCycle();
+ if (deadlock != null) {
+ deadlockHolder[0] = deadlock;
+ }
+ return deadlock != null;
+ });
+ LinkedList<Dependency> deadlock = deadlockHolder[0];
+ LogWriterUtils.getLogWriter().info("Deadlock=" + DeadlockDetector.prettyFormat(deadlock));
+ assertEquals(8, deadlock.size());
+ stopStuckThreads();
+ } finally {
+ try {
+ waitForAsyncInvocation(async1, 45, TimeUnit.SECONDS);
+ } finally {
+ waitForAsyncInvocation(async2, 45, TimeUnit.SECONDS);
+ }
+ }
}
- private AsyncInvocation lockTheLocks(VM vm0, final InternalDistributedMember member) {
+ private AsyncInvocation lockTheLocks(VM vm0, final InternalDistributedMember member,
+ final String gateToSignal, final String gateToWaitOn) {
return vm0.invokeAsync(new SerializableRunnable() {
public void run() {
lock.lock();
try {
- Thread.sleep(1000);
- } catch (InterruptedException e) {
- Assert.fail("interrupted", e);
+ try {
+ getBlackboard().signalGate(gateToSignal);
+ getBlackboard().waitForGate(gateToWaitOn, 10, TimeUnit.SECONDS);
+ } catch (TimeoutException | InterruptedException e) {
+ throw new RuntimeException("failed", e);
+ }
+ ResultCollector collector = FunctionService.onMember(member).execute(new TestFunction());
+ // wait the function to lock the lock on member.
+ collector.getResult();
+ } finally {
+ lock.unlock();
}
- ResultCollector collector = FunctionService.onMember(member).execute(new TestFunction());
- // wait the function to lock the lock on member.
- collector.getResult();
- lock.unlock();
}
});
}
@@ -244,14 +259,19 @@ public class GemFireDeadlockDetectorDUnitTest extends JUnit4CacheTestCase {
public void execute(FunctionContext context) {
+ boolean acquired = false;
try {
stuckThreads.add(Thread.currentThread());
- lock.tryLock(LOCK_WAIT_TIME, TimeUnit.SECONDS);
+ acquired = lock.tryLock(LOCK_WAIT_TIME, TimeUnit.SECONDS);
} catch (InterruptedException e) {
- // ingore
+ // ignore
+ } finally {
+ if (acquired) {
+ lock.unlock();
+ }
+ stuckThreads.remove(Thread.currentThread());
+ context.getResultSender().lastResult(null);
}
- stuckThreads.remove(Thread.currentThread());
- context.getResultSender().lastResult(null);
}
public String getId() {
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java b/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
index a097cd4..62c92bd 100755
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/DUnitBlackboard.java
@@ -56,6 +56,7 @@ public class DUnitBlackboard {
* signals a boolean gate
*/
public void signalGate(String gateName) {
+ // System.out.println(Thread.currentThread().getName()+": signaling gate " + gateName);
try {
blackboard.signalGate(gateName);
} catch (RemoteException e) {
@@ -68,6 +69,7 @@ public class DUnitBlackboard {
*/
public void waitForGate(String gateName, long timeout, TimeUnit units)
throws TimeoutException, InterruptedException {
+ // System.out.println(Thread.currentThread().getName()+": waiting for gate " + gateName);
try {
blackboard.waitForGate(gateName, timeout, units);
} catch (RemoteException e) {
@@ -77,6 +79,17 @@ public class DUnitBlackboard {
}
/**
+ * clear a gate
+ */
+ public void clearGate(String gateName) {
+ try {
+ blackboard.clearGate(gateName);
+ } catch (RemoteException e) {
+ throw new RuntimeException("remote call failed", e);
+ }
+ }
+
+ /**
* test to see if a gate has been signeled
*/
public boolean isGateSignaled(String gateName) {
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
index 63f833b..bc5b9b7 100755
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboard.java
@@ -50,6 +50,11 @@ public interface InternalBlackboard extends Remote, Serializable {
throws RemoteException, TimeoutException, InterruptedException;
/**
+ * clears a gate
+ */
+ void clearGate(String gateName) throws RemoteException;
+
+ /**
* test to see if a gate has been signeled
*/
boolean isGateSignaled(String gateName) throws RemoteException;
http://git-wip-us.apache.org/repos/asf/geode/blob/50686b0b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
index e7657ed..feeae15 100755
--- a/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
+++ b/geode-core/src/test/java/org/apache/geode/test/dunit/internal/InternalBlackboardImpl.java
@@ -79,6 +79,11 @@ public class InternalBlackboardImpl extends UnicastRemoteObject implements Inter
}
@Override
+ public void clearGate(final String gateName) throws RemoteException {
+ gates.remove(gateName);
+ }
+
+ @Override
public void signalGate(final String gateName) throws RemoteException {
gates.put(gateName, Boolean.TRUE);
}
[11/11] geode git commit: Merge remote-tracking branch
'origin/develop' into feature/GEODE-2801
Posted by ds...@apache.org.
Merge remote-tracking branch 'origin/develop' into feature/GEODE-2801
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/ab47b200
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/ab47b200
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/ab47b200
Branch: refs/heads/feature/GEODE-2801
Commit: ab47b200e496df6e9858b372e0c59388f53b73c7
Parents: 232a61e 6eb9ff3
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Tue Apr 25 11:29:04 2017 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Tue Apr 25 11:29:04 2017 -0700
----------------------------------------------------------------------
.../modules/session/catalina/DeltaSession7.java | 14 +-
.../modules/session/catalina/DeltaSession8.java | 14 +-
.../session/TestSessionsTomcat8Base.java | 34 +
.../modules/session/catalina/DeltaSession.java | 14 +-
.../geode/modules/session/CommandServlet.java | 4 +
.../geode/modules/session/QueryCommand.java | 2 +
.../geode/modules/session/TestSessionsBase.java | 34 +
.../query/internal/cq/CqServiceProvider.java | 22 +-
.../query/internal/cq/spi/CqServiceFactory.java | 8 +-
.../apache/geode/internal/ClassPathLoader.java | 10 +-
.../geode/internal/cache/LocalRegion.java | 2 +-
.../internal/cache/LocalRegionDataView.java | 14 +-
.../apache/geode/internal/cache/TXState.java | 1 +
.../execute/ServerRegionFunctionExecutor.java | 2 +-
.../parallel/ParallelGatewaySenderQueue.java | 58 +-
.../GemFireDeadlockDetectorDUnitTest.java | 116 ++--
.../ClassPathLoaderIntegrationTest.java | 22 +-
.../cache/TXStateProxyImplJUnitTest.java | 60 ++
.../OffHeapLRURecoveryRegressionTest.java | 27 +-
.../geode/test/dunit/DUnitBlackboard.java | 13 +
.../test/dunit/internal/InternalBlackboard.java | 5 +
.../dunit/internal/InternalBlackboardImpl.java | 5 +
.../cache/query/internal/cq/ClientCQImpl.java | 95 +--
.../cache/query/internal/cq/CqQueryImpl.java | 91 ++-
.../query/internal/cq/CqServiceFactoryImpl.java | 17 +-
.../cache/query/internal/cq/CqServiceImpl.java | 673 ++++---------------
.../internal/cq/CqServiceStatisticsImpl.java | 21 +-
.../query/internal/cq/CqServiceVsdStats.java | 73 +-
.../cache/query/internal/cq/ServerCQImpl.java | 121 +---
.../cache/tier/sockets/command/ExecuteCQ.java | 4 +-
.../cache/tier/sockets/command/ExecuteCQ61.java | 4 +-
.../tier/sockets/command/GetDurableCQs.java | 4 +-
.../cache/query/cq/dunit/CqStatsDUnitTest.java | 44 +-
.../cq/dunit/CqStatsUsingPoolDUnitTest.java | 47 +-
.../TopEntriesFunctionCollector.java | 22 +-
.../LuceneQueryFunctionJUnitTest.java | 54 +-
.../TopEntriesCollectorJUnitTest.java | 23 +-
.../TopEntriesFunctionCollectorJUnitTest.java | 48 +-
.../distributed/TopEntriesJUnitTest.java | 44 +-
.../cache/lucene/test/LuceneTestUtilities.java | 16 +
.../web/controllers/AbstractBaseController.java | 194 +++---
.../web/controllers/BaseControllerAdvice.java | 28 +-
.../web/controllers/CommonCrudController.java | 29 +-
.../controllers/FunctionAccessController.java | 39 +-
.../web/controllers/PdxBasedCrudController.java | 28 +-
.../web/controllers/QueryAccessController.java | 49 +-
.../web/controllers/support/CacheProvider.java | 22 +
.../controllers/support/CacheProviderImpl.java | 29 +
48 files changed, 1008 insertions(+), 1292 deletions(-)
----------------------------------------------------------------------
[10/11] geode git commit: GEODE-2097: fix flaky test
Posted by ds...@apache.org.
GEODE-2097: fix flaky test
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/6eb9ff36
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/6eb9ff36
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/6eb9ff36
Branch: refs/heads/feature/GEODE-2801
Commit: 6eb9ff36dddeaaea01f624b453b4fa6d308b93f4
Parents: 45dc674
Author: Darrel Schneider <ds...@pivotal.io>
Authored: Fri Apr 21 16:36:14 2017 -0700
Committer: Darrel Schneider <ds...@pivotal.io>
Committed: Mon Apr 24 08:59:47 2017 -0700
----------------------------------------------------------------------
.../OffHeapLRURecoveryRegressionTest.java | 27 ++++++++++----------
1 file changed, 14 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/6eb9ff36/geode-core/src/test/java/org/apache/geode/internal/offheap/OffHeapLRURecoveryRegressionTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/offheap/OffHeapLRURecoveryRegressionTest.java b/geode-core/src/test/java/org/apache/geode/internal/offheap/OffHeapLRURecoveryRegressionTest.java
index 489c62b..bb781e6 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/offheap/OffHeapLRURecoveryRegressionTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/offheap/OffHeapLRURecoveryRegressionTest.java
@@ -19,7 +19,9 @@ import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.junit.Assert.assertEquals;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import org.awaitility.Awaitility;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -49,7 +51,6 @@ public class OffHeapLRURecoveryRegressionTest {
@Test
public void recoveringTooMuchDataDoesNotRunOutOfOffHeapMemory() {
final int ENTRY_COUNT = 40;
- final int expectedObjectCount;
GemFireCacheImpl gfc = createCache();
try {
Region<Object, Object> r = createRegion(gfc);
@@ -57,29 +58,26 @@ public class OffHeapLRURecoveryRegressionTest {
for (int i = 0; i < ENTRY_COUNT; i++) {
r.put(i, v);
}
- // expect one more during recovery because of the way the LRU limit is
- // enforced during recover.
- expectedObjectCount = MemoryAllocatorImpl.getAllocator().getStats().getObjects() + 1;
} finally {
- gfc.close();
+ closeCache(gfc);
}
+ Awaitility.await().atMost(60, TimeUnit.SECONDS).until(() -> {
+ return MemoryAllocatorImpl.getAllocator().getStats().getObjects() == 0;
+ });
System.setProperty("gemfire.disk.recoverValuesSync", "true");
System.setProperty("gemfire.disk.recoverLruValues", "true");
try {
gfc = createCache();
try {
- Region<Object, Object> r = createRegion(gfc);
+ createDiskStore(gfc);
try {
- assertEquals(ENTRY_COUNT, r.size());
- assertEquals(expectedObjectCount,
- MemoryAllocatorImpl.getAllocator().getStats().getObjects());
+ assertEquals(10, MemoryAllocatorImpl.getAllocator().getStats().getObjects());
} finally {
- r.destroyRegion();
DiskStore ds = gfc.findDiskStore(DS_NAME);
ds.destroy();
}
} finally {
- gfc.close();
+ closeCache(gfc);
}
} finally {
System.clearProperty("gemfire.disk.recoverValuesSync");
@@ -97,9 +95,13 @@ public class OffHeapLRURecoveryRegressionTest {
return result;
}
- private Region<Object, Object> createRegion(GemFireCacheImpl gfc) {
+ private void createDiskStore(GemFireCacheImpl gfc) {
DiskStoreFactory dsf = gfc.createDiskStoreFactory();
dsf.create(DS_NAME);
+ }
+
+ private Region<Object, Object> createRegion(GemFireCacheImpl gfc) {
+ createDiskStore(gfc);
RegionFactory<Object, Object> rf =
gfc.createRegionFactory(RegionShortcut.LOCAL_PERSISTENT_OVERFLOW);
rf.setOffHeap(true);
@@ -110,6 +112,5 @@ public class OffHeapLRURecoveryRegressionTest {
private void closeCache(GemFireCacheImpl gfc) {
gfc.close();
- MemoryAllocatorImpl.freeOffHeapMemory();
}
}
[06/11] geode git commit: GEODE-2632: refactor code to use
InternalCache instead of GemFireCacheImpl
Posted by ds...@apache.org.
GEODE-2632: refactor code to use InternalCache instead of GemFireCacheImpl
* minor cleanup also
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/363e50d2
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/363e50d2
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/363e50d2
Branch: refs/heads/feature/GEODE-2801
Commit: 363e50d213d763f4cca6e0744b206941a4f2d52c
Parents: 0862174
Author: Kirk Lund <kl...@apache.org>
Authored: Wed Apr 19 14:41:42 2017 -0700
Committer: Kirk Lund <kl...@apache.org>
Committed: Fri Apr 21 13:45:22 2017 -0700
----------------------------------------------------------------------
.../query/internal/cq/CqServiceProvider.java | 22 +-
.../query/internal/cq/spi/CqServiceFactory.java | 8 +-
.../cache/query/internal/cq/ClientCQImpl.java | 95 +--
.../cache/query/internal/cq/CqQueryImpl.java | 91 ++-
.../query/internal/cq/CqServiceFactoryImpl.java | 17 +-
.../cache/query/internal/cq/CqServiceImpl.java | 673 ++++---------------
.../internal/cq/CqServiceStatisticsImpl.java | 21 +-
.../query/internal/cq/CqServiceVsdStats.java | 73 +-
.../cache/query/internal/cq/ServerCQImpl.java | 121 +---
.../cache/tier/sockets/command/ExecuteCQ.java | 4 +-
.../cache/tier/sockets/command/ExecuteCQ61.java | 4 +-
.../tier/sockets/command/GetDurableCQs.java | 4 +-
.../cache/query/cq/dunit/CqStatsDUnitTest.java | 44 +-
.../cq/dunit/CqStatsUsingPoolDUnitTest.java | 47 +-
.../TopEntriesFunctionCollector.java | 22 +-
.../LuceneQueryFunctionJUnitTest.java | 54 +-
.../TopEntriesCollectorJUnitTest.java | 23 +-
.../TopEntriesFunctionCollectorJUnitTest.java | 48 +-
.../distributed/TopEntriesJUnitTest.java | 44 +-
.../cache/lucene/test/LuceneTestUtilities.java | 16 +
20 files changed, 442 insertions(+), 989 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
index cded9c3..90fbf4b 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceProvider.java
@@ -16,7 +16,7 @@ package org.apache.geode.cache.query.internal.cq;
import org.apache.geode.cache.query.internal.cq.spi.CqServiceFactory;
import org.apache.geode.distributed.internal.DistributionConfig;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import java.io.DataInput;
import java.io.IOException;
@@ -26,17 +26,19 @@ import java.util.ServiceLoader;
public class CqServiceProvider {
private static final CqServiceFactory factory;
- // System property to maintain the CQ event references for optimizing the updates.
- // This will allows to run the CQ query only once during update events.
+
+ /**
+ * System property to maintain the CQ event references for optimizing the updates. This will allow
+ * running the CQ query only once during update events.
+ */
public static boolean MAINTAIN_KEYS = Boolean
- .valueOf(System.getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.MAINTAIN_KEYS", "true"))
- .booleanValue();
+ .valueOf(System.getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.MAINTAIN_KEYS", "true"));
+
/**
* A debug flag used for testing vMotion during CQ registration
*/
public static boolean VMOTION_DURING_CQ_REGISTRATION_FLAG = false;
-
static {
ServiceLoader<CqServiceFactory> loader = ServiceLoader.load(CqServiceFactory.class);
Iterator<CqServiceFactory> itr = loader.iterator();
@@ -48,8 +50,7 @@ public class CqServiceProvider {
}
}
- public static CqService create(GemFireCacheImpl cache) {
-
+ public static CqService create(InternalCache cache) {
if (factory == null) {
return new MissingCqService();
}
@@ -63,10 +64,7 @@ public class CqServiceProvider {
} else {
return factory.readCqQuery(in);
}
-
}
- private CqServiceProvider() {
-
- }
+ private CqServiceProvider() {}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
----------------------------------------------------------------------
diff --git a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
index 68ebbd5..2b8a47e 100644
--- a/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
+++ b/geode-core/src/main/java/org/apache/geode/cache/query/internal/cq/spi/CqServiceFactory.java
@@ -19,16 +19,16 @@ import java.io.IOException;
import org.apache.geode.cache.query.internal.cq.CqService;
import org.apache.geode.cache.query.internal.cq.ServerCQ;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
public interface CqServiceFactory {
- public void initialize();
+ void initialize();
/**
* Create a new CqService for the given cache
*/
- public CqService create(GemFireCacheImpl cache);
+ CqService create(InternalCache cache);
- public ServerCQ readCqQuery(DataInput in) throws ClassNotFoundException, IOException;
+ ServerCQ readCqQuery(DataInput in) throws ClassNotFoundException, IOException;
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
index 00a0aa5..111bf84 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/ClientCQImpl.java
@@ -35,7 +35,7 @@ import org.apache.geode.cache.query.CqResults;
import org.apache.geode.cache.query.CqStatusListener;
import org.apache.geode.cache.query.RegionNotFoundException;
import org.apache.geode.cache.query.internal.CqStateImpl;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
@@ -57,7 +57,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
*/
private volatile ConcurrentLinkedQueue<CqEventImpl> queuedEvents = null;
- public final Object queuedEventsSynchObject = new Object();
+ final Object queuedEventsSynchObject = new Object();
private boolean connected = false;
@@ -73,22 +73,15 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
return this.cqName;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#getCQProxy()
- */
- public ServerCQProxyImpl getCQProxy() {
+ ServerCQProxyImpl getCQProxy() {
return this.cqProxy;
}
/**
* Initializes the connection using the pool from the client region. Also sets the cqBaseRegion
* value of this CQ.
- *
- * @throws CqException
*/
- public void initConnectionProxy() throws CqException, RegionNotFoundException {
+ private void initConnectionProxy() throws CqException, RegionNotFoundException {
cqBaseRegion = (LocalRegion) cqService.getCache().getRegion(regionName);
// Check if the region exists on the local cache.
// In the current implementation of 5.1 the Server Connection is (ConnectionProxyImpl)
@@ -113,17 +106,9 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
throw new CqException(
"Unable to get the connection pool. The Region does not have a pool configured.");
}
-
- // if (proxy == null) {
- // throw new
- // CqException(LocalizedStrings.CqQueryImpl_UNABLE_TO_GET_THE_CONNECTIONPROXY_THE_REGION_MAY_NOT_HAVE_A_BRIDGEWRITER_OR_BRIDGECLIENT_INSTALLED_ON_IT.toLocalizedString());
- // } else if(!proxy.getEstablishCallbackConnection()){
- // throw new
- // CqException(LocalizedStrings.CqQueryImpl_THE_ESTABLISHCALLBACKCONNECTION_ON_BRIDGEWRITER_CLIENT_INSTALLED_ON_REGION_0_IS_SET_TO_FALSE
- // .toLocalizedString(regionName));
- // }
}
+ @Override
public void close() throws CqClosedException, CqException {
this.close(true);
}
@@ -182,15 +167,15 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (cqProxy == null || !sendRequestToServer || isClosed) {
// Stat update.
if (stateBeforeClosing == CqStateImpl.RUNNING) {
- cqService.stats.decCqsActive();
+ cqService.stats().decCqsActive();
} else if (stateBeforeClosing == CqStateImpl.STOPPED) {
- cqService.stats.decCqsStopped();
+ cqService.stats().decCqsStopped();
}
// Set the state to close, and update stats
this.cqState.setState(CqStateImpl.CLOSED);
- cqService.stats.incCqsClosed();
- cqService.stats.decCqsOnClient();
+ cqService.stats().incCqsClosed();
+ cqService.stats().decCqsOnClient();
if (this.stats != null)
this.stats.close();
} else {
@@ -201,7 +186,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (exception != null) {
throw new CqException(
LocalizedStrings.CqQueryImpl_FAILED_TO_CLOSE_THE_CQ_CQNAME_0_ERROR_FROM_LAST_ENDPOINT_1
- .toLocalizedString(new Object[] {this.cqName, exception.getLocalizedMessage()}),
+ .toLocalizedString(this.cqName, exception.getLocalizedMessage()),
exception.getCause());
} else {
throw new CqException(
@@ -261,31 +246,28 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
/**
* Clears the resource used by CQ.
- *
- * @throws CqException
*/
+ @Override
protected void cleanup() throws CqException {
this.cqService.removeFromBaseRegionToCqNameMap(this.regionName, this.getServerCqName());
}
+ @Override
public CqAttributes getCqAttributes() {
return cqAttributes;
}
-
-
/**
* @return Returns the cqListeners.
*/
public CqListener[] getCqListeners() {
-
return cqAttributes.getCqListeners();
}
-
/**
* Start or resume executing the query.
*/
+ @Override
public void execute() throws CqClosedException, RegionNotFoundException, CqException {
executeCqOnRedundantsAndPrimary(false);
}
@@ -293,7 +275,8 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
/**
* Start or resume executing the query. Gets or updates the CQ results and returns them.
*/
- public CqResults executeWithInitialResults()
+ @Override
+ public <E> CqResults<E> executeWithInitialResults()
throws CqClosedException, RegionNotFoundException, CqException {
synchronized (queuedEventsSynchObject) {
@@ -320,16 +303,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
CqResults initialResults;
try {
initialResults = (CqResults) executeCqOnRedundantsAndPrimary(true);
- } catch (CqClosedException e) {
- queuedEvents = null;
- throw e;
- } catch (RegionNotFoundException e) {
- queuedEvents = null;
- throw e;
- } catch (CqException e) {
- queuedEvents = null;
- throw e;
- } catch (RuntimeException e) {
+ } catch (RegionNotFoundException | CqException | RuntimeException e) {
queuedEvents = null;
throw e;
}
@@ -343,6 +317,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (!this.queuedEvents.isEmpty()) {
try {
Runnable r = new Runnable() {
+ @Override
public void run() {
Object[] eventArray = null;
if (CqQueryImpl.testHook != null) {
@@ -395,7 +370,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
* @param executeWithInitialResults boolean
* @return Object SelectResults in case of executeWithInitialResults
*/
- public Object executeCqOnRedundantsAndPrimary(boolean executeWithInitialResults)
+ private Object executeCqOnRedundantsAndPrimary(boolean executeWithInitialResults)
throws CqClosedException, RegionNotFoundException, CqException {
Object initialResults = null;
@@ -461,8 +436,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
} else {
String errMsg =
LocalizedStrings.CqQueryImpl_FAILED_TO_EXECUTE_THE_CQ_CQNAME_0_QUERY_STRING_IS_1_ERROR_FROM_LAST_SERVER_2
- .toLocalizedString(
- new Object[] {this.cqName, this.queryString, ex.getLocalizedMessage()});
+ .toLocalizedString(this.cqName, this.queryString, ex.getLocalizedMessage());
if (logger.isDebugEnabled()) {
logger.debug(errMsg, ex);
}
@@ -498,8 +472,8 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
}
}
// Update CQ-base region for book keeping.
- this.cqService.stats.incCqsActive();
- this.cqService.stats.decCqsStopped();
+ this.cqService.stats().incCqsActive();
+ this.cqService.stats().decCqsStopped();
return initialResults;
}
@@ -509,23 +483,22 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
* @return true if shutdown in progress else false.
*/
private boolean shutdownInProgress() {
- GemFireCacheImpl cache = GemFireCacheImpl.getInstance();
+ InternalCache cache = cqService.getInternalCache();
if (cache == null || cache.isClosed()) {
return true; // bail, things are shutting down
}
-
String reason = cqProxy.getPool().getCancelCriterion().cancelInProgress();
if (reason != null) {
return true;
}
return false;
-
}
/**
* Stop or pause executing the query.
*/
+ @Override
public void stop() throws CqClosedException, CqException {
boolean isStopped = false;
synchronized (this.cqState) {
@@ -558,8 +531,8 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (cqProxy == null || isStopped) {
// Change state and stats on the client side
this.cqState.setState(CqStateImpl.STOPPED);
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.decCqsActive();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().decCqsActive();
if (logger.isDebugEnabled()) {
logger.debug("Successfully stopped the CQ. {}", cqName);
}
@@ -568,7 +541,7 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
if (exception != null) {
throw new CqException(
LocalizedStrings.CqQueryImpl_FAILED_TO_STOP_THE_CQ_CQNAME_0_ERROR_FROM_LAST_SERVER_1
- .toLocalizedString(new Object[] {this.cqName, exception.getLocalizedMessage()}),
+ .toLocalizedString(this.cqName, exception.getLocalizedMessage()),
exception.getCause());
} else {
throw new CqException(
@@ -579,24 +552,15 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
}
}
+ @Override
public CqAttributesMutator getCqAttributesMutator() {
return (CqAttributesMutator) this.cqAttributes;
}
-
- public ConcurrentLinkedQueue<CqEventImpl> getQueuedEvents() {
+ ConcurrentLinkedQueue<CqEventImpl> getQueuedEvents() {
return this.queuedEvents;
}
-
-
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqQuery2#setProxyCache(org.apache.geode.cache.
- * client.internal.ProxyCache)
- */
@Override
public void setProxyCache(ProxyCache proxyCache) {
this.proxyCache = proxyCache;
@@ -612,7 +576,6 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
@Override
public void createOn(Connection conn, boolean isDurable) {
-
byte regionDataPolicyOrdinal = getCqBaseRegion() == null ? (byte) 0
: getCqBaseRegion().getAttributes().getDataPolicy().ordinal;
@@ -620,6 +583,4 @@ public class ClientCQImpl extends CqQueryImpl implements ClientCQ {
this.cqProxy.createOn(getName(), conn, getQueryString(), state, isDurable,
regionDataPolicyOrdinal);
}
-
-
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
index 22b4137..07e3171 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqQueryImpl.java
@@ -21,11 +21,9 @@ import java.util.Set;
import org.apache.logging.log4j.Logger;
import org.apache.geode.StatisticsFactory;
-import org.apache.geode.cache.Cache;
import org.apache.geode.cache.query.CqClosedException;
import org.apache.geode.cache.query.CqEvent;
import org.apache.geode.cache.query.CqException;
-import org.apache.geode.cache.query.CqExistsException;
import org.apache.geode.cache.query.CqState;
import org.apache.geode.cache.query.CqStatistics;
import org.apache.geode.cache.query.Query;
@@ -38,7 +36,7 @@ import org.apache.geode.cache.query.internal.CqStateImpl;
import org.apache.geode.cache.query.internal.DefaultQuery;
import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.cache.query.internal.QueryExecutionContext;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.InternalLogWriter;
@@ -58,13 +56,13 @@ public abstract class CqQueryImpl implements InternalCqQuery {
protected String queryString;
- protected static final Object TOKEN = new Object();
+ static final Object TOKEN = new Object();
- protected LocalRegion cqBaseRegion;
+ LocalRegion cqBaseRegion;
protected Query query = null;
- protected InternalLogWriter securityLogWriter;
+ InternalLogWriter securityLogWriter;
protected CqServiceImpl cqService;
@@ -72,14 +70,14 @@ public abstract class CqQueryImpl implements InternalCqQuery {
protected boolean isDurable = false;
- // Stats counters
- protected CqStatisticsImpl cqStats;
+ /** Stats counters */
+ private CqStatisticsImpl cqStats;
protected CqQueryVsdStats stats;
protected final CqStateImpl cqState = new CqStateImpl();
- protected ExecutionContext queryExecutionContext = null;
+ private ExecutionContext queryExecutionContext = null;
public static TestHook testHook = null;
@@ -100,6 +98,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* returns CQ name
*/
+ @Override
public String getName() {
return this.cqName;
}
@@ -109,6 +108,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
this.cqName = cqName;
}
+ @Override
public void setCqService(CqService cqService) {
this.cqService = (CqServiceImpl) cqService;
}
@@ -121,25 +121,24 @@ public abstract class CqQueryImpl implements InternalCqQuery {
return this.regionName;
}
- public void updateCqCreateStats() {
+ void updateCqCreateStats() {
// Initialize the VSD statistics
StatisticsFactory factory = cqService.getCache().getDistributedSystem();
this.stats = new CqQueryVsdStats(factory, getServerCqName());
this.cqStats = new CqStatisticsImpl(this);
// Update statistics with CQ creation.
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.incCqsCreated();
- this.cqService.stats.incCqsOnClient();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().incCqsCreated();
+ this.cqService.stats().incCqsOnClient();
}
/**
* Validates the CQ. Checks for cq constraints. Also sets the base region name.
*/
- public void validateCq() {
- Cache cache = cqService.getCache();
- DefaultQuery locQuery =
- (DefaultQuery) ((GemFireCacheImpl) cache).getLocalQueryService().newQuery(this.queryString);
+ void validateCq() {
+ InternalCache cache = cqService.getInternalCache();
+ DefaultQuery locQuery = (DefaultQuery) cache.getLocalQueryService().newQuery(this.queryString);
this.query = locQuery;
// assert locQuery != null;
@@ -221,10 +220,8 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* Removes the CQ from CQ repository.
- *
- * @throws CqException
*/
- protected void removeFromCqMap() throws CqException {
+ void removeFromCqMap() throws CqException {
try {
cqService.removeCq(this.getServerCqName());
} catch (Exception ex) {
@@ -243,6 +240,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* Returns the QueryString of this CQ.
*/
+ @Override
public String getQueryString() {
return queryString;
}
@@ -252,23 +250,16 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return the Query for the query string
*/
+ @Override
public Query getQuery() {
return query;
}
-
- /**
- * @see org.apache.geode.cache.query.CqQuery#getStatistics()
- */
+ @Override
public CqStatistics getStatistics() {
return cqStats;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#getCqBaseRegion()
- */
@Override
public LocalRegion getCqBaseRegion() {
return this.cqBaseRegion;
@@ -279,11 +270,12 @@ public abstract class CqQueryImpl implements InternalCqQuery {
/**
* @return Returns the Region name on which this cq is created.
*/
- public String getBaseRegionName() {
+ String getBaseRegionName() {
return this.regionName;
}
+ @Override
public abstract String getServerCqName();
/**
@@ -291,15 +283,11 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return STOPPED RUNNING or CLOSED
*/
+ @Override
public CqState getState() {
return this.cqState;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqQuery2#setCqState(int)
- */
@Override
public void setCqState(int state) {
if (this.isClosed()) {
@@ -309,18 +297,13 @@ public abstract class CqQueryImpl implements InternalCqQuery {
synchronized (cqState) {
if (state == CqStateImpl.RUNNING) {
- if (this.isRunning()) {
- // throw new
- // IllegalStateException(LocalizedStrings.CqQueryImpl_CQ_IS_NOT_IN_RUNNING_STATE_STOP_CQ_DOES_NOT_APPLY_CQNAME_0
- // .toLocalizedString(this.cqName));
- }
this.cqState.setState(CqStateImpl.RUNNING);
- this.cqService.stats.incCqsActive();
- this.cqService.stats.decCqsStopped();
+ this.cqService.stats().incCqsActive();
+ this.cqService.stats().decCqsStopped();
} else if (state == CqStateImpl.STOPPED) {
this.cqState.setState(CqStateImpl.STOPPED);
- this.cqService.stats.incCqsStopped();
- this.cqService.stats.decCqsActive();
+ this.cqService.stats().incCqsStopped();
+ this.cqService.stats().decCqsActive();
} else if (state == CqStateImpl.CLOSING) {
this.cqState.setState(state);
}
@@ -332,7 +315,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @param cqEvent object
*/
- public void updateStats(CqEvent cqEvent) {
+ void updateStats(CqEvent cqEvent) {
this.stats.updateStats(cqEvent); // Stats for VSD
}
@@ -341,15 +324,17 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return true if running, false otherwise
*/
+ @Override
public boolean isRunning() {
return this.cqState.isRunning();
}
/**
- * Return true if the CQ is in Sstopped state
+ * Return true if the CQ is in stopped state
*
* @return true if stopped, false otherwise
*/
+ @Override
public boolean isStopped() {
return this.cqState.isStopped();
}
@@ -359,6 +344,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return true if closed, false otherwise
*/
+ @Override
public boolean isClosed() {
return this.cqState.isClosed();
}
@@ -377,6 +363,7 @@ public abstract class CqQueryImpl implements InternalCqQuery {
*
* @return true if durable, false otherwise
*/
+ @Override
public boolean isDurable() {
return this.isDurable;
}
@@ -391,22 +378,22 @@ public abstract class CqQueryImpl implements InternalCqQuery {
return stats;
}
- public ExecutionContext getQueryExecutionContext() {
+ ExecutionContext getQueryExecutionContext() {
return queryExecutionContext;
}
- public void setQueryExecutionContext(ExecutionContext queryExecutionContext) {
+ private void setQueryExecutionContext(ExecutionContext queryExecutionContext) {
this.queryExecutionContext = queryExecutionContext;
}
/** Test Hook */
public interface TestHook {
- public void pauseUntilReady();
+ void pauseUntilReady();
- public void ready();
+ void ready();
- public int numQueuedEvents();
+ int numQueuedEvents();
- public void setEventCount(int count);
+ void setEventCount(int count);
}
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
index db90632..9cc2eea 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceFactoryImpl.java
@@ -22,7 +22,7 @@ import java.util.Map;
import org.apache.geode.cache.query.internal.cq.spi.CqServiceFactory;
import org.apache.geode.internal.Version;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
+import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.tier.Command;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CommandInitializer;
@@ -36,14 +36,13 @@ import org.apache.geode.internal.cache.tier.sockets.command.StopCQ;
public class CqServiceFactoryImpl implements CqServiceFactory {
+ @Override
public void initialize() {
- {
- Map<Version, Command> versions = new HashMap<Version, Command>();
- versions.put(Version.GFE_57, ExecuteCQ.getCommand());
- versions.put(Version.GFE_61, ExecuteCQ61.getCommand());
- CommandInitializer.registerCommand(MessageType.EXECUTECQ_MSG_TYPE, versions);
- CommandInitializer.registerCommand(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE, versions);
- }
+ Map<Version, Command> versions = new HashMap<>();
+ versions.put(Version.GFE_57, ExecuteCQ.getCommand());
+ versions.put(Version.GFE_61, ExecuteCQ61.getCommand());
+ CommandInitializer.registerCommand(MessageType.EXECUTECQ_MSG_TYPE, versions);
+ CommandInitializer.registerCommand(MessageType.EXECUTECQ_WITH_IR_MSG_TYPE, versions);
CommandInitializer.registerCommand(MessageType.GETCQSTATS_MSG_TYPE,
Collections.singletonMap(Version.GFE_57, GetCQStats.getCommand()));
@@ -58,7 +57,7 @@ public class CqServiceFactoryImpl implements CqServiceFactory {
}
@Override
- public CqService create(GemFireCacheImpl cache) {
+ public CqService create(InternalCache cache) {
return new CqServiceImpl(cache);
}
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
index f1ca832..570c06c 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceImpl.java
@@ -14,19 +14,63 @@
*/
package org.apache.geode.cache.query.internal.cq;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+import org.apache.logging.log4j.Logger;
+
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
-import org.apache.geode.cache.*;
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheEvent;
+import org.apache.geode.cache.CacheLoaderException;
+import org.apache.geode.cache.EntryEvent;
+import org.apache.geode.cache.Operation;
+import org.apache.geode.cache.RegionEvent;
+import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.client.Pool;
-import org.apache.geode.cache.client.internal.*;
-import org.apache.geode.cache.query.*;
-import org.apache.geode.cache.query.internal.*;
+import org.apache.geode.cache.client.internal.GetEventValueOp;
+import org.apache.geode.cache.client.internal.InternalPool;
+import org.apache.geode.cache.client.internal.QueueManager;
+import org.apache.geode.cache.client.internal.ServerCQProxyImpl;
+import org.apache.geode.cache.client.internal.UserAttributes;
+import org.apache.geode.cache.query.CqAttributes;
+import org.apache.geode.cache.query.CqClosedException;
+import org.apache.geode.cache.query.CqException;
+import org.apache.geode.cache.query.CqExistsException;
+import org.apache.geode.cache.query.CqListener;
+import org.apache.geode.cache.query.CqQuery;
+import org.apache.geode.cache.query.CqServiceStatistics;
+import org.apache.geode.cache.query.CqStatusListener;
+import org.apache.geode.cache.query.QueryException;
+import org.apache.geode.cache.query.QueryInvalidException;
+import org.apache.geode.cache.query.RegionNotFoundException;
+import org.apache.geode.cache.query.SelectResults;
+import org.apache.geode.cache.query.internal.CompiledSelect;
+import org.apache.geode.cache.query.internal.CqQueryVsdStats;
+import org.apache.geode.cache.query.internal.CqStateImpl;
+import org.apache.geode.cache.query.internal.DefaultQuery;
+import org.apache.geode.cache.query.internal.ExecutionContext;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.i18n.StringId;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
-import org.apache.geode.internal.cache.*;
+import org.apache.geode.internal.cache.EntryEventImpl;
+import org.apache.geode.internal.cache.EventID;
+import org.apache.geode.internal.cache.FilterProfile;
+import org.apache.geode.internal.cache.FilterRoutingInfo;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.cache.LocalRegion;
import org.apache.geode.internal.cache.tier.MessageType;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
@@ -35,57 +79,43 @@ import org.apache.geode.internal.cache.tier.sockets.Part;
import org.apache.geode.internal.i18n.LocalizedStrings;
import org.apache.geode.internal.logging.LogService;
import org.apache.geode.internal.logging.log4j.LocalizedMessage;
-import org.apache.logging.log4j.Logger;
-
-import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
/**
- * @since GemFire 5.5
- *
- * Implements the CqService functionality.
- *
- */
-/**
+ * Implements the CqService functionality.
*
+ * @since GemFire 5.5
*/
public final class CqServiceImpl implements CqService {
private static final Logger logger = LogService.getLogger();
- private static final Integer MESSAGE_TYPE_LOCAL_CREATE =
- Integer.valueOf(MessageType.LOCAL_CREATE);
- private static final Integer MESSAGE_TYPE_LOCAL_UPDATE =
- Integer.valueOf(MessageType.LOCAL_UPDATE);
- private static final Integer MESSAGE_TYPE_LOCAL_DESTROY =
- Integer.valueOf(MessageType.LOCAL_DESTROY);
- private static final Integer MESSAGE_TYPE_EXCEPTION = Integer.valueOf(MessageType.EXCEPTION);
+ private static final Integer MESSAGE_TYPE_LOCAL_CREATE = MessageType.LOCAL_CREATE;
+ private static final Integer MESSAGE_TYPE_LOCAL_UPDATE = MessageType.LOCAL_UPDATE;
+ private static final Integer MESSAGE_TYPE_LOCAL_DESTROY = MessageType.LOCAL_DESTROY;
+ private static final Integer MESSAGE_TYPE_EXCEPTION = MessageType.EXCEPTION;
/**
* System property to evaluate the query even though the initial results are not required when cq
* is executed using the execute() method.
*/
- public static boolean EXECUTE_QUERY_DURING_INIT = Boolean
- .valueOf(System
- .getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.EXECUTE_QUERY_DURING_INIT", "true"))
- .booleanValue();
+ public static boolean EXECUTE_QUERY_DURING_INIT = Boolean.valueOf(System
+ .getProperty(DistributionConfig.GEMFIRE_PREFIX + "cq.EXECUTE_QUERY_DURING_INIT", "true"));
private static final String CQ_NAME_PREFIX = "GfCq";
- private final Cache cache;
+ private final InternalCache cache;
/**
* Manages cq pools to determine if a status of connect or disconnect needs to be sent out
*/
- private final HashMap<String, Boolean> cqPoolsConnected = new HashMap<String, Boolean>();
-
+ private final HashMap<String, Boolean> cqPoolsConnected = new HashMap<>();
/**
* Manages CQ objects. uses serverCqName as key and CqQueryImpl as value
*
- * @guarded.By cqQueryMapLock
+ * GuardedBy cqQueryMapLock
*/
- private volatile HashMap<String, CqQueryImpl> cqQueryMap = new HashMap<String, CqQueryImpl>();
+ private volatile HashMap<String, CqQueryImpl> cqQueryMap = new HashMap<>();
+
private final Object cqQueryMapLock = new Object();
private volatile boolean isRunning = false;
@@ -93,36 +123,21 @@ public final class CqServiceImpl implements CqService {
/**
* Used by client when multiuser-authentication is true.
*/
- private final HashMap<String, UserAttributes> cqNameToUserAttributesMap =
- new HashMap<String, UserAttributes>();
-
- // private boolean isServer = true;
-
- /*
- * // Map to manage CQ to satisfied CQ events (keys) for optimizing updates. private final HashMap
- * cqToCqEventKeysMap = CqService.MAINTAIN_KEYS ? new HashMap() : null;
- */
+ private final HashMap<String, UserAttributes> cqNameToUserAttributesMap = new HashMap<>();
// Map to manage the similar CQs (having same query - performance optimization).
// With query as key and Set of CQs as values.
private final ConcurrentHashMap matchingCqMap;
// CQ Service statistics
- public final CqServiceStatisticsImpl cqServiceStats;
- public final CqServiceVsdStats stats;
+ private final CqServiceStatisticsImpl cqServiceStats;
+ private final CqServiceVsdStats stats;
// CQ identifier, also used in auto generated CQ names
private volatile long cqId = 1;
- /**
- * Used to synchronize access to CQs in the repository
- */
- final Object cqSync = new Object();
-
/* This is to manage region to CQs map, client side book keeping. */
- private HashMap<String, ArrayList<String>> baseRegionToCqNameMap =
- new HashMap<String, ArrayList<String>>();
-
+ private HashMap<String, ArrayList<String>> baseRegionToCqNameMap = new HashMap<>();
/**
* Access and modification to the contents of this map do not necessarily need to be lock
@@ -135,33 +150,24 @@ public final class CqServiceImpl implements CqService {
/**
* Constructor.
- *
- * @param c The cache used for the service
+ *
+ * @param cache The cache used for the service
*/
- public CqServiceImpl(final Cache c) {
- if (c == null) {
+ public CqServiceImpl(final InternalCache cache) {
+ if (cache == null) {
throw new IllegalStateException(LocalizedStrings.CqService_CACHE_IS_NULL.toLocalizedString());
}
- GemFireCacheImpl gfc = (GemFireCacheImpl) c;
- gfc.getCancelCriterion().checkCancelInProgress(null);
-
- this.cache = gfc;
+ cache.getCancelCriterion().checkCancelInProgress(null);
+ this.cache = cache;
// Initialize the Map which maintains the matching cqs.
this.matchingCqMap = new ConcurrentHashMap<String, HashSet<String>>();
// Initialize the VSD statistics
- StatisticsFactory factory = cache.getDistributedSystem();
+ StatisticsFactory factory = this.cache.getDistributedSystem();
this.stats = new CqServiceVsdStats(factory);
this.cqServiceStats = new CqServiceStatisticsImpl(this);
-
- // final LoggingThreadGroup group =
- // LoggingThreadGroup.createThreadGroup("CqExecutor Threads", logger);
-
- // if (this.cache.getCacheServers().isEmpty()) {
- // isServer = false;
- // }
}
/**
@@ -171,13 +177,14 @@ public final class CqServiceImpl implements CqService {
return this.cache;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#newCq(java.lang.String,
- * java.lang.String, org.apache.geode.cache.query.CqAttributes,
- * org.apache.geode.cache.client.internal.ServerCQProxy, boolean)
- */
+ public InternalCache getInternalCache() {
+ return this.cache;
+ }
+
+ public CqServiceVsdStats stats() {
+ return this.stats;
+ }
+
@Override
public synchronized ClientCQ newCq(String cqName, String queryString, CqAttributes cqAttributes,
InternalPool pool, boolean isDurable)
@@ -242,22 +249,15 @@ public final class CqServiceImpl implements CqService {
return cQuery;
}
-
/**
* Executes the given CqQuery, if the CqQuery for that name is not there it registers the one and
* executes. This is called on the Server.
*
- * @param cqName
- * @param queryString
- * @param cqState
- * @param clientProxyId
- * @param ccn
* @param manageEmptyRegions whether to update the 6.1 emptyRegions map held in the CCN
* @param regionDataPolicy the data policy of the region associated with the query. This is only
* needed if manageEmptyRegions is true.
* @param emptyRegionsMap map of empty regions.
* @throws IllegalStateException if this is called at client side.
- * @throws CqException
*/
@Override
public synchronized ServerCQ executeCq(String cqName, String queryString, int cqState,
@@ -271,7 +271,7 @@ public final class CqServiceImpl implements CqService {
}
String serverCqName = constructServerCqName(cqName, clientProxyId);
- ServerCQImpl cQuery = null;
+ ServerCQImpl cQuery;
// If this CQ is not yet registered in Server, register CQ.
if (!isCqExists(serverCqName)) {
@@ -292,7 +292,6 @@ public final class CqServiceImpl implements CqService {
logger.info(LocalizedMessage.create(
LocalizedStrings.CqService_EXCEPTION_WHILE_REGISTERING_CQ_ON_SERVER_CQNAME___0,
cQuery.getName()));
- cQuery = null;
throw cqe;
}
@@ -308,6 +307,7 @@ public final class CqServiceImpl implements CqService {
return cQuery;
}
+ @Override
public void resumeCQ(int cqState, ServerCQ cQuery) {
// Initialize the state of CQ.
if (((CqStateImpl) cQuery.getState()).getState() != cqState) {
@@ -324,25 +324,10 @@ public final class CqServiceImpl implements CqService {
}
}
- /*
- * public void addToCqEventKeysMap(CqQuery cq){ if (cqToCqEventKeysMap != null) { synchronized
- * (cqToCqEventKeysMap){ String serverCqName = ((CqQueryImpl)cq).getServerCqName(); if
- * (!cqToCqEventKeysMap.containsKey(serverCqName)){ cqToCqEventKeysMap.put(serverCqName, new
- * HashSet()); if (_logger.isDebugEnabled()) {
- * _logger.debug("CQ Event key maintenance for CQ, CqName: " + serverCqName + " is Enabled." +
- * " key maintenance map size is: " + cqToCqEventKeysMap.size()); } } } // synchronized } }
- */
-
- public boolean hasCq() {
- HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
- return (cqMap.size() > 0);
- }
-
-
/**
* Adds the given CQ and cqQuery object into the CQ map.
*/
- public void addToCqMap(CqQueryImpl cq) throws CqExistsException, CqException {
+ void addToCqMap(CqQueryImpl cq) throws CqExistsException, CqException {
// On server side cqName will be server side cqName.
String sCqName = cq.getServerCqName();
if (logger.isDebugEnabled()) {
@@ -355,7 +340,7 @@ public final class CqServiceImpl implements CqService {
.toLocalizedString(sCqName));
}
synchronized (cqQueryMapLock) {
- HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<String, CqQueryImpl>(cqQueryMap);
+ HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap);
try {
tmpCqQueryMap.put(sCqName, cq);
} catch (Exception ex) {
@@ -377,66 +362,34 @@ public final class CqServiceImpl implements CqService {
/**
* Removes given CQ from the cqMap..
*/
- public void removeCq(String cqName) {
+ void removeCq(String cqName) {
// On server side cqName will be server side cqName.
synchronized (cqQueryMapLock) {
- HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<String, CqQueryImpl>(cqQueryMap);
+ HashMap<String, CqQueryImpl> tmpCqQueryMap = new HashMap<>(cqQueryMap);
tmpCqQueryMap.remove(cqName);
this.cqNameToUserAttributesMap.remove(cqName);
cqQueryMap = tmpCqQueryMap;
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#getClientCqFromServer(org.apache.geode.
- * internal.cache.tier.sockets.ClientProxyMembershipID, java.lang.String)
- */
@Override
public CqQuery getClientCqFromServer(ClientProxyMembershipID clientProxyId, String clientCqName) {
// On server side cqName will be server side cqName.
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
- return (CqQuery) cqMap.get(this.constructServerCqName(clientCqName, clientProxyId));
+ return cqMap.get(this.constructServerCqName(clientCqName, clientProxyId));
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getCq(java.lang.String)
- */
@Override
public InternalCqQuery getCq(String cqName) {
// On server side cqName will be server side cqName.
- return (InternalCqQuery) cqQueryMap.get(cqName);
+ return cqQueryMap.get(cqName);
}
- /**
- * Clears the CQ Query Map.
- */
- public void clearCqQueryMap() {
- // On server side cqName will be server side cqName.
- synchronized (cqQueryMapLock) {
- cqQueryMap = new HashMap<String, CqQueryImpl>();
- }
- }
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getAllCqs()
- */
@Override
public Collection<? extends InternalCqQuery> getAllCqs() {
return cqQueryMap.values();
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getAllCqs(java.lang.String)
- */
@Override
public Collection<? extends InternalCqQuery> getAllCqs(final String regionName)
throws CqException {
@@ -445,7 +398,7 @@ public final class CqServiceImpl implements CqService {
LocalizedStrings.CqService_NULL_ARGUMENT_0.toLocalizedString("regionName"));
}
- String[] cqNames = null;
+ String[] cqNames;
synchronized (this.baseRegionToCqNameMap) {
ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName);
@@ -456,7 +409,7 @@ public final class CqServiceImpl implements CqService {
cqs.toArray(cqNames);
}
- ArrayList<InternalCqQuery> cQueryList = new ArrayList<InternalCqQuery>();
+ ArrayList<InternalCqQuery> cQueryList = new ArrayList<>();
for (int cqCnt = 0; cqCnt < cqNames.length; cqCnt++) {
InternalCqQuery cq = getCq(cqNames[cqCnt]);
if (cq != null) {
@@ -467,34 +420,16 @@ public final class CqServiceImpl implements CqService {
return cQueryList;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#executeAllClientCqs()
- */
@Override
public synchronized void executeAllClientCqs() throws CqException {
executeCqs(this.getAllCqs());
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#executeAllRegionCqs(java.lang.String)
- */
@Override
public synchronized void executeAllRegionCqs(final String regionName) throws CqException {
executeCqs(getAllCqs(regionName));
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#executeCqs(org.apache.geode.cache.query
- * .CqQuery[])
- */
@Override
public synchronized void executeCqs(Collection<? extends InternalCqQuery> cqs)
throws CqException {
@@ -503,53 +438,31 @@ public final class CqServiceImpl implements CqService {
}
String cqName = null;
for (InternalCqQuery internalCq : cqs) {
- CqQuery cq = (CqQuery) internalCq;
+ CqQuery cq = internalCq;
if (!cq.isClosed() && cq.isStopped()) {
try {
cqName = cq.getName();
cq.execute();
- } catch (QueryException qe) {
- if (logger.isDebugEnabled()) {
- logger.debug("Failed to execute the CQ, CqName : {} Error : {}", cqName,
- qe.getMessage());
- }
- } catch (CqClosedException cce) {
+ } catch (QueryException | CqClosedException e) {
if (logger.isDebugEnabled()) {
logger.debug("Failed to execute the CQ, CqName : {} Error : {}", cqName,
- cce.getMessage());
+ e.getMessage());
}
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#stopAllClientCqs()
- */
@Override
public synchronized void stopAllClientCqs() throws CqException {
stopCqs(this.getAllCqs());
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#stopAllRegionCqs(java.lang.String)
- */
@Override
public synchronized void stopAllRegionCqs(final String regionName) throws CqException {
stopCqs(this.getAllCqs(regionName));
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#stopCqs(org.apache.geode.cache.query.
- * CqQuery[])
- */
@Override
public synchronized void stopCqs(Collection<? extends InternalCqQuery> cqs) throws CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -567,29 +480,20 @@ public final class CqServiceImpl implements CqService {
String cqName = null;
for (InternalCqQuery internalCqQuery : cqs) {
- CqQuery cq = (CqQuery) internalCqQuery;
+ CqQuery cq = internalCqQuery;
if (!cq.isClosed() && cq.isRunning()) {
try {
cqName = cq.getName();
cq.stop();
- } catch (QueryException qe) {
- if (isDebugEnabled) {
- logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, qe.getMessage());
- }
- } catch (CqClosedException cce) {
+ } catch (QueryException | CqClosedException e) {
if (isDebugEnabled) {
- logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, cce.getMessage());
+ logger.debug("Failed to stop the CQ, CqName : {} Error : {}", cqName, e.getMessage());
}
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeCqs(java.lang.String)
- */
@Override
public void closeCqs(final String regionName) throws CqException {
Collection<? extends InternalCqQuery> cqs = this.getAllCqs(regionName);
@@ -603,8 +507,8 @@ public final class CqServiceImpl implements CqService {
// invoked on the server
cq.close(false);
} else {
- // @todo grid: if regionName has a pool check its keepAlive
- boolean keepAlive = ((GemFireCacheImpl) this.cache).keepDurableSubscriptionsAlive();
+ // TODO: grid: if regionName has a pool check its keepAlive
+ boolean keepAlive = this.cache.keepDurableSubscriptionsAlive();
if (cq.isDurable() && keepAlive) {
logger.warn(LocalizedMessage.create(
LocalizedStrings.CqService_NOT_SENDING_CQ_CLOSE_TO_THE_SERVER_AS_IT_IS_A_DURABLE_CQ));
@@ -614,14 +518,9 @@ public final class CqServiceImpl implements CqService {
}
}
- } catch (QueryException qe) {
+ } catch (QueryException | CqClosedException e) {
if (logger.isDebugEnabled()) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName, qe.getMessage());
- }
- } catch (CqClosedException cce) {
- if (logger.isDebugEnabled()) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName,
- cce.getMessage());
+ logger.debug("Failed to close the CQ, CqName : {} Error : {}", cqName, e.getMessage());
}
}
}
@@ -630,10 +529,6 @@ public final class CqServiceImpl implements CqService {
/**
* Called directly on server side.
- *
- * @param cqName
- * @param clientId
- * @throws CqException
*/
@Override
public void stopCq(String cqName, ClientProxyMembershipID clientId) throws CqException {
@@ -650,8 +545,6 @@ public final class CqServiceImpl implements CqService {
try {
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
if (!cqMap.containsKey(serverCqName)) {
- // throw new
- // CqException(LocalizedStrings.CqService_CQ_NOT_FOUND_FAILED_TO_STOP_THE_SPECIFIED_CQ_0.toLocalizedString(serverCqName));
/*
* gregp 052808: We should silently fail here instead of throwing error. This is to deal
* with races in recovery
@@ -689,15 +582,8 @@ public final class CqServiceImpl implements CqService {
}
// Send stop message to peers.
cQuery.getCqBaseRegion().getFilterProfile().stopCq(cQuery);
-
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeCq(java.lang.String,
- * org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public void closeCq(String cqName, ClientProxyMembershipID clientProxyId) throws CqException {
String serverCqName = cqName;
@@ -713,9 +599,6 @@ public final class CqServiceImpl implements CqService {
try {
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
if (!cqMap.containsKey(serverCqName)) {
- // throw new
- // CqException(LocalizedStrings.CqService_CQ_NOT_FOUND_FAILED_TO_CLOSE_THE_SPECIFIED_CQ_0
- // .toLocalizedString(serverCqName));
/*
* gregp 052808: We should silently fail here instead of throwing error. This is to deal
* with races in recovery
@@ -791,12 +674,6 @@ public final class CqServiceImpl implements CqService {
}
}
-
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeAllCqs(boolean)
- */
@Override
public void closeAllCqs(boolean clientInitiated) {
closeAllCqs(clientInitiated, getAllCqs());
@@ -807,21 +684,13 @@ public final class CqServiceImpl implements CqService {
* CqQuerys created by other VMs are unaffected.
*/
private void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs) {
- closeAllCqs(clientInitiated, cqs,
- ((GemFireCacheImpl) this.cache).keepDurableSubscriptionsAlive());
+ closeAllCqs(clientInitiated, cqs, this.cache.keepDurableSubscriptionsAlive());
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeAllCqs(boolean,
- * org.apache.geode.cache.query.CqQuery[], boolean)
- */
@Override
public void closeAllCqs(boolean clientInitiated, Collection<? extends InternalCqQuery> cqs,
boolean keepAlive) {
- // CqQuery[] cqs = getAllCqs();
if (cqs != null) {
String cqName = null;
if (logger.isDebugEnabled()) {
@@ -830,7 +699,6 @@ public final class CqServiceImpl implements CqService {
for (InternalCqQuery cQuery : cqs) {
try {
cqName = cQuery.getName();
- // boolean keepAlive = ((GemFireCache)this.cache).keepDurableSubscriptionsAlive();
if (isServer()) {
cQuery.close(false);
@@ -847,47 +715,26 @@ public final class CqServiceImpl implements CqService {
}
}
}
- } catch (QueryException cqe) {
+ } catch (QueryException | CqClosedException e) {
if (!isRunning()) {
// Not cache shutdown
logger
.warn(LocalizedMessage.create(LocalizedStrings.CqService_FAILED_TO_CLOSE_CQ__0___1,
- new Object[] {cqName, cqe.getMessage()}));
+ new Object[] {cqName, e.getMessage()}));
}
if (logger.isDebugEnabled()) {
- logger.debug(cqe.getMessage(), cqe);
- }
- } catch (CqClosedException cqe) {
- if (!isRunning()) {
- // Not cache shutdown
- logger
- .warn(LocalizedMessage.create(LocalizedStrings.CqService_FAILED_TO_CLOSE_CQ__0___1,
- new Object[] {cqName, cqe.getMessage()}));
- }
- if (logger.isDebugEnabled()) {
- logger.debug(cqe.getMessage(), cqe);
+ logger.debug(e.getMessage(), e);
}
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getCqStatistics()
- */
@Override
public CqServiceStatistics getCqStatistics() {
return cqServiceStats;
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#closeClientCqs(org.apache.geode.
- * internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public void closeClientCqs(ClientProxyMembershipID clientProxyId) throws CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
@@ -899,30 +746,19 @@ public final class CqServiceImpl implements CqService {
CqQueryImpl cQuery = (CqQueryImpl) cq;
try {
cQuery.close(false);
- } catch (QueryException qe) {
+ } catch (QueryException | CqClosedException e) {
if (isDebugEnabled) {
logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- qe.getMessage());
- }
- } catch (CqClosedException cce) {
- if (isDebugEnabled) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- cce.getMessage());
+ e.getMessage());
}
}
}
}
- /*
- * (non-Javadoc)
- *
- * @see org.apache.geode.cache.query.internal.InternalCqService#getAllClientCqs(org.apache.geode.
- * internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public List<ServerCQ> getAllClientCqs(ClientProxyMembershipID clientProxyId) {
Collection<? extends InternalCqQuery> cqs = getAllCqs();
- ArrayList<ServerCQ> clientCqs = new ArrayList<ServerCQ>();
+ ArrayList<ServerCQ> clientCqs = new ArrayList<>();
for (InternalCqQuery cq : cqs) {
ServerCQImpl cQuery = (ServerCQImpl) cq;
@@ -934,23 +770,16 @@ public final class CqServiceImpl implements CqService {
return clientCqs;
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#getAllDurableClientCqs(org.apache.geode
- * .internal.cache.tier.sockets.ClientProxyMembershipID)
- */
@Override
public List<String> getAllDurableClientCqs(ClientProxyMembershipID clientProxyId)
throws CqException {
if (clientProxyId == null) {
throw new CqException(
LocalizedStrings.CqService_UNABLE_TO_RETRIEVE_DURABLE_CQS_FOR_CLIENT_PROXY_ID
- .toLocalizedString(clientProxyId));
+ .toLocalizedString(null));
}
List<ServerCQ> cqs = getAllClientCqs(clientProxyId);
- ArrayList<String> durableClientCqs = new ArrayList<String>();
+ ArrayList<String> durableClientCqs = new ArrayList<>();
for (ServerCQ cq : cqs) {
ServerCQImpl cQuery = (ServerCQImpl) cq;
@@ -966,9 +795,6 @@ public final class CqServiceImpl implements CqService {
/**
* Server side method. Closes non-durable CQs for the given client proxy id.
- *
- * @param clientProxyId
- * @throws CqException
*/
@Override
public void closeNonDurableClientCqs(ClientProxyMembershipID clientProxyId) throws CqException {
@@ -983,15 +809,10 @@ public final class CqServiceImpl implements CqService {
if (!cQuery.isDurable()) {
cQuery.close(false);
}
- } catch (QueryException qe) {
- if (isDebugEnabled) {
- logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- qe.getMessage());
- }
- } catch (CqClosedException cce) {
+ } catch (QueryException | CqClosedException e) {
if (isDebugEnabled) {
logger.debug("Failed to close the CQ, CqName : {} Error : {}", cQuery.getName(),
- cce.getMessage());
+ e.getMessage());
}
}
}
@@ -1028,6 +849,7 @@ public final class CqServiceImpl implements CqService {
return this.isRunning;
}
+ @Override
public void start() {
this.isRunning = true;
}
@@ -1035,9 +857,10 @@ public final class CqServiceImpl implements CqService {
/**
* @return Returns the serverCqName.
*/
+ @Override
public String constructServerCqName(String cqName, ClientProxyMembershipID clientProxyId) {
- ConcurrentHashMap<ClientProxyMembershipID, String> cache = serverCqNameCache
- .computeIfAbsent(cqName, key -> new ConcurrentHashMap<ClientProxyMembershipID, String>());
+ ConcurrentHashMap<ClientProxyMembershipID, String> cache =
+ serverCqNameCache.computeIfAbsent(cqName, key -> new ConcurrentHashMap<>());
String cName = cache.get(clientProxyId);
if (null == cName) {
@@ -1065,7 +888,7 @@ public final class CqServiceImpl implements CqService {
}
}
- /*
+ /**
* Checks if CQ with the given name already exists.
*
* @param cqName name of the CQ.
@@ -1073,17 +896,15 @@ public final class CqServiceImpl implements CqService {
* @return true if exists else false.
*/
private synchronized boolean isCqExists(String cqName) {
- boolean status = false;
HashMap<String, CqQueryImpl> cqMap = cqQueryMap;
- status = cqMap.containsKey(cqName);
- return status;
+ return cqMap.containsKey(cqName);
}
- /*
+ /**
* Generates a name for CQ. Checks if CQ with that name already exists if so generates a new
* cqName.
*/
- public synchronized String generateCqName() {
+ private synchronized String generateCqName() {
while (true) {
String cqName = CQ_NAME_PREFIX + (cqId++);
if (!isCqExists(cqName)) {
@@ -1092,18 +913,9 @@ public final class CqServiceImpl implements CqService {
}
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#dispatchCqListeners(java.util.HashMap,
- * int, java.lang.Object, java.lang.Object, byte[],
- * org.apache.geode.cache.client.internal.QueueManager, org.apache.geode.internal.cache.EventID)
- */
@Override
public void dispatchCqListeners(HashMap<String, Integer> cqs, int messageType, Object key,
Object value, byte[] delta, QueueManager qManager, EventID eventId) {
- ClientCQImpl cQuery = null;
Object[] fullValue = new Object[1];
Iterator<Map.Entry<String, Integer>> iter = cqs.entrySet().iterator();
String cqName = null;
@@ -1112,7 +924,7 @@ public final class CqServiceImpl implements CqService {
try {
Map.Entry<String, Integer> entry = iter.next();
cqName = entry.getKey();
- cQuery = (ClientCQImpl) this.getCq(cqName);
+ ClientCQImpl cQuery = (ClientCQImpl) this.getCq(cqName);
if (cQuery == null || (!cQuery.isRunning() && cQuery.getQueuedEvents() == null)) {
if (isDebugEnabled) {
@@ -1122,7 +934,7 @@ public final class CqServiceImpl implements CqService {
continue;
}
- Integer cqOp = (Integer) entry.getValue();
+ Integer cqOp = entry.getValue();
// If Region destroy event, close the cq.
if (cqOp.intValue() == MessageType.DESTROY_REGION) {
@@ -1136,8 +948,7 @@ public final class CqServiceImpl implements CqService {
}
// Construct CqEvent.
- CqEventImpl cqEvent = null;
- cqEvent = new CqEventImpl(cQuery, getOperation(messageType), getOperation(cqOp.intValue()),
+ CqEventImpl cqEvent = new CqEventImpl(cQuery, getOperation(messageType), getOperation(cqOp),
key, value, delta, qManager, eventId);
// Update statistics
@@ -1181,11 +992,11 @@ public final class CqServiceImpl implements CqService {
} // iteration.
}
- public void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent) {
+ void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent) {
invokeListeners(cqName, cQuery, cqEvent, null);
}
- public void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent,
+ private void invokeListeners(String cqName, ClientCQImpl cQuery, CqEventImpl cqEvent,
Object[] fullValue) {
if (!cQuery.isRunning() || cQuery.getCqAttributes() == null) {
return;
@@ -1217,8 +1028,8 @@ public final class CqServiceImpl implements CqService {
}
Part result = (Part) GetEventValueOp
.executeOnPrimary(cqEvent.getQueueManager().getPool(), cqEvent.getEventID(), null);
- Object newVal = null;
- if (result == null || (newVal = result.getObject()) == null) {
+ Object newVal = result.getObject();
+ if (result == null || newVal == null) {
if (!cache.getCancelCriterion().isCancelInProgress()) {
Exception ex =
new Exception("Failed to retrieve full value from server for eventID "
@@ -1231,7 +1042,7 @@ public final class CqServiceImpl implements CqService {
}
}
} else {
- ((GemFireCacheImpl) this.cache).getCachePerfStats().incDeltaFullValuesRequested();
+ this.cache.getCachePerfStats().incDeltaFullValuesRequested();
cqEvent = new CqEventImpl(cQuery, cqEvent.getBaseOperation(),
cqEvent.getQueryOperation(), cqEvent.getKey(), newVal, cqEvent.getDeltaValue(),
cqEvent.getQueueManager(), cqEvent.getEventID());
@@ -1278,7 +1089,7 @@ public final class CqServiceImpl implements CqService {
}
}
- public void invokeCqConnectedListeners(String cqName, ClientCQImpl cQuery, boolean connected) {
+ private void invokeCqConnectedListeners(String cqName, ClientCQImpl cQuery, boolean connected) {
if (!cQuery.isRunning() || cQuery.getCqAttributes() == null) {
return;
}
@@ -1335,12 +1146,8 @@ public final class CqServiceImpl implements CqService {
}
}
-
/**
* Returns the Operation for the given EnumListenerEvent type.
- *
- * @param eventType
- * @return Operation
*/
private Operation getOperation(int eventType) {
Operation op = null;
@@ -1372,15 +1179,6 @@ public final class CqServiceImpl implements CqService {
return op;
}
- /*
- * (non-Javadoc)
- *
- * @see
- * org.apache.geode.cache.query.internal.InternalCqService#processEvents(org.apache.geode.cache.
- * CacheEvent, org.apache.geode.distributed.internal.DistributionAdvisor.Profile,
- * org.apache.geode.distributed.internal.DistributionAdvisor.Profile[],
- * org.apache.geode.internal.cache.FilterRoutingInfo)
- */
@Override
public void processEvents(CacheEvent event, Profile localProfile, Profile[] profiles,
FilterRoutingInfo frInfo) throws CqException {
@@ -1421,7 +1219,7 @@ public final class CqServiceImpl implements CqService {
continue;
}
Map cqs = pf.getCqMap();
- HashMap<Long, Integer> cqInfo = new HashMap<Long, Integer>();
+ HashMap<Long, Integer> cqInfo = new HashMap<>();
Iterator cqIter = cqs.entrySet().iterator();
while (cqIter.hasNext()) {
Map.Entry cqEntry = (Map.Entry) cqIter.next();
@@ -1454,10 +1252,10 @@ public final class CqServiceImpl implements CqService {
private void processEntryEvent(CacheEvent event, Profile localProfile, Profile[] profiles,
FilterRoutingInfo frInfo) throws CqException {
final boolean isDebugEnabled = logger.isDebugEnabled();
- HashSet<Object> cqUnfilteredEventsSet_newValue = new HashSet<Object>();
- HashSet<Object> cqUnfilteredEventsSet_oldValue = new HashSet<Object>();
- boolean b_cqResults_newValue = false;
- boolean b_cqResults_oldValue = false;
+ HashSet<Object> cqUnfilteredEventsSet_newValue = new HashSet<>();
+ HashSet<Object> cqUnfilteredEventsSet_oldValue = new HashSet<>();
+ boolean b_cqResults_newValue;
+ boolean b_cqResults_oldValue;
boolean queryOldValue;
EntryEvent entryEvent = (EntryEvent) event;
Object eventKey = entryEvent.getKey();
@@ -1472,8 +1270,8 @@ public final class CqServiceImpl implements CqService {
|| event.getOperation().isDestroy() || event.getOperation().isInvalidate()
|| (event.getOperation().isCreate() && isDupEvent));
- HashMap<String, Integer> matchedCqs = new HashMap<String, Integer>();
- long executionStartTime = 0;
+ HashMap<String, Integer> matchedCqs = new HashMap<>();
+ long executionStartTime;
for (int i = -1; i < profiles.length; i++) {
CacheProfile cf;
if (i < 0) {
@@ -1498,7 +1296,6 @@ public final class CqServiceImpl implements CqService {
continue;
}
-
// Get new value. If its not retrieved.
if (cqUnfilteredEventsSet_newValue.isEmpty()
&& (event.getOperation().isCreate() || event.getOperation().isUpdate())) {
@@ -1509,7 +1306,7 @@ public final class CqServiceImpl implements CqService {
}
}
- HashMap<Long, Integer> cqInfo = new HashMap<Long, Integer>();
+ HashMap<Long, Integer> cqInfo = new HashMap<>();
Iterator cqIter = cqs.entrySet().iterator();
while (cqIter.hasNext()) {
@@ -1546,7 +1343,6 @@ public final class CqServiceImpl implements CqService {
}
} else {
boolean error = false;
- // synchronized (cQuery)
{
try {
synchronized (cQuery) {
@@ -1644,7 +1440,7 @@ public final class CqServiceImpl implements CqService {
cQuery.markAsDestroyedInCqResultKeys(eventKey);
}
}
- } // end synchronized(cQuery)
+ }
// Get the matching CQs if any.
// synchronized (this.matchingCqMap){
@@ -1663,7 +1459,6 @@ public final class CqServiceImpl implements CqService {
}
}
}
- // }
}
if (cqEvent != null && cQuery.isRunning()) {
@@ -1694,153 +1489,35 @@ public final class CqServiceImpl implements CqService {
} // iteration over Profiles.
}
-
- /*
- * public void processEvents (EnumListenerEvent operation, CacheEvent event, ClientUpdateMessage
- * clientMessage, CM<ClientProxyMembershipID, CM<CqQuery, Boolean>> clientIds) throws CqException
- * {
- *
- * //Is this a region event or an entry event if (event instanceof RegionEvent){
- * processRegionEvent(operation, event, clientMessage, clientIds); } else { processEntryEvent
- * (operation, event, clientMessage, clientIds); }
- *
- * }
- *
- * private void processRegionEvent(EnumListenerEvent operation, CacheEvent event,
- * ClientUpdateMessage clientMessage, CM<ClientProxyMembershipID, CM<CqQuery, Boolean>> clientIds)
- * throws CqException {
- *
- * if (logger.isDebugEnabled()) { logger.debug("Processing region event for region " +
- * ((LocalRegion)(event.getRegion())).getName()); } HashMap filteredCqs = new HashMap(); Integer
- * cqRegionEvent = generateCqRegionEvent(operation); Iterator it =
- * clientIds.entrySet().iterator(); while (it.hasNext()) { Map.Entry me = (Map.Entry)it.next();
- * ClientProxyMembershipID clientId = (ClientProxyMembershipID)me.getKey(); CM cqsToBooleans =
- * (CM)me.getValue(); if (cqsToBooleans == null) { continue; } Set<CqQuery> cqs =
- * cqsToBooleans.keySet(); if (cqs.isEmpty()) { continue; } filteredCqs.clear(); Iterator cqIt =
- * cqs.iterator(); while (cqIt.hasNext()) { CqQueryImpl cQuery = (CqQueryImpl)cqIt.next(); if
- * (operation == EnumListenerEvent.AFTER_REGION_DESTROY) { try { if (logger.isDebugEnabled()){
- * logger.debug("Closing CQ on region destroy event. CqName :" + cQuery.getName()); }
- * cQuery.close(false); } catch (Exception ex) {
- * logger.debug("Failed to Close CQ on region destroy. CqName :" + cQuery.getName(), ex); }
- *
- * } filteredCqs.put(cQuery.cqName, cqRegionEvent);
- * cQuery.getVsdStats().updateStats(cqRegionEvent);
- *
- * } if (!filteredCqs.isEmpty()){ ((ClientUpdateMessageImpl)clientMessage).addClientCqs( clientId,
- * filteredCqs); }
- *
- * }
- *
- * }
- *
- * private void processEntryEvent(EnumListenerEvent operation, CacheEvent event,
- * ClientUpdateMessage clientMessage, CM<ClientProxyMembershipID, CM<CqQuery, Boolean>> clientIds)
- * throws CqException { HashSet cqUnfilteredEventsSet_newValue = new HashSet(); HashSet
- * cqUnfilteredEventsSet_oldValue = new HashSet(); boolean b_cqResults_newValue = false; boolean
- * b_cqResults_oldValue = false; EntryEvent entryEvent = (EntryEvent)event; Object eventKey =
- * entryEvent.getKey(); if (operation == EnumListenerEvent.AFTER_CREATE || operation ==
- * EnumListenerEvent.AFTER_UPDATE) { if (entryEvent.getNewValue() != null) { //We have a new value
- * to run the query on cqUnfilteredEventsSet_newValue.clear();
- * cqUnfilteredEventsSet_newValue.add(entryEvent.getNewValue()); } }
- *
- * HashMap matchedCqs = new HashMap(); long executionStartTime = 0; Iterator it =
- * clientIds.entrySet().iterator(); while (it.hasNext()) { Map.Entry me = (Map.Entry)it.next();
- * ClientProxyMembershipID clientId = (ClientProxyMembershipID)me.getKey(); if
- * (logger.isDebugEnabled()) { logger.debug("Processing event for CQ filter, ClientId : " +
- * clientId); } CM cqsToBooleans = (CM)me.getValue(); if (cqsToBooleans == null) { continue; }
- * Set<CqQuery> cqs = cqsToBooleans.keySet(); if (cqs.isEmpty()) { continue; } HashMap filteredCqs
- * = new HashMap(); Iterator cqIt = cqs.iterator(); while (cqIt.hasNext()) { CqQueryImpl cQuery =
- * (CqQueryImpl)cqIt.next(); b_cqResults_newValue = false; b_cqResults_oldValue = false; if
- * (cQuery == null || !(cQuery.isRunning())){ continue; } String cqName =
- * cQuery.getServerCqName(); Integer cqEvent = null; if (matchedCqs.containsKey(cqName)) { if
- * (logger.isDebugEnabled()){ logger.
- * debug("Similar cq/query is already processed, getting the cq event-type from the matched cq.");
- * } cqEvent = (Integer)matchedCqs.get(cqName); } else { boolean error = false; boolean
- * hasSeenEvent = false; HashSet cqEventKeys = null; synchronized (cQuery) { try { // Apply query
- * on new value. if (!cqUnfilteredEventsSet_newValue.isEmpty()) { executionStartTime =
- * this.stats.startCqQueryExecution(); b_cqResults_newValue = evaluateQuery(cQuery, new Object[]
- * {cqUnfilteredEventsSet_newValue}); this.stats.endCqQueryExecution(executionStartTime); } //
- * Check if old value is cached, if not apply query on old value. if (cqToCqEventKeysMap != null)
- * { synchronized (cqToCqEventKeysMap) { if ((cqEventKeys =
- * (HashSet)cqToCqEventKeysMap.get(cqName)) != null) { hasSeenEvent =
- * cqEventKeys.contains(eventKey); } } } if (!hasSeenEvent) { // get the oldValue. // In case of
- * Update, destroy and invalidate. if (operation == EnumListenerEvent.AFTER_UPDATE || operation ==
- * EnumListenerEvent.AFTER_DESTROY || operation == EnumListenerEvent.AFTER_INVALIDATE) { if
- * (entryEvent.getOldValue() != null) { cqUnfilteredEventsSet_oldValue.clear();
- * cqUnfilteredEventsSet_oldValue.add(entryEvent.getOldValue()); // Apply query on old value.
- * executionStartTime = this.stats.startCqQueryExecution(); b_cqResults_oldValue =
- * evaluateQuery(cQuery, new Object[] {cqUnfilteredEventsSet_oldValue});
- * this.stats.endCqQueryExecution(executionStartTime); } } } } catch (Exception ex) { //Any
- * exception in running the query // should be caught here and buried //because this code is
- * running inline with the //message processing code and we don't want to //kill that thread error
- * = true; logger.info( LocalizedStrings.
- * CqService_ERROR_WHILE_PROCESSING_CQ_ON_THE_EVENT_KEY_0_CQNAME_1_CLIENTID_2_ERROR_3, new
- * Object[] { ((EntryEvent)event).getKey(), cQuery.getName(), clientId,
- * ex.getLocalizedMessage()}); }
- *
- * if (error) { cqEvent = Integer.valueOf(MessageType.EXCEPTION); } else { if
- * (b_cqResults_newValue) { if (hasSeenEvent || b_cqResults_oldValue) { cqEvent =
- * Integer.valueOf(MessageType.LOCAL_UPDATE); } else { cqEvent =
- * Integer.valueOf(MessageType.LOCAL_CREATE); } // If its create and caching is enabled, cache the
- * key for this CQ. if (!hasSeenEvent && cqEventKeys != null) { cqEventKeys.add(eventKey); } }
- * else if (hasSeenEvent || (b_cqResults_oldValue)) { // Base invalidate operation is treated as
- * destroy. // When the invalidate comes through, the entry will no longer satisfy // the query
- * and will need to be deleted. cqEvent = Integer.valueOf(MessageType.LOCAL_DESTROY); // If
- * caching is enabled, remove this event's key from the cache. if (hasSeenEvent && cqEventKeys !=
- * null) { cqEventKeys.remove(eventKey); } } }
- *
- * } //end synchronized(cQuery)
- *
- * // Get the matching CQs if any. synchronized (this.matchingCqMap){ String query =
- * cQuery.getQueryString(); ArrayList matchingCqs = (ArrayList)matchingCqMap.get(query); if
- * (matchingCqs != null) { Iterator iter = matchingCqs.iterator(); while (iter.hasNext()) { String
- * matchingCqName = (String)iter.next(); if (!matchingCqName.equals(cqName)){
- * matchedCqs.put(matchingCqName, cqEvent); } } } }
- *
- * }
- *
- * if (cqEvent != null){ if (logger.isDebugEnabled()) {
- * logger.debug("Event is added for the CQ, CqName (clientside): " + cQuery.cqName +
- * " With CQ Op : " + cqEvent + " for Client : " + clientId); } filteredCqs.put(cQuery.cqName,
- * cqEvent); cQuery.getVsdStats().updateStats(cqEvent); }
- *
- * } // iteration over cqsToBooleans.keySet() if (!filteredCqs.isEmpty()){
- * logger.debug("Adding event map for client : "+clientId +
- * " with event map size : "+filteredCqs.size());
- * ((ClientUpdateMessageImpl)clientMessage).addClientCqs(clientId, filteredCqs); } } // iteration
- * over clientIds.entrySet() }
- */
-
private Integer generateCqRegionEvent(CacheEvent event) {
Integer cqEvent = null;
if (event.getOperation().isRegionDestroy()) {
- cqEvent = Integer.valueOf(MessageType.DESTROY_REGION);
+ cqEvent = MessageType.DESTROY_REGION;
} else if (event.getOperation().isRegionInvalidate()) {
- cqEvent = Integer.valueOf(MessageType.INVALIDATE_REGION);
+ cqEvent = MessageType.INVALIDATE_REGION;
} else if (event.getOperation().isClear()) {
- cqEvent = Integer.valueOf(MessageType.CLEAR_REGION);
+ cqEvent = MessageType.CLEAR_REGION;
}
return cqEvent;
}
-
/**
* Manages the CQs created for the base region. This is managed here, instead of on the base
* region; since the cq could be created on the base region, before base region is created (using
* newCq()).
*/
- public void addToBaseRegionToCqNameMap(String regionName, String cqName) {
+ private void addToBaseRegionToCqNameMap(String regionName, String cqName) {
synchronized (this.baseRegionToCqNameMap) {
ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName);
if (cqs == null) {
- cqs = new ArrayList<String>();
+ cqs = new ArrayList<>();
}
cqs.add(cqName);
this.baseRegionToCqNameMap.put(regionName, cqs);
}
}
- public void removeFromBaseRegionToCqNameMap(String regionName, String cqName) {
+ void removeFromBaseRegionToCqNameMap(String regionName, String cqName) {
synchronized (this.baseRegionToCqNameMap) {
ArrayList<String> cqs = this.baseRegionToCqNameMap.get(regionName);
if (cqs != null) {
@@ -1864,37 +1541,12 @@ public final class CqServiceImpl implements CqService {
}
/**
- * Removes this CQ from CQ event Cache map. This disables the caching events for this CQ.
- *
- * @param cqName
- */
- /*
- * synchronized public void removeCQFromCaching(String cqName){ if (cqToCqEventKeysMap != null) {
- * // Take a lock on CqQuery object. In processEvents the maps are // handled under CqQuery
- * object. if (cqToCqEventKeysMap != null){ synchronized (cqToCqEventKeysMap) {
- * cqToCqEventKeysMap.remove(cqName); } } } }
- */
-
- /**
- * Returns the CQ event cache map.
- *
- * @return HashMap cqToCqEventKeysMap
- *
- * Caller must synchronize on the returned value in order to inspect.
- */
- /*
- * public HashMap getCqToCqEventKeysMap(){ return cqToCqEventKeysMap; }
- */
-
- /**
* Adds the query from the given CQ to the matched CQ map.
- *
- * @param cq
*/
- public void addToMatchingCqMap(CqQueryImpl cq) {
+ void addToMatchingCqMap(CqQueryImpl cq) {
synchronized (this.matchingCqMap) {
String cqQuery = cq.getQueryString();
- Set<String> matchingCQs = null;
+ Set<String> matchingCQs;
if (!matchingCqMap.containsKey(cqQuery)) {
matchingCQs = Collections.newSetFromMap(new ConcurrentHashMap());
matchingCqMap.put(cqQuery, matchingCQs);
@@ -1912,10 +1564,8 @@ public final class CqServiceImpl implements CqService {
/**
* Removes the query from the given CQ from the matched CQ map.
- *
- * @param cq
*/
- public void removeFromMatchingCqMap(CqQueryImpl cq) {
+ private void removeFromMatchingCqMap(CqQueryImpl cq) {
synchronized (this.matchingCqMap) {
String cqQuery = cq.getQueryString();
if (matchingCqMap.containsKey(cqQuery)) {
@@ -1947,10 +1597,6 @@ public final class CqServiceImpl implements CqService {
* Applies the query on the event. This method takes care of the performance related changed done
* to improve the CQ-query performance. When CQ-query is executed first time, it saves the query
* related information in the execution context and uses that info in later executions.
- *
- * @param cQuery
- * @param event
- * @return boolean
*/
private boolean evaluateQuery(CqQueryImpl cQuery, Object[] event) throws Exception {
ExecutionContext execContext = cQuery.getQueryExecutionContext();
@@ -1983,19 +1629,6 @@ public final class CqServiceImpl implements CqService {
return this.cqNameToUserAttributesMap.get(cqName);
}
- // public static void memberLeft(String poolName) {
- // if (cqServiceSingleton != null && !cqServiceSingleton.isServer()) {
- // cqServiceSingleton.sendMemberDisconnectedMessageToCqs(poolName);
- // }
- // }
- //
- // public static void memberCrashed(String poolName) {
- // if (cqServiceSingleton != null && !cqServiceSingleton.isServer()) {
- // cqServiceSingleton.sendMemberDisconnectedMessageToCqs(poolName);
- // }
- // }
- //
-
@Override
public void cqsDisconnected(Pool pool) {
invokeCqsConnected(pool, false);
@@ -2014,7 +1647,7 @@ public final class CqServiceImpl implements CqService {
// Check to see if we are already connected/disconnected.
// If state has not changed, do not invoke another connected/disconnected
synchronized (cqPoolsConnected) {
- // don't repeatily send same connect/disconnect message to cq's on repeated fails of
+ // don't repeatedly send same connect/disconnect message to cq's on repeated fails of
// RedundancySatisfier
if (cqPoolsConnected.containsKey(poolName) && connected == cqPoolsConnected.get(poolName)) {
return;
@@ -2059,13 +1692,6 @@ public final class CqServiceImpl implements CqService {
SystemFailure.checkFailure();
logger.warn(LocalizedMessage
.create(LocalizedStrings.CqService_ERROR_SENDING_CQ_CONNECTION_STATUS, cqName), t);
-
- if (t instanceof VirtualMachineError) {
- logger.warn(LocalizedMessage.create(
- LocalizedStrings.CqService_VIRTUALMACHINEERROR_PROCESSING_CQLISTENER_FOR_CQ_0,
- cqName), t);
- return;
- }
}
}
}
@@ -2075,7 +1701,4 @@ public final class CqServiceImpl implements CqService {
public List<String> getAllDurableCqsFromServer(InternalPool pool) {
return new ServerCQProxyImpl(pool).getAllDurableCqsFromServer();
}
-
-
}
-
http://git-wip-us.apache.org/repos/asf/geode/blob/363e50d2/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
----------------------------------------------------------------------
diff --git a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
index ba71143..a675162 100644
--- a/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
+++ b/geode-cq/src/main/java/org/apache/geode/cache/query/internal/cq/CqServiceStatisticsImpl.java
@@ -14,11 +14,9 @@
*/
package org.apache.geode.cache.query.internal.cq;
-import org.apache.geode.cache.CacheFactory;
import org.apache.geode.cache.query.CqServiceStatistics;
import org.apache.geode.cache.query.CqQuery;
import org.apache.geode.cache.query.internal.DefaultQueryService;
-import org.apache.geode.internal.cache.GemFireCacheImpl;
/**
* Provides statistical information about CqService.
@@ -26,24 +24,22 @@ import org.apache.geode.internal.cache.GemFireCacheImpl;
* @since GemFire 5.5
*/
public class CqServiceStatisticsImpl implements CqServiceStatistics {
+
private CqServiceImpl cqService;
- // private long activeCqs;
- // private long stoppedCqs;
- // private long closedCqs;
- // private long createdCqs;
/**
* Constructor for CqStatisticsImpl
*
* @param cqs - CqService
*/
- public CqServiceStatisticsImpl(CqServiceImpl cqs) {
+ CqServiceStatisticsImpl(CqServiceImpl cqs) {
cqService = cqs;
}
/**
* Returns the number of CQs currently executing
*/
+ @Override
public long numCqsActive() {
return this.cqService.getCqServiceVsdStats().getNumCqsActive();
}
@@ -53,6 +49,7 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
*
* @return long number of cqs created.
*/
+ @Override
public long numCqsCreated() {
return this.cqService.getCqServiceVsdStats().getNumCqsCreated();
}
@@ -60,6 +57,7 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
/**
* Returns number of Cqs that are closed.
*/
+ @Override
public long numCqsClosed() {
return this.cqService.getCqServiceVsdStats().getNumCqsClosed();
}
@@ -67,6 +65,7 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
/**
* Returns number of Cqs that are stopped.
*/
+ @Override
public long numCqsStopped() {
return this.cqService.getCqServiceVsdStats().getNumCqsStopped();
}
@@ -74,20 +73,18 @@ public class CqServiceStatisticsImpl implements CqServiceStatistics {
/**
* Returns number of CQs created from the client.
*/
+ @Override
public long numCqsOnClient() {
return this.cqService.getCqServiceVsdStats().getNumCqsOnClient();
}
/**
* Returns the number of CQs (active + suspended) on the given region.
- *
- * @param regionName
*/
+ @Override
public long numCqsOnRegion(String regionName) {
-
DefaultQueryService queryService =
- (DefaultQueryService) ((GemFireCacheImpl) CacheFactory.getAnyInstance())
- .getLocalQueryService();
+ (DefaultQueryService) cqService.getInternalCache().getLocalQueryService();
try {
CqQuery[] cqs = queryService.getCqs(regionName);