You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by rv...@apache.org on 2016/01/20 03:22:32 UTC
[26/51] [partial] incubator-geode git commit: WAN and CQ code drop
under the Pivotal SGA
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java
new file mode 100644
index 0000000..91df070
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/ClientPostAuthorizationDUnitTest.java
@@ -0,0 +1,390 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
+import java.util.Random;
+
+import security.AuthzCredentialGenerator;
+import security.CredentialGenerator;
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+import com.gemstone.gemfire.internal.AvailablePort;
+
+import dunit.Host;
+
+/**
+ * Tests for authorization from client to server. This tests for authorization
+ * with post-process callbacks in case return values of operations and for
+ * notifications along-with failover.
+ *
+ * @author sumedh
+ * @since 5.5
+ */
+public class ClientPostAuthorizationDUnitTest extends
+ ClientAuthorizationTestBase {
+
+
+ /** constructor */
+ public ClientPostAuthorizationDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+
+ super.setUp();
+ final Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+
+ server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { clientExpectedExceptions });
+ SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+ }
+
+ // Region: Tests
+
+ public void testAllPostOps() {
+
+ OperationWithAction[] allOps = {
+ // Test CREATE and verify with a GET
+ new OperationWithAction(OperationCode.PUT),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.CHECK_NOKEY, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_NOTAUTHZ, 4),
+
+ // OPBLOCK_END indicates end of an operation block that needs to
+ // be executed on each server when doing failover
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a GET
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 3,
+ OpFlags.USE_OLDCONN | OpFlags.CHECK_NOKEY | OpFlags.USE_NEWVAL
+ | OpFlags.CHECK_NOTAUTHZ, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a KEY_SET
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 6),
+ new OperationWithAction(OperationCode.KEY_SET, 2, OpFlags.NONE, 6),
+ new OperationWithAction(OperationCode.KEY_SET, 3,
+ OpFlags.CHECK_NOTAUTHZ, 6),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a QUERY
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 7),
+ new OperationWithAction(OperationCode.QUERY, 2, OpFlags.USE_NEWVAL, 7),
+ new OperationWithAction(OperationCode.QUERY, 3, OpFlags.USE_NEWVAL
+ | OpFlags.CHECK_NOTAUTHZ, 7),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a EXECUTE_CQ initial results
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 2, OpFlags.NONE, 8),
+ new OperationWithAction(OperationCode.EXECUTE_CQ, 3,
+ OpFlags.CHECK_NOTAUTHZ, 8),
+
+ OperationWithAction.OPBLOCK_END };
+
+ Iterator iter = getDummyGeneratorCombos().iterator();
+ while (iter.hasNext()) {
+ AuthzCredentialGenerator gen = (AuthzCredentialGenerator)iter.next();
+ CredentialGenerator cGen = gen.getCredentialGenerator();
+ Properties extraAuthProps = cGen.getSystemProperties();
+ Properties javaProps = cGen.getJavaProperties();
+ Properties extraAuthzProps = gen.getSystemProperties();
+ String authenticator = cGen.getAuthenticator();
+ String authInit = cGen.getAuthInit();
+ String accessor = gen.getAuthorizationCallback();
+ TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+ getLogWriter().info("testAllPostOps: Using authinit: " + authInit);
+ getLogWriter().info(
+ "testAllPostOps: Using authenticator: " + authenticator);
+ getLogWriter().info("testAllPostOps: Using accessor: " + accessor);
+
+ // Start servers with all required properties
+ Properties serverProps = buildProperties(authenticator, accessor, true,
+ extraAuthProps, extraAuthzProps);
+ // Get ports for the servers
+ Integer port1 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer port2 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+
+ // Close down any running servers
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+
+ // Perform all the ops on the clients
+ List opBlock = new ArrayList();
+ Random rnd = new Random();
+ for (int opNum = 0; opNum < allOps.length; ++opNum) {
+ // Start client with valid credentials as specified in
+ // OperationWithAction
+ OperationWithAction currentOp = allOps[opNum];
+ if (currentOp.equals(OperationWithAction.OPBLOCK_END)
+ || currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+ // End of current operation block; execute all the operations
+ // on the servers with failover
+ if (opBlock.size() > 0) {
+ // Start the first server and execute the operation block
+ server1.invoke(ClientAuthorizationTestBase.class,
+ "createCacheServer", new Object[] {
+ SecurityTestUtil.getLocatorPort(), port1, serverProps,
+ javaProps });
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+ extraAuthzProps, tgen, rnd);
+ if (!currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+ // Failover to the second server and run the block again
+ server2.invoke(ClientAuthorizationTestBase.class,
+ "createCacheServer", new Object[] {
+ SecurityTestUtil.getLocatorPort(), port2, serverProps,
+ javaProps });
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+ extraAuthzProps, tgen, rnd);
+ }
+ opBlock.clear();
+ }
+ }
+ else {
+ currentOp.setOpNum(opNum);
+ opBlock.add(currentOp);
+ }
+ }
+ }
+ }
+
+ public void testAllOpsNotifications() {
+
+ OperationWithAction[] allOps = {
+ // Test CREATE and verify with a GET
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 2, OpFlags.USE_REGEX
+ | OpFlags.REGISTER_POLICY_NONE, 8),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 3, OpFlags.USE_REGEX
+ | OpFlags.REGISTER_POLICY_NONE | OpFlags.USE_NOTAUTHZ, 8),
+ new OperationWithAction(OperationCode.PUT),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.CHECK_FAIL, 4),
+
+ // OPBLOCK_END indicates end of an operation block that needs to
+ // be executed on each server when doing failover
+ OperationWithAction.OPBLOCK_END,
+
+ // Test UPDATE and verify with a GET
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 2, OpFlags.USE_REGEX
+ | OpFlags.REGISTER_POLICY_NONE, 8),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 3, OpFlags.USE_REGEX
+ | OpFlags.REGISTER_POLICY_NONE | OpFlags.USE_NOTAUTHZ, 8),
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.USE_NEWVAL, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.USE_NEWVAL | OpFlags.CHECK_FAIL, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Test DESTROY and verify with GET that keys should not exist
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.NONE, 8),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 2, OpFlags.USE_REGEX, 8),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+ OpFlags.USE_REGEX | OpFlags.USE_OLDCONN | OpFlags.REGISTER_POLICY_NONE, 8),
+ // registerInterest now clears the keys, so a dummy put to add
+ // those keys back for the case when updates should not come
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN, 8),
+ new OperationWithAction(OperationCode.DESTROY, 1, OpFlags.USE_OLDCONN,
+ 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.CHECK_FAIL, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+ // Repopulate the region
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Do REGION_CLEAR and check with GET
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.NONE, 8),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 2, OpFlags.USE_ALL_KEYS, 1),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+ OpFlags.USE_ALL_KEYS | OpFlags.USE_OLDCONN | OpFlags.REGISTER_POLICY_NONE, 1),
+ // registerInterest now clears the keys, so a dummy put to add
+ // those keys back for the case when updates should not come
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN, 8),
+ new OperationWithAction(OperationCode.REGION_CLEAR, 1,
+ OpFlags.USE_OLDCONN, 1),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.CHECK_FAIL, 8),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 8),
+ // Repopulate the region
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN, 8),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Do REGION_CREATE and check with CREATE/GET
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 2, OpFlags.USE_ALL_KEYS | OpFlags.ENABLE_DRF, 1),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 3, OpFlags.USE_ALL_KEYS | OpFlags.ENABLE_DRF
+ | OpFlags.USE_NOTAUTHZ | OpFlags.REGISTER_POLICY_NONE, 1),
+ new OperationWithAction(OperationCode.REGION_CREATE, 1,
+ OpFlags.ENABLE_DRF, 1),
+ new OperationWithAction(OperationCode.PUT, 1, OpFlags.USE_OLDCONN
+ | OpFlags.USE_SUBREGION, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+ | OpFlags.NO_CREATE_SUBREGION, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+ | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOREGION, 4),
+
+ // Do REGION_DESTROY of the sub-region and check with GET
+ new OperationWithAction(OperationCode.REGION_DESTROY, 1,
+ OpFlags.USE_OLDCONN | OpFlags.USE_SUBREGION, 1),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+ | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOREGION, 4),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.USE_SUBREGION | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_EXCEPTION, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.USE_SUBREGION
+ | OpFlags.NO_CREATE_SUBREGION | OpFlags.CHECK_NOREGION, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.USE_SUBREGION | OpFlags.CHECK_NOKEY
+ | OpFlags.CHECK_EXCEPTION, 4),
+
+ OperationWithAction.OPBLOCK_END,
+
+ // Do REGION_DESTROY of the region and check with GET
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.NONE, 8),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST,
+ OperationCode.GET, 2, OpFlags.USE_ALL_KEYS, 1),
+ new OperationWithAction(OperationCode.REGISTER_INTEREST, 3,
+ OpFlags.USE_ALL_KEYS | OpFlags.USE_OLDCONN | OpFlags.REGISTER_POLICY_NONE, 1),
+ // registerInterest now clears the keys, so a dummy put to add
+ // those keys back for the case when updates should not come
+ new OperationWithAction(OperationCode.PUT, 3, OpFlags.USE_OLDCONN, 8),
+ new OperationWithAction(OperationCode.REGION_DESTROY, 1, OpFlags.NONE,
+ 1),
+ new OperationWithAction(OperationCode.GET, 2, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP | OpFlags.CHECK_NOREGION, 4),
+ new OperationWithAction(OperationCode.GET, 3, OpFlags.USE_OLDCONN
+ | OpFlags.LOCAL_OP, 4),
+
+ OperationWithAction.OPBLOCK_NO_FAILOVER };
+
+ AuthzCredentialGenerator gen = getXmlAuthzGenerator();
+ getLogWriter().info("Executing opblocks with credential generator " + gen);
+ CredentialGenerator cGen = gen.getCredentialGenerator();
+ Properties extraAuthProps = cGen.getSystemProperties();
+ Properties javaProps = cGen.getJavaProperties();
+ Properties extraAuthzProps = gen.getSystemProperties();
+ String authenticator = cGen.getAuthenticator();
+ String authInit = cGen.getAuthInit();
+ String accessor = gen.getAuthorizationCallback();
+ TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+ getLogWriter().info(
+ "testAllOpsNotifications: Using authinit: " + authInit);
+ getLogWriter().info(
+ "testAllOpsNotifications: Using authenticator: " + authenticator);
+ getLogWriter().info(
+ "testAllOpsNotifications: Using accessor: " + accessor);
+
+ // Start servers with all required properties
+ Properties serverProps = buildProperties(authenticator, accessor, true,
+ extraAuthProps, extraAuthzProps);
+ // Get ports for the servers
+ Integer port1 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer port2 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+
+ // Perform all the ops on the clients
+ List opBlock = new ArrayList();
+ Random rnd = new Random();
+ for (int opNum = 0; opNum < allOps.length; ++opNum) {
+ // Start client with valid credentials as specified in
+ // OperationWithAction
+ OperationWithAction currentOp = allOps[opNum];
+ if (currentOp.equals(OperationWithAction.OPBLOCK_END)
+ || currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+ // End of current operation block; execute all the operations
+ // on the servers with failover
+ if (opBlock.size() > 0) {
+ // Start the first server and execute the operation block
+ server1.invoke(ClientAuthorizationTestBase.class,
+ "createCacheServer", new Object[] {
+ SecurityTestUtil.getLocatorPort(), port1, serverProps,
+ javaProps });
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+ extraAuthzProps, tgen, rnd);
+ if (!currentOp.equals(OperationWithAction.OPBLOCK_NO_FAILOVER)) {
+ // Failover to the second server and run the block again
+ server2.invoke(ClientAuthorizationTestBase.class,
+ "createCacheServer", new Object[] {
+ SecurityTestUtil.getLocatorPort(), port2, serverProps,
+ javaProps });
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ executeOpBlock(opBlock, port1, port2, authInit, extraAuthProps,
+ extraAuthzProps, tgen, rnd);
+ }
+ opBlock.clear();
+ }
+ }
+ else {
+ currentOp.setOpNum(opNum);
+ opBlock.add(currentOp);
+ }
+ }
+ }
+
+ // End Region: Tests
+
+ public void tearDown2() throws Exception {
+
+ super.tearDown2();
+ // close the clients first
+ client1.invoke(SecurityTestUtil.class, "closeCache");
+ client2.invoke(SecurityTestUtil.class, "closeCache");
+ SecurityTestUtil.closeCache();
+ // then close the servers
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java
new file mode 100644
index 0000000..9c5d18a
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserAPIDUnitTest.java
@@ -0,0 +1,382 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.security;
+
+import hydra.Log;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLHandshakeException;
+
+import security.CredentialGenerator;
+
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.execute.FunctionService;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqException;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.Query;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.cache.PoolManagerImpl;
+
+import dunit.Host;
+import dunit.VM;
+import security.DummyCredentialGenerator;
+
+public class MultiuserAPIDUnitTest extends ClientAuthorizationTestBase {
+
+ /** constructor */
+ public MultiuserAPIDUnitTest(String name) {
+ super(name);
+ }
+
+ private VM server1 = null;
+
+ private VM server2 = null;
+
+ private VM client1 = null;
+
+ private VM client2 = null;
+
+ private static final String[] serverExpectedExceptions = {
+ AuthenticationRequiredException.class.getName(),
+ AuthenticationFailedException.class.getName(),
+ GemFireSecurityException.class.getName(),
+ ClassNotFoundException.class.getName(), IOException.class.getName(),
+ SSLException.class.getName(), SSLHandshakeException.class.getName()};
+
+ private static final String[] clientExpectedExceptions = {
+ AuthenticationRequiredException.class.getName(),
+ AuthenticationFailedException.class.getName(),
+ SSLHandshakeException.class.getName()};
+
+ public void setUp() throws Exception {
+ super.setUp();
+ final Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+
+ server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] {serverExpectedExceptions});
+ server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] {serverExpectedExceptions});
+ client1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] {clientExpectedExceptions});
+ client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] {clientExpectedExceptions});
+ }
+
+ public static Integer createCacheServer(Object dsPort, Object locatorString,
+ Object authenticator, Object extraProps, Object javaProps) {
+
+ Properties authProps;
+ if (extraProps == null) {
+ authProps = new Properties();
+ } else {
+ authProps = (Properties)extraProps;
+ }
+ if (authenticator != null) {
+ authProps.setProperty(
+ DistributionConfig.SECURITY_CLIENT_AUTHENTICATOR_NAME, authenticator
+ .toString());
+ }
+ return SecurityTestUtil.createCacheServer(authProps, javaProps,
+ (Integer)dsPort, (String)locatorString, null, new Integer(
+ SecurityTestUtil.NO_EXCEPTION));
+ }
+
+ private static void createCacheClient(Object authInit, Properties authProps,
+ Properties javaProps, Integer[] ports, Object numConnections,
+ Boolean multiUserMode, Integer expectedResult) {
+
+ String authInitStr = (authInit == null ? null : authInit.toString());
+ SecurityTestUtil.createCacheClient(authInitStr, authProps, javaProps,
+ ports, numConnections, multiUserMode.toString(), expectedResult);
+ }
+
+ public static void createCacheClient(Object authInit, Object authProps,
+ Object javaProps, Integer port1, Integer port2, Object numConnections,
+ Boolean multiUserMode, Integer expectedResult) {
+
+ createCacheClient(authInit, (Properties)authProps, (Properties)javaProps,
+ new Integer[] {port1, port2}, numConnections, multiUserMode,
+ expectedResult);
+ }
+
+ public static void registerAllInterest() {
+ Region region = SecurityTestUtil.getCache().getRegion(
+ SecurityTestUtil.regionName);
+ assertNotNull(region);
+ region.registerInterestRegex(".*");
+ }
+
+ private void setUpVMs(CredentialGenerator gen, Boolean multiUser) {
+ Properties extraProps = gen.getSystemProperties();
+ Properties javaProps = gen.getJavaProperties();
+ String authenticator = gen.getAuthenticator();
+ String authInit = gen.getAuthInit();
+
+ getLogWriter().info(
+ "testValidCredentials: Using scheme: " + gen.classCode());
+ getLogWriter().info(
+ "testValidCredentials: Using authenticator: " + authenticator);
+ getLogWriter().info("testValidCredentials: Using authinit: " + authInit);
+
+ // Start the servers
+ Integer locPort1 = SecurityTestUtil.getLocatorPort();
+ Integer locPort2 = SecurityTestUtil.getLocatorPort();
+ String locString = SecurityTestUtil.getLocatorString();
+ Integer port1 = (Integer)server1.invoke(MultiuserAPIDUnitTest.class,
+ "createCacheServer", new Object[] {locPort1, locString, authenticator,
+ extraProps, javaProps});
+ Integer port2 = (Integer)server2.invoke(MultiuserAPIDUnitTest.class,
+ "createCacheServer", new Object[] {locPort2, locString, authenticator,
+ extraProps, javaProps});
+
+ // Start the clients with valid credentials
+ Properties credentials1 = gen.getValidCredentials(1);
+ Properties javaProps1 = gen.getJavaProperties();
+ getLogWriter().info(
+ "testValidCredentials: For first client credentials: " + credentials1
+ + " : " + javaProps1);
+ Properties credentials2 = gen.getValidCredentials(2);
+ Properties javaProps2 = gen.getJavaProperties();
+ getLogWriter().info(
+ "testValidCredentials: For second client credentials: " + credentials2
+ + " : " + javaProps2);
+ client1.invoke(MultiuserAPIDUnitTest.class, "createCacheClient",
+ new Object[] {authInit, credentials1, javaProps1, port1, port2, null,
+ multiUser, new Integer(SecurityTestUtil.NO_EXCEPTION)});
+ }
+
+ public void testSingleUserUnsupportedAPIs() {
+ // Start servers
+ // Start clients with multiuser-authentication set to false
+ setUpVMs(new DummyCredentialGenerator(), Boolean.FALSE);
+ client1.invoke(MultiuserAPIDUnitTest.class, "verifyDisallowedOps",
+ new Object[] {Boolean.FALSE});
+ }
+
+ public void testMultiUserUnsupportedAPIs() {
+ // Start servers.
+ // Start clients with multiuser-authentication set to true.
+ setUpVMs(new DummyCredentialGenerator(), Boolean.TRUE);
+ client1.invoke(MultiuserAPIDUnitTest.class, "verifyDisallowedOps",
+ new Object[] {Boolean.TRUE});
+ }
+
+ public static void verifyDisallowedOps(Boolean multiuserMode) {
+ String op = "unknown";
+ boolean success = false;
+ if (!multiuserMode) {
+ success = false;
+ try {
+ // Attempt cache.createAuthenticatedCacheView() and expect an exception, fail otherwise
+ op = "Pool.createSecureUserCache()";
+ GemFireCacheImpl.getInstance().createAuthenticatedView(new Properties(), "testPool");
+ } catch (IllegalStateException uoe) {
+ Log.getLogWriter().info(op + ": Got expected exception: " + uoe);
+ success = true;
+ } catch (Exception e) {
+ fail("Got unexpected exception while doing " + op, e);
+ }
+ if (!success) {
+ fail("Did not get exception while doing " + op);
+ }
+ } else { // multiuser mode
+ Region realRegion = GemFireCacheImpl.getInstance().getRegion(
+ SecurityTestUtil.regionName);
+ Region proxyRegion = SecurityTestUtil.proxyCaches[0]
+ .getRegion(SecurityTestUtil.regionName);
+ Pool pool = PoolManagerImpl.getPMI().find("testPool");
+ for (int i = 0; i <= 27; i++) {
+ success = false;
+ try {
+ switch (i) {
+ // Attempt (real) Region.create/put/get/containsKeyOnServer/destroy/
+ // destroyRegion/clear/remove/registerInterest/unregisterInterest()
+ // and expect an exception, fail otherwise.
+ case 0:
+ op = "Region.create()";
+ realRegion.create("key", "value");
+ break;
+ case 1:
+ op = "Region.put()";
+ realRegion.put("key", "value");
+ break;
+ case 2:
+ op = "Region.get()";
+ realRegion.get("key");
+ break;
+ case 3:
+ op = "Region.containsKeyOnServer()";
+ realRegion.containsKeyOnServer("key");
+ break;
+ case 4:
+ op = "Region.remove()";
+ realRegion.remove("key");
+ break;
+ case 5:
+ op = "Region.destroy()";
+ realRegion.destroy("key");
+ break;
+ case 6:
+ op = "Region.destroyRegion()";
+ realRegion.destroyRegion();
+ break;
+ case 7:
+ op = "Region.registerInterest()";
+ realRegion.registerInterest("key");
+ break;
+ // case 8:
+ // op = "Region.unregisterInterest()";
+ // realRegion.unregisterInterest("key");
+ // break;
+ case 8:
+ op = "Region.clear()";
+ realRegion.clear();
+ break;
+ // Attempt ProxyRegion.createSubregion/forceRolling/
+ // getAttributesMutator/registerInterest/loadSnapShot/saveSnapshot/
+ // setUserAttribute/unregisterInterest/writeToDisk
+ // and expect an exception, fail otherwise.
+ case 9:
+ op = "ProxyRegion.createSubregion()";
+ proxyRegion.createSubregion("subregion",
+ null);
+ break;
+ case 10:
+ op = "ProxyRegion.forceRolling()";
+ proxyRegion.forceRolling();
+ break;
+ case 11:
+ op = "ProxyRegion.getAttributesMutator()";
+ proxyRegion.getAttributesMutator();
+ break;
+ case 12:
+ op = "ProxyRegion.registerInterest()";
+ proxyRegion.registerInterest("key");
+ break;
+ case 13:
+ op = "ProxyRegion.loadSnapshot()";
+ proxyRegion.loadSnapshot(null);
+ break;
+ case 14:
+ op = "ProxyRegion.saveSnapshot()";
+ proxyRegion.saveSnapshot(null);
+ break;
+ case 15:
+ op = "ProxyRegion.setUserAttribute()";
+ proxyRegion.setUserAttribute(null);
+ break;
+ case 16:
+ op = "ProxyRegion.unregisterInterestRegex()";
+ proxyRegion.unregisterInterestRegex("*");
+ break;
+ // Attempt FunctionService.onRegion/onServer/s(pool) and expect an
+ // exception, fail otherwise.
+ case 17:
+ op = "FunctionService.onRegion()";
+ FunctionService.onRegion(realRegion);
+ break;
+ case 18:
+ op = "FunctionService.onServer(pool)";
+ FunctionService.onServer(pool);
+ break;
+ case 19:
+ op = "FunctionService.onServers(pool)";
+ FunctionService.onServers(pool);
+ break;
+ // Attempt
+ // QueryService.newQuery().execute()/newCq().execute/executeWithInitialResults()
+ case 20:
+ op = "QueryService.newQuery.execute()";
+ Query query = pool.getQueryService().newQuery(
+ "SELECT * FROM /" + SecurityTestUtil.regionName);
+ query.execute();
+ break;
+ case 21:
+ op = "QueryService.newCq.execute()";
+ CqQuery cqQuery = pool.getQueryService().newCq(
+ "SELECT * FROM /" + SecurityTestUtil.regionName,
+ new CqAttributesFactory().create());
+ try {
+ cqQuery.execute();
+ } catch (CqException ce) {
+ throw (Exception)ce.getCause();
+ }
+ break;
+ case 22:
+ op = "QueryService.newCq.executeWithInitialResults()";
+ cqQuery = pool.getQueryService().newCq(
+ "SELECT * FROM /" + SecurityTestUtil.regionName,
+ new CqAttributesFactory().create());
+ try {
+ cqQuery.executeWithInitialResults();
+ } catch (CqException ce) {
+ throw (Exception)ce.getCause();
+ }
+ break;
+ // Attempt ProxyQueryService.getIndex/createIndex/removeIndex() and
+ // expect an exception, fail otherwise.
+ case 23:
+ op = "ProxyQueryService().getIndexes()";
+ SecurityTestUtil.proxyCaches[0].getQueryService()
+ .getIndexes(null);
+ break;
+ case 24:
+ op = "ProxyQueryService().createIndex()";
+ SecurityTestUtil.proxyCaches[0].getQueryService().createIndex(
+ null, null, null );
+ break;
+ case 25:
+ op = "ProxyQueryService().removeIndexes()";
+ SecurityTestUtil.proxyCaches[0].getQueryService().removeIndexes();
+ break;
+ case 26:
+ op = "ProxyRegion.localDestroy()";
+ proxyRegion.localDestroy("key");
+ break;
+ case 27:
+ op = "ProxyRegion.localInvalidate()";
+ proxyRegion.localInvalidate("key");
+ break;
+ default:
+ fail("Unknown op code: " + i);
+ break;
+ }
+ } catch (UnsupportedOperationException uoe) {
+ Log.getLogWriter().info(op + ": Got expected exception: " + uoe);
+ success = true;
+ } catch (Exception e) {
+ fail("Got unexpected exception while doing " + op, e);
+ }
+ if (!success) {
+ fail("Did not get exception while doing " + op);
+ }
+ }
+ }
+ }
+
+ public void tearDown2() throws Exception {
+ super.tearDown2();
+ // close the clients first
+ client1.invoke(SecurityTestUtil.class, "closeCache");
+ client2.invoke(SecurityTestUtil.class, "closeCache");
+ // then close the servers
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java
new file mode 100644
index 0000000..99c55b1
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/security/MultiuserDurableCQAuthzDUnitTest.java
@@ -0,0 +1,484 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+/**
+ *
+ */
+package com.gemstone.gemfire.security;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Random;
+
+import security.AuthzCredentialGenerator;
+import security.CredentialGenerator;
+
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.operations.OperationContext.OperationCode;
+import com.gemstone.gemfire.cache.query.CqAttributes;
+import com.gemstone.gemfire.cache.query.CqAttributesFactory;
+import com.gemstone.gemfire.cache.query.CqListener;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.query.SelectResults;
+import com.gemstone.gemfire.cache.query.cq.dunit.CqQueryTestListener;
+import com.gemstone.gemfire.cache.query.internal.cq.ClientCQImpl;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.internal.AvailablePort;
+import com.gemstone.gemfire.internal.cache.GemFireCacheImpl;
+import com.gemstone.gemfire.internal.logging.InternalLogWriter;
+
+import dunit.Host;
+import dunit.SerializableRunnable;
+
+/**
+ * @author ashetkar
+ *
+ */
+public class MultiuserDurableCQAuthzDUnitTest extends
+ ClientAuthorizationTestBase {
+
+ public static final Map<String, String> cqNameToQueryStrings = new HashMap<String, String>();
+
+ static {
+ cqNameToQueryStrings.put("CQ_0", "SELECT * FROM ");
+ cqNameToQueryStrings.put("CQ_1", "SELECT * FROM ");
+ }
+
+ public MultiuserDurableCQAuthzDUnitTest(String name) {
+ super(name);
+ }
+
+ public void setUp() throws Exception {
+ super.setUp();
+ getSystem();
+ invokeInEveryVM(new SerializableRunnable("getSystem") {
+ public void run() {
+ getSystem();
+ }
+ });
+
+ final Host host = Host.getHost(0);
+ server1 = host.getVM(0);
+ server2 = host.getVM(1);
+ client1 = host.getVM(2);
+ client2 = host.getVM(3);
+
+ server1.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ server2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { serverExpectedExceptions });
+ client2.invoke(SecurityTestUtil.class, "registerExpectedExceptions",
+ new Object[] { clientExpectedExceptions });
+ SecurityTestUtil.registerExpectedExceptions(clientExpectedExceptions);
+ }
+
+ public void testCQForDurableClientsWithDefaultClose() throws Exception {
+ /*
+ * 1. Start a server.
+ * 2. Start a durable client in mulituser secure mode.
+ * 3. Create two users registering unique durable CQs on server.
+ * 4. Invoke GemFireCache.close() at client.
+ * 5. Put some events on server satisfying both the CQs.
+ * 6. Up the client and the two users.
+ * 7. Confirm that the users receive the events which were enqueued at server while they were away.
+ * 8. Same for ProxyCache.close()
+ */
+ Integer numOfUsers = 2;
+ Integer numOfPuts = 5;
+ Boolean[] postAuthzAllowed = new Boolean[] {Boolean.TRUE, Boolean.TRUE};
+
+ doTest(numOfUsers, numOfPuts, postAuthzAllowed,
+ getXmlAuthzGenerator(), null);
+ }
+
+ public void testCQForDurableClientsWithCloseKeepAliveTrue() throws Exception {
+ /*
+ * 1. Start a server.
+ * 2. Start a durable client in mulituser secure mode.
+ * 3. Create two users registering unique durable CQs on server.
+ * 4. Invoke GemFireCache.close(false) at client.
+ * 5. Put some events on server satisfying both the CQs.
+ * 6. Up the client and the two users.
+ * 7. Observer the behaviour.
+ * 8. Same for ProxyCache.close(false)
+ */
+ Integer numOfUsers = 2;
+ Integer numOfPuts = 5;
+ Boolean[] postAuthzAllowed = new Boolean[] {Boolean.TRUE, Boolean.TRUE};
+
+ doTest(numOfUsers, numOfPuts, postAuthzAllowed,
+ getXmlAuthzGenerator(), Boolean.TRUE);
+ }
+
+ public void testCQForDurableClientsWithCloseKeepAliveFalse() throws Exception {
+ /*
+ * 1. Start a server.
+ * 2. Start a durable client in mulituser secure mode.
+ * 3. Create two users registering unique durable CQs on server.
+ * 4. Invoke GemFireCache.close(true) at client.
+ * 5. Put some events on server satisfying both the CQs.
+ * 6. Up the client and the two users.
+ * 7. Observer the behaviour.
+ * 8. Same for ProxyCache.close(true)
+ */
+ Integer numOfUsers = 2;
+ Integer numOfPuts = 5;
+ Boolean[] postAuthzAllowed = new Boolean[] {Boolean.TRUE, Boolean.TRUE};
+
+ doTest(numOfUsers, numOfPuts, postAuthzAllowed,
+ getXmlAuthzGenerator(), Boolean.FALSE);
+ }
+
+ private void doTest(Integer numOfUsers, Integer numOfPuts,
+ Boolean[] postAuthzAllowed, AuthzCredentialGenerator gen, Boolean keepAlive)
+ throws Exception {
+ CredentialGenerator cGen = gen.getCredentialGenerator();
+ Properties extraAuthProps = cGen.getSystemProperties();
+ Properties javaProps = cGen.getJavaProperties();
+ Properties extraAuthzProps = gen.getSystemProperties();
+ String authenticator = cGen.getAuthenticator();
+ String accessor = gen.getAuthorizationCallback();
+ String authInit = cGen.getAuthInit();
+ TestAuthzCredentialGenerator tgen = new TestAuthzCredentialGenerator(gen);
+
+ Properties serverProps = buildProperties(authenticator, accessor, true,
+ extraAuthProps, extraAuthzProps);
+
+ Properties opCredentials;
+ cGen = tgen.getCredentialGenerator();
+ Properties javaProps2 = null;
+ if (cGen != null) {
+ javaProps2 = cGen.getJavaProperties();
+ }
+
+ int[] indices = new int[numOfPuts];
+ for (int index = 0; index < numOfPuts; ++index) {
+ indices[index] = index;
+ }
+
+ Random rnd = new Random();
+ Properties[] authProps = new Properties[numOfUsers];
+ String durableClientId = "multiuser_durable_client_1";
+ Properties client2Credentials = null;
+ for (int i = 0; i < numOfUsers; i++) {
+ int rand = rnd.nextInt(100) + 1;
+ if (postAuthzAllowed[i]) {
+ opCredentials = tgen.getAllowedCredentials(new OperationCode[] {
+ OperationCode.EXECUTE_CQ, OperationCode.GET}, // For callback, GET should be allowed
+ new String[] {regionName}, indices, rand);
+ } else {
+ opCredentials = tgen.getDisallowedCredentials(new OperationCode[] {
+ OperationCode.GET}, // For callback, GET should be disallowed
+ new String[] {regionName}, indices, rand);
+ }
+ authProps[i] = SecurityTestUtil.concatProperties(new Properties[] {
+ opCredentials, extraAuthProps, extraAuthzProps});
+
+ if (client2Credentials == null) {
+ client2Credentials = tgen.getAllowedCredentials(new OperationCode[] {
+ OperationCode.PUT},
+ new String[] {regionName}, indices, rand);
+ }
+ }
+
+ // Get ports for the servers
+ Integer port1 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer port2 = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ Integer locatorPort = new Integer(AvailablePort
+ .getRandomAvailablePort(AvailablePort.SOCKET));
+ // Close down any running servers
+ server1.invoke(SecurityTestUtil.class, "closeCache");
+ server2.invoke(SecurityTestUtil.class, "closeCache");
+
+ server1.invoke(MultiuserDurableCQAuthzDUnitTest.class,
+ "createServerCache", new Object[] {serverProps, javaProps, locatorPort, port1});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class,
+ "createClientCache", new Object[] {javaProps2, authInit, authProps,
+ new Integer[] {port1, port2}, numOfUsers, durableClientId, postAuthzAllowed});
+
+// client2.invoke(SecurityTestUtil.class, "createCacheClient",
+// new Object[] {authInit, client2Credentials, javaProps2,
+// new Integer[] {port1, port2}, null, SecurityTestUtil.NO_EXCEPTION});
+
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "createCQ",
+ new Object[] {numOfUsers, Boolean.TRUE});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "executeCQ",
+ new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+ new String[numOfUsers]});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "readyForEvents");
+
+ if (keepAlive == null) {
+ client1.invoke(SecurityTestUtil.class, "closeCache");
+ } else {
+ client1.invoke(SecurityTestUtil.class, "closeCache",
+ new Object[] {keepAlive});
+ }
+
+ server1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "doPuts",
+ new Object[] {numOfPuts, Boolean.TRUE/* put last key */});
+
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class,
+ "createClientCache", new Object[] {javaProps2, authInit, authProps,
+ new Integer[] {port1, port2}, numOfUsers, durableClientId, postAuthzAllowed});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "createCQ",
+ new Object[] {numOfUsers, Boolean.TRUE});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "executeCQ",
+ new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+ new String[numOfUsers]});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "readyForEvents");
+
+ if (!postAuthzAllowed[0] || keepAlive == null || !keepAlive) {
+ // Don't wait as no user is authorized to receive cq events.
+ Thread.sleep(1000);
+ } else {
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "waitForLastKey",
+ new Object[] {Integer.valueOf(0), Boolean.TRUE});
+ }
+ Integer numOfCreates = (keepAlive == null) ? 0
+ : (keepAlive) ? (numOfPuts + 1/* last key */) : 0;
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "checkCQListeners",
+ new Object[] {numOfUsers, postAuthzAllowed, numOfCreates, 0});
+
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "proxyCacheClose",
+ new Object[] {new Integer[] {0, 1}, keepAlive});
+
+ client1.invoke(SecurityTestUtil.class, "createProxyCache",
+ new Object[] {new Integer[] {0, 1}, authProps});
+
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "createCQ",
+ new Object[] {numOfUsers, Boolean.TRUE});
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "executeCQ",
+ new Object[] {numOfUsers, new Boolean[] {false, false}, numOfPuts,
+ new String[numOfUsers]});
+
+ server1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "doPuts",
+ new Object[] {numOfPuts, Boolean.TRUE/* put last key */});
+
+ if (!postAuthzAllowed[0] || keepAlive == null || !keepAlive) {
+ // Don't wait as no user is authorized to receive cq events.
+ Thread.sleep(1000);
+ } else {
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "waitForLastKey",
+ new Object[] {Integer.valueOf(0), Boolean.FALSE});
+ }
+ Integer numOfUpdates = numOfPuts + 1;
+ client1.invoke(MultiuserDurableCQAuthzDUnitTest.class, "checkCQListeners",
+ new Object[] {numOfUsers, postAuthzAllowed, 0, numOfUpdates});
+ }
+
+ public static void createServerCache(Properties serverProps,
+ Properties javaProps, Integer locatorPort, Integer serverPort) {
+ SecurityTestUtil.createCacheServer((Properties)serverProps, javaProps,
+ locatorPort, null, serverPort, Boolean.TRUE, new Integer(
+ SecurityTestUtil.NO_EXCEPTION));
+ }
+
+ public static void createClientCache(Properties javaProps,
+ String authInit, Properties[] authProps, Integer ports[],
+ Integer numOfUsers, Boolean[] postAuthzAllowed) {
+ SecurityTestUtil.createCacheClientForMultiUserMode(numOfUsers, authInit,
+ authProps, javaProps, ports, null, Boolean.FALSE,
+ SecurityTestUtil.NO_EXCEPTION);
+ }
+
+ public static void readyForEvents() {
+ GemFireCacheImpl.getInstance().readyForEvents();
+ }
+
+ public static void createClientCache(Properties javaProps,
+ String authInit, Properties[] authProps, Integer ports[],
+ Integer numOfUsers, String durableId, Boolean[] postAuthzAllowed) {
+ SecurityTestUtil.createCacheClientForMultiUserMode(numOfUsers, authInit,
+ authProps, javaProps, ports, null, Boolean.FALSE, durableId,
+ SecurityTestUtil.NO_EXCEPTION);
+ }
+
+ public static void createCQ(Integer num) {
+ createCQ(num, false);
+ }
+
+ public static void createCQ(Integer num, Boolean isDurable) {
+ for (int i = 0; i < num; i++) {
+ QueryService cqService = SecurityTestUtil.proxyCaches[i].getQueryService();
+ String cqName = "CQ_" + i;
+ String queryStr = cqNameToQueryStrings.get(cqName)
+ + SecurityTestUtil.proxyCaches[i].getRegion(regionName).getFullPath();
+ // Create CQ Attributes.
+ CqAttributesFactory cqf = new CqAttributesFactory();
+ CqListener[] cqListeners = {new CqQueryTestListener(getLogWriter())};
+ ((CqQueryTestListener)cqListeners[0]).cqName = cqName;
+
+ cqf.initCqListeners(cqListeners);
+ CqAttributes cqa = cqf.create();
+
+ // Create CQ.
+ try {
+ CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa, isDurable);
+ assertTrue("newCq() state mismatch", cq1.getState().isStopped());
+ } catch (Exception ex) {
+ AssertionError err = new AssertionError("Failed to create CQ " + cqName
+ + " . ");
+ err.initCause(ex);
+ getLogWriter().info("CqService is :" + cqService, err);
+ throw err;
+ }
+ }
+ }
+
+ public static void executeCQ(Integer num, Boolean[] initialResults,
+ Integer expectedResultsSize, String[] expectedErr) {
+ InternalLogWriter logWriter = InternalDistributedSystem.getStaticInternalLogWriter();
+ for (int i = 0; i < num; i++) {
+ try {
+ if (expectedErr[i] != null) {
+ logWriter.info(
+ "<ExpectedException action=add>" + expectedErr[i]
+ + "</ExpectedException>");
+ }
+ CqQuery cq1 = null;
+ String cqName = "CQ_" + i;
+ String queryStr = cqNameToQueryStrings.get(cqName)
+ + SecurityTestUtil.proxyCaches[i].getRegion(regionName)
+ .getFullPath();
+ QueryService cqService = SecurityTestUtil.proxyCaches[i]
+ .getQueryService();
+
+ // Get CqQuery object.
+ try {
+ cq1 = cqService.getCq(cqName);
+ if (cq1 == null) {
+ getLogWriter().info(
+ "Failed to get CqQuery object for CQ name: " + cqName);
+ fail("Failed to get CQ " + cqName);
+ } else {
+ getLogWriter().info("Obtained CQ, CQ name: " + cq1.getName());
+ assertTrue("newCq() state mismatch", cq1.getState().isStopped());
+ }
+ } catch (Exception ex) {
+ getLogWriter().info("CqService is :" + cqService);
+ getLogWriter().error(ex);
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ex);
+ throw err;
+ }
+
+ if (initialResults[i]) {
+ SelectResults cqResults = null;
+
+ try {
+ cqResults = cq1.executeWithInitialResults();
+ } catch (Exception ex) {
+ getLogWriter().info("CqService is: " + cqService);
+ ex.printStackTrace();
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ex);
+ throw err;
+ }
+ getLogWriter().info("initial result size = " + cqResults.size());
+ assertTrue("executeWithInitialResults() state mismatch", cq1
+ .getState().isRunning());
+ if (expectedResultsSize >= 0) {
+ assertEquals("unexpected results size", expectedResultsSize
+ .intValue(), cqResults.size());
+ }
+ } else {
+ try {
+ cq1.execute();
+ } catch (Exception ex) {
+ AssertionError err = new AssertionError("Failed to execute CQ "
+ + cqName);
+ err.initCause(ex);
+ if (expectedErr == null) {
+ getLogWriter().info("CqService is: " + cqService, err);
+ }
+ throw err;
+ }
+ assertTrue("execute() state mismatch", cq1.getState().isRunning());
+ }
+ } finally {
+ if (expectedErr[i] != null) {
+ logWriter.info(
+ "<ExpectedException action=remove>" + expectedErr[i]
+ + "</ExpectedException>");
+ }
+ }
+ }
+ }
+
+ public static void doPuts(Integer num, Boolean putLastKey) {
+ Region region = GemFireCacheImpl.getInstance().getRegion(regionName);
+ for (int i = 0; i < num; i++) {
+ region.put("CQ_key"+i, "CQ_value"+i);
+ }
+ if (putLastKey) {
+ region.put("LAST_KEY", "LAST_KEY");
+ }
+ }
+
+ public static void putLastKey() {
+ Region region = GemFireCacheImpl.getInstance().getRegion(regionName);
+ region.put("LAST_KEY", "LAST_KEY");
+ }
+
+ public static void waitForLastKey(Integer cqIndex, Boolean isCreate) {
+ String cqName = "CQ_" + cqIndex;
+ QueryService qService = SecurityTestUtil.proxyCaches[cqIndex].getQueryService();
+ ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+ if (isCreate) {
+ ((CqQueryTestListener)cqQuery.getCqListeners()[cqIndex])
+ .waitForCreated("LAST_KEY");
+ } else {
+ ((CqQueryTestListener)cqQuery.getCqListeners()[cqIndex])
+ .waitForUpdated("LAST_KEY");
+ }
+ }
+
+ public static void checkCQListeners(Integer numOfUsers,
+ Boolean[] expectedListenerInvocation, Integer createEventsSize,
+ Integer updateEventsSize) {
+ for (int i = 0; i < numOfUsers; i++) {
+ String cqName = "CQ_" + i;
+ QueryService qService = SecurityTestUtil.proxyCaches[i].getQueryService();
+ ClientCQImpl cqQuery = (ClientCQImpl)qService.getCq(cqName);
+ if (expectedListenerInvocation[i]) {
+ for (CqListener listener : cqQuery.getCqListeners()) {
+ assertEquals(createEventsSize.intValue(),
+ ((CqQueryTestListener)listener).getCreateEventCount());
+ assertEquals(updateEventsSize.intValue(),
+ ((CqQueryTestListener)listener).getUpdateEventCount());
+ }
+ } else {
+ for (CqListener listener : cqQuery.getCqListeners()) {
+ assertEquals(0, ((CqQueryTestListener)listener).getTotalEventCount());
+ }
+ }
+ }
+ }
+
+ public static void proxyCacheClose(Integer[] userIndices) {
+ proxyCacheClose(userIndices, null);
+ }
+
+ public static void proxyCacheClose(Integer[] userIndices, Boolean keepAliveFlags) {
+ if (keepAliveFlags != null) {
+ for (int i : userIndices) {
+ SecurityTestUtil.proxyCaches[i].close(keepAliveFlags);
+ }
+ } else {
+ for (int i : userIndices) {
+ SecurityTestUtil.proxyCaches[i].close();
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/.keepme
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/.keepme b/gemfire-cq/src/test/resources/.keepme
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt
new file mode 100644
index 0000000..c55df1d
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/excludedClasses.txt
@@ -0,0 +1,2 @@
+# e.g.,
+#com/gemstone/gemfire/cache/query/CqAttributesFactory
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt
new file mode 100644
index 0000000..6c40e24
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/openBugs.txt
@@ -0,0 +1,21 @@
+# This is a list of classes excluded due to open bugs about their having
+# incompatible changes. There should be no entries in this file at the
+# time of a product release.
+
+# Each entry should be a bug number followed by a comma and the
+# full class name. The package components can be delimited with a period
+# or a comma. Don't include ".class" or ".java" at the end of the name.
+
+# example: 50174,com/gemstone/gemfire/distributed/internal/StartupResponseWithVersionMessage
+# example: 50175,com.gemstone.org.jgroups.Message$Header
+
+
+# ~~~~~~~~~~~~~~~~~~~ DataSerializables ~~~~~~~~~~~~~~~~~~~~~~~~~
+# these are failures from testDataSerializables
+
+
+# ~~~~~~~~~~~~~~~~~~~ Serializables ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+# these are failures from testSerializables
+
+
+
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
new file mode 100644
index 0000000..eb9d8d6
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedDataSerializables.txt
@@ -0,0 +1,4 @@
+com/gemstone/gemfire/cache/query/internal/cq/ServerCQImpl,2
+fromData,64,2ab40046594dc22ab400462bb80099b6009ab6007f2cc3a700084e2cc32dbf2a2bb8009bb6009cb5009d2a2bb8009eb5005d2a2bb9009f0100b800a0b50008b1
+toData,47,2ab40046b60047b800a12bb800a22ab4009db8007c2bb800a32ab4005d2bb800a42b2ab40008b600a5b900a60300b1
+
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt
new file mode 100755
index 0000000..ddd9718
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/codeAnalysis/sanctionedSerializables.txt
@@ -0,0 +1 @@
+com/gemstone/gemfire/cache/query/internal/cq/CqConflatable,true,-7215022132135862557,conflate:boolean,id:com/gemstone/gemfire/internal/cache/EventID,key:java/lang/Object,regionname:java/lang/String,value:java/lang/Object
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml
new file mode 100644
index 0000000..afde78f
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-client-cache.xml
@@ -0,0 +1,21 @@
+<?xml version="1.0"?>
+ <!DOCTYPE client-cache PUBLIC
+ "-//GemStone Systems, Inc.//GemFire Declarative Caching 6.6//EN"
+ "http://www.gemstone.com/dtd/cache6_6.dtd">
+
+
+<client-cache>
+ <pool
+ name="client"
+ subscription-enabled="true"
+ load-conditioning-interval="60000"
+ read-timeout="30000" retry-attempts="5" >
+ <server host="localhost" port="10188" />
+ </pool>
+
+ <region name="testReadyForEventsNotCalledImplicitlyWithCacheXML_region" >
+ <region-attributes id="testReadyForEventsNotCalledImplicitlyWithCacheXML_region" statistics-enabled="true" pool-name="client" refid="PROXY">
+ <cache-listener><class-name>com.gemstone.gemfire.internal.cache.tier.sockets.CacheServerTestUtil$ControlListener</class-name></cache-listener>
+ </region-attributes>
+ </region>
+</client-cache>
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml
new file mode 100644
index 0000000..2d58506
--- /dev/null
+++ b/gemfire-cq/src/test/resources/com/gemstone/gemfire/internal/cache/tier/sockets/durablecq-server-cache.xml
@@ -0,0 +1,16 @@
+<?xml version="1.0"?>
+<!DOCTYPE cache PUBLIC
+ "-//GemStone Systems, Inc.//GemFire Declarative Caching 6.6//EN"
+ "http://www.gemstone.com/dtd/cache6_6.dtd">
+
+<cache is-server="true">
+ <!-- Define this cache server -->
+ <cache-server port="10188" >
+ <client-subscription eviction-policy="entry" capacity="1000"/>
+ </cache-server>
+ <region name="testReadyForEventsNotCalledImplicitlyWithCacheXML_region">
+ <region-attributes data-policy="partition" enable-subscription-conflation="false">
+ <partition-attributes redundant-copies="0" total-num-buckets="113"/>
+ </region-attributes>
+ </region>
+</cache>
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt b/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt
new file mode 100644
index 0000000..ea6685c
--- /dev/null
+++ b/gemfire-cq/src/test/resources/cq/statistics/cqStats.bt
@@ -0,0 +1,3 @@
+cq/statistics/cqStatsHCT.conf
+edgeHosts =1 edgeVMsPerHost = 1 edgeThreadsPerVM = 1
+bridgeHosts =1 bridgeVMsPerHost = 1 bridgeThreadsPerVM = 1
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf b/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf
new file mode 100644
index 0000000..4623a5d
--- /dev/null
+++ b/gemfire-cq/src/test/resources/cq/statistics/cqStatsHCT.conf
@@ -0,0 +1,100 @@
+hydra.Prms-testRequirement = "Test CQ with a variety of operations with careful validation and serial execution in a bridge configuration";
+hydra.Prms-testDescription = "
+This test executes operations on entries on a region and carefully
+validates for correctness in the CQs.
+The test uses serial round robin; the first thread
+in the round does a random operation, then all other threads in the round
+verify their view of the operation. After the last thread in the round verifies,
+it then becomes the first in the next round, thus the thread doing the random
+operation changes for each round.
+";
+
+INCLUDE $JTESTS/hydraconfig/hydraparams1.inc;
+INCLUDE $JTESTS/hydraconfig/topology_hct.inc;
+
+hydra.VmPrms-extraClassPaths =
+ fcn "hydra.TestConfigFcns.duplicate(\"$GEMFIRE/lib/antlr.jar\", ${bridgeHosts})" ncf;
+
+THREADGROUP bridgeThreads
+ totalThreads = fcn
+ ${bridgeHosts} * ${bridgeVMsPerHost} * ${bridgeThreadsPerVM}
+ ncf
+ clientNames = fcn "hydra.TestConfigFcns.generateNames
+ (\"bridge\", ${bridgeHosts}, true)"
+ ncf;
+THREADGROUP edgeThreads
+ totalThreads = fcn
+ ${edgeHosts} * ${edgeVMsPerHost} * ${edgeThreadsPerVM}
+ ncf
+ clientNames = fcn "hydra.TestConfigFcns.generateNames
+ (\"edge\", ${edgeHosts}, true)"
+ ncf;
+
+INITTASK taskClass = cq.statistics.CQStatisticsTest taskMethod = HydraTask_initializeBridgeServer
+ threadGroups = bridgeThreads
+ runMode = always;
+
+INITTASK taskClass = cq.statistics.CQStatisticsTest taskMethod = HydraTask_initializeClient
+ threadGroups = edgeThreads
+ runMode = always;
+
+INITTASK taskClass = cq.statistics.CQStatisticsTest taskMethod = HydraTask_verifyCqStatistics
+ threadGroups = edgeThreads;
+
+hydra.GemFirePrms-stopSystemsAfterTest = true;
+
+hydra.Prms-totalTaskTimeSec = 1200;
+hydra.Prms-maxResultWaitSec = 600;
+hydra.Prms-serialExecution = true;
+hydra.Prms-roundRobin = true;
+
+INCLUDE $JTESTS/util/randomValues.inc;
+util.RandomValuesPrms-objectType = byte[];
+util.RandomValuesPrms-elementSize = 1000;
+util.RandomValuesPrms-borderCasePercentage = 0;
+util.ValueHolderPrms-useExtraObject = true;
+hydra.GemFirePrms-conserveSockets = ONEOF true false FOENO;
+
+hydra.CachePrms-names = cache1;
+hydra.CachePrms-searchTimeout = 600;
+
+hydra.Prms-useFixedRandomInMaster= true;
+hydra.RegionPrms-names = clientRegion serverRegion;
+hydra.RegionPrms-regionName = testRegion testRegion;
+hydra.RegionPrms-scope = local ack;
+hydra.RegionPrms-poolName = edgeDescript none;
+hydra.RegionPrms-dataPolicy = normal replicate;
+hydra.RegionPrms-cacheListeners = util.SilenceListener, hct.BridgeEventListener;
+hydra.RegionPrms-partitionName = none none;
+
+cq.CQUtilPrms-numOpsPerTask = 100;
+
+util.CachePrms-useDeclarativeXmlFile = ONEOF true false FOENO;
+util.TestHelperPrms-minTaskGranularitySec = 30;
+
+cq.CQUtilPrms-serverEntryOperations = ONEOF add add update invalidate get destroy FOENO;
+cq.CQUtilPrms-clientEntryOperations = ONEOF add add update get destroy FOENO;
+cq.CQUtilPrms-upperThreshold = 500;
+cq.CQUtilPrms-upperThresholdServerOperations = ONEOF destroy FOENO;
+cq.CQUtilPrms-upperThresholdClientOperations = ONEOF destroy FOENO;
+cq.CQUtilPrms-lowerThreshold = 0;
+cq.CQUtilPrms-lowerThresholdServerOperations = add;
+cq.CQUtilPrms-lowerThresholdClientOperations = add;
+cq.CQUtilPrms-queryDepth = 7;
+cq.CQUtilPrms-numQueriesPerClientVM = 100;
+cq.CQUtilPrms-QueryServicePoolName = qservice;
+cq.CQUtilPrms-QueryServiceUsingPool = ONEOF true false FOENO;
+
+// define the edge clients
+hydra.PoolPrms-names = edgeDescript qservice;
+hydra.PoolPrms-minConnections = 2;
+hydra.PoolPrms-subscriptionEnabled = true;
+hydra.PoolPrms-threadLocalConnections = true;
+hydra.PoolPrms-readTimeout = 800000;
+hydra.PoolPrms-subscriptionRedundancy = RANGE 0
+ fcn ${bridgeHosts} * ${bridgeVMsPerHost} ncf
+ EGNAR;
+
+// define the bridge servers
+hydra.BridgePrms-names = bridge;
+
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/build.gradle
----------------------------------------------------------------------
diff --git a/gemfire-wan/build.gradle b/gemfire-wan/build.gradle
new file mode 100644
index 0000000..eba2539
--- /dev/null
+++ b/gemfire-wan/build.gradle
@@ -0,0 +1,5 @@
+dependencies {
+ provided project(':gemfire-core')
+ provided project(path: ':gemfire-core', configuration: 'testOutput')
+ provided project(path: ':gemfire-junit', configuration: 'testOutput')
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/main/java/.keepme
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/.keepme b/gemfire-wan/src/main/java/.keepme
new file mode 100644
index 0000000..e69de29
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
new file mode 100644
index 0000000..15be615
--- /dev/null
+++ b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/GatewaySenderBatchOp.java
@@ -0,0 +1,304 @@
+/*=========================================================================
+ * Copyright (c) 2002-2014, Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.cache.client.internal;
+
+import com.gemstone.gemfire.InternalGemFireError;
+import com.gemstone.gemfire.cache.client.ServerOperationException;
+import com.gemstone.gemfire.internal.Version;
+import com.gemstone.gemfire.internal.cache.EventID;
+import com.gemstone.gemfire.internal.cache.tier.MessageType;
+import com.gemstone.gemfire.internal.cache.tier.sockets.ChunkedMessage;
+import com.gemstone.gemfire.internal.cache.tier.sockets.Message;
+import com.gemstone.gemfire.internal.cache.tier.sockets.Part;
+import com.gemstone.gemfire.internal.cache.wan.BatchException70;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventImpl;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventRemoteDispatcher;
+import com.gemstone.gemfire.internal.cache.wan.GatewaySenderEventRemoteDispatcher.GatewayAck;
+import com.gemstone.gemfire.internal.i18n.LocalizedStrings;
+import com.gemstone.gemfire.internal.logging.LogService;
+
+import java.net.SocketTimeoutException;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.logging.log4j.Logger;
+
+public class GatewaySenderBatchOp {
+
+ private static final Logger logger = LogService.getLogger();
+
+ /**
+ * Send a list of gateway events to a server to execute
+ * using connections from the given pool
+ * to communicate with the server.
+ * @param con the connection to send the message on.
+ * @param pool the pool to use to communicate with the server.
+ * @param events list of gateway events
+ * @param batchId the ID of this batch
+ * @param removeFromQueueOnException true if the events should be processed even after some exception
+ */
+ public static void executeOn(Connection con, ExecutablePool pool, List events, int batchId, boolean removeFromQueueOnException, boolean isRetry)
+ {
+ AbstractOp op = null;
+ //System.out.println("Version: "+con.getWanSiteVersion());
+ if (Version.GFE_651.compareTo(con.getWanSiteVersion()) >= 0) {
+ op = new GatewaySenderGFEBatchOpImpl(events, batchId, removeFromQueueOnException, con.getDistributedSystemId(), isRetry);
+ } else {
+ // Default should create a batch of server version (ACCEPTOR.VERSION)
+ op = new GatewaySenderGFEBatchOpImpl(events, batchId, removeFromQueueOnException, con.getDistributedSystemId(), isRetry);
+ }
+ pool.executeOn(con, op, true/*timeoutFatal*/);
+ }
+
+
+ public static Object executeOn(Connection con, ExecutablePool pool)
+ {
+ AbstractOp op = null;
+ //System.out.println("Version: "+con.getWanSiteVersion());
+ // [sumedh] both cases are now same; why switch-case?
+ if (Version.GFE_651.compareTo(con.getWanSiteVersion()) >= 0) {
+ op = new GatewaySenderGFEBatchOpImpl();
+ } else {
+ // Default should create a batch of server version (ACCEPTOR.VERSION)
+ op = new GatewaySenderGFEBatchOpImpl();
+ }
+ return pool.executeOn(con, op, true/*timeoutFatal*/);
+ }
+
+ private GatewaySenderBatchOp() {
+ // no instances allowed
+ }
+
+ static class GatewaySenderGFEBatchOpImpl extends AbstractOp {
+
+ /**
+ * @throws com.gemstone.gemfire.SerializationException if serialization fails
+ */
+ public GatewaySenderGFEBatchOpImpl(List events, int batchId, boolean removeFromQueueOnException, int dsId, boolean isRetry) {
+ super(MessageType.GATEWAY_RECEIVER_COMMAND, calcPartCount(events));
+ if (isRetry) {
+ getMessage().setIsRetry();
+ }
+ getMessage().addIntPart(events.size());
+ getMessage().addIntPart(batchId);
+ getMessage().addIntPart(dsId);
+ getMessage().addBytesPart(
+ new byte[] { removeFromQueueOnException ? (byte)1 : (byte)0 });
+ // Add each event
+ for (Iterator i = events.iterator(); i.hasNext();) {
+ GatewaySenderEventImpl event = (GatewaySenderEventImpl)i.next();
+ // Add action
+ int action = event.getAction();
+ getMessage().addIntPart(action);
+ { // Add posDup flag
+ byte posDupByte = (byte)(event.getPossibleDuplicate()?0x01:0x00);
+ getMessage().addBytesPart(new byte[] {posDupByte});
+ }
+ if (action >= 0 && action <= 3) {
+ // 0 = create
+ // 1 = update
+ // 2 = destroy
+ String regionName = event.getRegionPath();
+ EventID eventId = event.getEventId();
+ Object key = event.getKey();
+ Object callbackArg = event.getSenderCallbackArgument();
+
+ // Add region name
+ getMessage().addStringPart(regionName);
+ // Add event id
+ getMessage().addObjPart(eventId);
+ // Add key
+ getMessage().addStringOrObjPart(key);
+ if (action < 2 /* it is 0 or 1 */) {
+ byte[] value = event.getSerializedValue();
+ byte valueIsObject = event.getValueIsObject();;
+ // Add value (which is already a serialized byte[])
+ getMessage().addRawPart(value, (valueIsObject == 0x01));
+ }
+ // Add callback arg if necessary
+ if (callbackArg == null) {
+ getMessage().addBytesPart(new byte[] {0x00});
+ } else {
+ getMessage().addBytesPart(new byte[] {0x01});
+ getMessage().addObjPart(callbackArg);
+ }
+ getMessage().addLongPart(event.getVersionTimeStamp());
+ }
+ }
+ }
+
+ public GatewaySenderGFEBatchOpImpl() {
+ super(MessageType.GATEWAY_RECEIVER_COMMAND, 0);
+ }
+
+ @Override
+ public Object attempt(Connection cnx) throws Exception {
+ if (getMessage().getNumberOfParts() == 0) {
+ return attemptRead(cnx);
+ }
+ this.failed = true;
+ this.timedOut = false;
+ long start = startAttempt(cnx.getStats());
+ try {
+ try {
+ attemptSend(cnx);
+ this.failed = false;
+ } finally {
+ endSendAttempt(cnx.getStats(), start);
+ }
+ } finally {
+ endAttempt(cnx.getStats(), start);
+ }
+ return this.failed;
+ }
+
+ private Object attemptRead(Connection cnx) throws Exception {
+ this.failed = true;
+ try {
+ Object result = attemptReadResponse(cnx);
+ this.failed = false;
+ return result;
+ } catch (SocketTimeoutException ste) {
+ this.failed = false;
+ this.timedOut = true;
+ throw ste;
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+
+ /**
+ * Attempts to read a response to this operation by reading it from the
+ * given connection, and returning it.
+ * @param cnx the connection to read the response from
+ * @return the result of the operation
+ * or <code>null</code> if the operation has no result.
+ * @throws Exception if the execute failed
+ */
+ protected Object attemptReadResponse(Connection cnx) throws Exception {
+ Message msg = createResponseMessage();
+ if (msg != null) {
+ msg.setComms(cnx.getSocket(), cnx.getInputStream(),
+ cnx.getOutputStream(),
+ ((ConnectionImpl)cnx).getCommBufferForAsyncRead(), cnx.getStats());
+ if (msg instanceof ChunkedMessage) {
+ try {
+ return processResponse(msg, cnx);
+ } finally {
+ msg.unsetComms();
+ // TODO (ashetkar) Handle the case when we fail to read the
+ // connection id.
+ processSecureBytes(cnx, msg);
+ }
+ }
+
+ try {
+ msg.recv();
+ } finally {
+ msg.unsetComms();
+ processSecureBytes(cnx, msg);
+ }
+ return processResponse(msg, cnx);
+ }
+
+ return null;
+ }
+
+
+ private static int calcPartCount(List events) {
+ int numberOfParts = 4; // for the number of events and the batchId
+ for (Iterator i = events.iterator(); i.hasNext();) {
+ GatewaySenderEventImpl event = (GatewaySenderEventImpl)i.next();
+ numberOfParts += event.getNumberOfParts();
+ }
+ return numberOfParts;
+ }
+
+ @Override
+ protected void processSecureBytes(Connection cnx, Message message)
+ throws Exception {
+ }
+
+ @Override
+ protected boolean needsUserId() {
+ return false;
+ }
+
+ @Override
+ protected void sendMessage(Connection cnx) throws Exception {
+ getMessage().setEarlyAck((byte)(getMessage().getEarlyAckByte() & Message.MESSAGE_HAS_SECURE_PART));
+ getMessage().send(false);
+ }
+
+ @Override
+ protected Object processResponse(Message msg) throws Exception {
+ GatewayAck ack = null;
+ try {
+ // Read the header which describes the type of message following
+ switch (msg.getMessageType()) {
+ case MessageType.REPLY:
+ // Read the chunk
+ int batchId = msg.getPart(0).getInt();
+ int numEvents = msg.getPart(1).getInt();
+ ack = new GatewayAck(batchId, numEvents);
+ break;
+ case MessageType.EXCEPTION:
+ Part part0 = msg.getPart(0);
+
+ Object obj = part0.getObject();
+ if (obj instanceof List) {
+ List<BatchException70> l = (List<BatchException70>)part0.getObject();
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("We got an exception from the GatewayReceiver. MessageType : {} obj :{}", msg.getMessageType(), obj);
+ }
+ // don't throw Exception but set it in the Ack
+ BatchException70 be = new BatchException70(l);
+ ack = new GatewayAck(be, l.get(0).getBatchId());
+
+ } else if (obj instanceof Throwable) {
+ String s = ": While reading Ack from receiver "
+ + ((Throwable)obj).getMessage();
+ throw new ServerOperationException(s, (Throwable)obj);
+ }
+ break;
+ default:
+ throw new InternalGemFireError(
+ LocalizedStrings.Op_UNKNOWN_MESSAGE_TYPE_0
+ .toLocalizedString(Integer.valueOf(msg.getMessageType())));
+ }
+ } finally {
+ msg.clear();
+ }
+ return ack;
+ }
+
+ @Override
+ protected boolean isErrorResponse(int msgType) {
+ return false;
+ }
+ @Override
+ protected long startAttempt(ConnectionStats stats) {
+ return stats.startGatewayBatch();
+ }
+ @Override
+ protected void endSendAttempt(ConnectionStats stats, long start) {
+ stats.endGatewayBatchSend(start, hasFailed());
+ }
+ @Override
+ protected void endAttempt(ConnectionStats stats, long start) {
+ stats.endGatewayBatch(start, hasTimedOut(), hasFailed());
+ }
+
+ @Override
+ public boolean isGatewaySenderOp() {
+ return true;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java
new file mode 100644
index 0000000..d2030b3
--- /dev/null
+++ b/gemfire-wan/src/main/java/com/gemstone/gemfire/cache/client/internal/SenderProxy.java
@@ -0,0 +1,34 @@
+/*=========================================================================
+ * Copyright (c) 2002-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.cache.client.internal;
+
+import java.util.List;
+
+import com.gemstone.gemfire.cache.query.SelectResults;
+import com.gemstone.gemfire.distributed.internal.ServerLocation;
+
+/**
+ * Used to send operations from a sender to a receiver.
+ * @author skumar
+ * @since 8.1
+ */
+public class SenderProxy extends ServerProxy{
+ public SenderProxy(InternalPool pool) {
+ super(pool);
+ }
+
+ public void dispatchBatch_NewWAN(Connection con, List events, int batchId, boolean removeFromQueueOnException, boolean isRetry)
+ {
+ GatewaySenderBatchOp.executeOn(con, this.pool, events, batchId, removeFromQueueOnException, isRetry);
+ }
+
+ public Object receiveAckFromReceiver(Connection con)
+ {
+ return GatewaySenderBatchOp.executeOn(con, this.pool);
+ }
+}